#!/usr/bin/env python3 """ Abrechnung4: Summarize transactions by country """ import os import codecs import csv import logging import shutil from typing import Dict, List, Set from common import setup_logging, pad_row, get_data_root # pylint: disable=invalid-name currency_dict: Dict[str, Dict[str, float]] = dict() ean_dict: Dict[str, float] = dict() class Transaction(): # pylint: disable=too-many-instance-attributes """ Transaction representation for reportig purposes """ def __init__( # pylint: disable=too-many-arguments self, market_place: str, sales_channel: str, from_country: str, to_country: str, amt: float, tax_rate: float, period, tx_type: str, sku: str): self.MarketPlace = market_place self.SalesChannel = sales_channel self.DepartureCountry = from_country self.ArrivalCountry = to_country self._total = amt self.TaxRate = tax_rate self.TransactionPeriod = period self.TransactionType = tx_type self.ExchangeRate = lookup_exchange_rate(self.MarketPlace, self.TransactionPeriod) self.Sku = sku[:13] if sku and isinstance(sku, str) else None if tx_type == "FC_TRANSFER" and self.Sku: self.GrossAmount = ean_dict.get(self.Sku, 10.0) else: self.GrossAmount = (amt / self.ExchangeRate if self.ExchangeRate > 0 else 0.0) # if to_country == 'FR': # print(f"Trans Type {tx_type}, total {amt}, tax_rate {tax_rate}") def is_valid_exchange_rate(self): """ Test sanity of exchange rate. 0 implies missing value """ return self.ExchangeRate > 0.01 @property def CountryCombo(self): """ Create combination key from country codes """ return "{}-{}".format(self.DepartureCountry, self.ArrivalCountry) @property def NetAmount(self): """ Calculate the net amount """ return self.GrossAmount / (1 + self.TaxRate) @property def VatAmount(self): """ Calculate the VAT amount """ return self.GrossAmount - self.NetAmount class TransactionSummaryReport(): # pylint: disable=too-many-instance-attributes """ Class to aggregate and group the transactions for reporting """ def __init__(self): self.market_place_txs: Dict[str, List[Transaction]] = dict() self.sales_channel_txs: Dict[str, List[Transaction]] = dict() self.dest_country_txs: Dict[str, List[Transaction]] = dict() self.inter_country_txs: Dict[str, List[Transaction]] = dict() self.bad_exchange_rate_txs: List[Transaction] = [] self.grand_total: float = 0.0 self.sales_channels_totals: Dict[str, float] = dict() self.exhange_rates_used: Dict[List[Dict[str, float]]] = dict() self.market_places: Set[str] = set() self.sales_channels: Set[str] = set() self.sales_tax_rates: Dict[Set[float]] = dict() self.transaction_types: Set[str] = set() def add_transaction(self, tx: Transaction): # pylint: disable=too-many-branches """ Populate the groupings """ self.sales_channels.add(tx.SalesChannel) self.market_places.add(tx.MarketPlace) self.transaction_types.add(tx.TransactionType) if tx.ExchangeRate != 1.0: if tx.MarketPlace in self.exhange_rates_used: self.exhange_rates_used[tx.MarketPlace].add(tx.ExchangeRate) else: self.exhange_rates_used[tx.MarketPlace] = {tx.ExchangeRate} if tx.MarketPlace in self.market_place_txs: self.market_place_txs[tx.MarketPlace].append(tx) else: self.market_place_txs[tx.MarketPlace] = [tx] if tx.SalesChannel in self.sales_channel_txs: self.sales_channel_txs[tx.SalesChannel].append(tx) else: self.sales_channel_txs[tx.SalesChannel] = [tx] if tx.ArrivalCountry in self.dest_country_txs: self.dest_country_txs[tx.ArrivalCountry].append(tx) else: self.dest_country_txs[tx.ArrivalCountry] = [tx] if tx.CountryCombo in self.inter_country_txs: self.inter_country_txs[tx.CountryCombo].append(tx) else: self.inter_country_txs[tx.CountryCombo] = [tx] if tx.is_valid_exchange_rate(): self.grand_total += tx.GrossAmount else: self.bad_exchange_rate_txs.append(tx) if tx.SalesChannel in self.sales_channels_totals: self.sales_channels_totals[tx.SalesChannel] += tx.GrossAmount else: self.sales_channels_totals[tx.SalesChannel] = tx.GrossAmount if tx.TaxRate > 0: if tx.ArrivalCountry in self.sales_tax_rates: self.sales_tax_rates[tx.ArrivalCountry].add(tx.TaxRate) else: self.sales_tax_rates[tx.ArrivalCountry] = {tx.TaxRate} def get_market_places(self) -> List[str]: """ Return a sorted list of all market places """ return sorted(self.market_place_txs.keys()) def get_market_place_totals(self, tx_filter: List[str] = None ) -> Dict[str, float]: """ Calculate (filtered) transaction totals by market place """ tx_filter = tx_filter if tx_filter else list(self.transaction_types) market_place_totals: Dict[str, float] = dict() for mp in sorted(self.market_places): txs = self.market_place_txs.get(mp) market_place_totals[mp] = sum([ tx.GrossAmount if tx.TransactionType in tx_filter else 0 for tx in txs ]) if txs else 0.0 return market_place_totals def get_country_totals(self, market_place: str, tx_filter: List[str] = None) -> Dict[str, float]: """ Calculate transaction totals by country combination """ tx_filter = tx_filter if tx_filter else list(self.transaction_types) country_totals: Dict[str, float] = dict() txs = self.market_place_txs.get(market_place) if txs: for tx in txs: if not tx.TransactionType in tx_filter: continue if tx.CountryCombo in country_totals: country_totals[tx.CountryCombo] += tx.GrossAmount else: country_totals[tx.CountryCombo] = tx.GrossAmount return country_totals def get_sales_channels_totals( self, market_place: str, tx_filter: List[str] = None) -> Dict[str, float]: """ Calculate transaction totals by sales channel """ tx_filter = tx_filter if tx_filter else list(self.transaction_types) sales_channels_totals: Dict[str, float] = dict() txs = self.market_place_txs.get(market_place) if txs: for tx in txs: if not tx.TransactionType in tx_filter: continue if tx.SalesChannel in sales_channels_totals: sales_channels_totals[tx.SalesChannel] += tx.GrossAmount else: sales_channels_totals[tx.SalesChannel] = tx.GrossAmount return sales_channels_totals def get_sales_channels_grand_totals(self) -> Dict[str, Dict[str, float]]: """ Calculate grand total by sales channel """ sales_channels_grand_totals: Dict[str, Dict[str, float]] = dict() for mp in sorted(self.market_places): sales_channels_grand_totals[mp] = self.get_sales_channels_totals( mp) return sales_channels_grand_totals def get_dest_country_totals(self, tx_filter: List[str] = None ) -> Dict[str, Dict[str, float]]: """ Calculate total by arrival country """ tx_filter = tx_filter if tx_filter else list(self.transaction_types) country_totals: Dict[str, Dict[str, float]] = {} if self.dest_country_txs: for country, txs in self.dest_country_txs.items(): for tx in txs: if not tx.TransactionType in tx_filter: continue if country in country_totals: country_totals[country]["net"] += tx.NetAmount country_totals[country]["vat"] += tx.VatAmount country_totals[country]["gross"] += tx.GrossAmount else: country_totals[country] = { "net": tx.NetAmount, "vat": tx.VatAmount, "gross": tx.GrossAmount } return country_totals def get_sales_channels(self) -> List[str]: """ Return sorted list of sales channels """ return sorted(list(self.sales_channels)) def get_grand_total(self, tx_filter: List[str] = None) -> float: """ Calculate (filtered) grand total """ tx_filter = tx_filter if tx_filter else list(self.transaction_types) total = 0.0 mp_total = self.get_market_place_totals(tx_filter=tx_filter) for mp in self.market_places: total += mp_total.get(mp, 0.0) return total def parse_currency(value: str, ndigits: int = 2) -> float: """ Parse amount and round to n digits. Default = 2 """ try: total = float(value) except ValueError: total = 0.0 return round(total, ndigits) def format_value(value): """ Format amount in Euro fashion FIXME """ return str(round(value, 2)).replace(".", ",") def read_currency_ref(ref_path): """ Read exchange rate from ref file """ header_read = False full_ref_path = os.path.join(ref_path, "Ref_Kurse.csv") with codecs.open(full_ref_path, "rb", encoding="utf-8") as f: reader = csv.reader(f, delimiter=";") period_index = -1 market_place_index = -1 exchange_rate_index = -1 for row in reader: if header_read: market = row[market_place_index] period = row[period_index] exchange_rate = parse_currency( row[exchange_rate_index].replace(",", "."), 4) if market in currency_dict: currency_dict[market][period] = exchange_rate else: currency_dict[market] = {period: exchange_rate} else: period_index = row.index("ACTIVITY_PERIOD") market_place_index = row.index("MARKETPLACE") exchange_rate_index = row.index("KURS_UK") header_read = True def read_ean_lookup(ref_path) -> None: """ Load EAN price reference into global dict """ header_read = False full_ref_path = os.path.join(ref_path, "Ref_SSW_FBA.csv") with codecs.open(full_ref_path, "rb", encoding='utf-8') as f: reader = csv.reader(f, delimiter=';') ean_idx = -1 price_idx = -1 for row in reader: if header_read: ean = row[ean_idx].strip() price_euro = row[price_idx] price_conv = price_euro.replace(".", "").replace(",", ".") try: price = float(price_conv) ean_dict[ean] = price except ValueError as ve: logging.exception(ve) msg = "Failed row {0} with price {1}".format( ean, price_conv) logging.error(msg) else: ean_idx = row.index("VariationEAN") price_idx = row.index("PurchasePrice") header_read = True def lookup_exchange_rate(market_place: str, period: str): """ Perform a exchange rate lookup in ref data """ exchange_rate = 1.0 # Default currency is Euro if market_place in currency_dict: # Lookup the exchange rate for the period or set to 0.0 exchange_rate = currency_dict[market_place][ period] if period in currency_dict[market_place] else 0.0 return exchange_rate def write_summary_reports( # pylint: disable=too-many-locals, too-many-statements out_path: str, # pylint: disable=too-many-locals in_filename: str, rpt: TransactionSummaryReport): """ Write the report to CSV file """ out_file = f"{in_filename}.csv" out_file_path = os.path.join(out_path, out_file) with codecs.open(out_file_path, "w", encoding="utf-8") as of: writer = csv.writer(of, delimiter=";") sales_channels = sorted(rpt.sales_channels) sales_channel_hdr = [f"{x} Total" for x in sales_channels] sales_channel_blank = ["" for x in sales_channels] hdr = [ "Marketplace", "Depart/Arrival", "Country Subtotal", "Marketplace Total" ] + sales_channel_hdr + [ "Sales", "Refunds", "Returns", "Commingling Buy", "Commingling Sell", "FC Transfer", "Retourenquote" ] writer.writerow(hdr) num_cols = len(hdr) blank_row = [""] * num_cols mp_totals = rpt.get_market_place_totals() mp_totals_sales = rpt.get_market_place_totals(tx_filter=["SALE"]) mp_totals_returns = rpt.get_market_place_totals(tx_filter=["RETURN"]) mp_totals_refunds = rpt.get_market_place_totals(tx_filter=["REFUND"]) mp_totals_cbuys = rpt.get_market_place_totals( tx_filter=["COMMINGLING_BUY"]) mp_totals_csells = rpt.get_market_place_totals( tx_filter=["COMMINGLING_SELL"]) mp_totals_fctransfer = rpt.get_market_place_totals( tx_filter=["FC_TRANSFER"]) for mp in sorted(rpt.market_places): mp_sc_totals = rpt.get_sales_channels_totals(mp) # Write market place name and totals mp_tx_total = round(mp_totals.get(mp, 0.0), 2) mp_sales_total = round(mp_totals_sales.get(mp, 0.0), 2) mp_refund_total = round(mp_totals_refunds.get(mp, 0.0), 2) mp_return_total = round(mp_totals_returns.get(mp, 0.0), 2) mp_cbuy_total = round(mp_totals_cbuys.get(mp, 0.0), 2) mp_csell_total = round(mp_totals_csells.get(mp, 0.0), 2) mp_fctransfer_total = round(mp_totals_fctransfer.get(mp, 0.0), 2) mp_return_rate = round( (abs(mp_refund_total) / (mp_tx_total if mp_tx_total != 0.0 else 1.0)) * 100.0, 2) r = [mp, "", "", mp_tx_total] + [ round(mp_sc_totals.get(sc, 0.0), 2) for sc in sales_channels ] + [ mp_sales_total, mp_refund_total, mp_return_total, mp_cbuy_total, mp_csell_total, mp_fctransfer_total, mp_return_rate ] writer.writerow(pad_row(r, num_cols)) # Write departure/arrival country and shipment total c_tx_totals = rpt.get_country_totals(mp) c_sales_totals = rpt.get_country_totals(mp, tx_filter=["SALE"]) c_return_totals = rpt.get_country_totals(mp, tx_filter=["RETURN"]) c_refund_totals = rpt.get_country_totals(mp, tx_filter=["REFUND"]) c_cbuy_totals = rpt.get_country_totals( mp, tx_filter=["COMMINGLING_BUY"]) c_csell_totals = rpt.get_country_totals( mp, tx_filter=["COMMINGLING_SELL"]) c_fctransfer_totals = rpt.get_country_totals( mp, tx_filter=["FC_TRANSFER"]) for c in sorted(c_tx_totals.keys()): c_total = round(c_tx_totals.get(c, 0.0), 2) c_sales_subtotal = round(c_sales_totals.get(c, 0.0), 2) c_return_subtotal = round(c_return_totals.get(c, 0.0), 2) c_refund_subtotal = round(c_refund_totals.get(c, 0.0), 2) c_cbuy_subtotal = round(c_cbuy_totals.get(c, 0.0), 2) c_csell_subtotal = round(c_csell_totals.get(c, 0.0), 2) c_fctransfer_subtotal = round(c_fctransfer_totals.get(c, 0.0), 2) r = ["", c, c_total, ""] + sales_channel_blank + [ c_sales_subtotal, c_refund_subtotal, c_return_subtotal, c_cbuy_subtotal, c_csell_subtotal, c_fctransfer_subtotal ] writer.writerow(pad_row(r, num_cols)) writer.writerow(blank_row) # Write grand totals gt_tx_total = round(rpt.get_grand_total(), 2) gt_sales_total = round(rpt.get_grand_total(["SALE"]), 2) gt_refund_total = round(rpt.get_grand_total(["REFUND"]), 2) gt_return_total = round(rpt.get_grand_total(["RETURN"]), 2) gt_cbuy_total = round(rpt.get_grand_total(["COMMINGLING_BUY"]), 2) gt_csell_total = round(rpt.get_grand_total(["COMMINGLING_SELL"]), 2) gt_fctransfer_total = round(rpt.get_grand_total(["FC_TRANSFER"]), 2) gt_return_rate = round( (abs(gt_refund_total) / (gt_tx_total if gt_tx_total != 0.0 else 1.0)) * 100.0, 2) r = ["GRAND TOTAL", "", "", gt_tx_total] + [ round(rpt.sales_channels_totals.get(sc, 0.0), 2) for sc in sales_channels ] + [ gt_sales_total, gt_refund_total, gt_return_total, gt_cbuy_total, gt_csell_total, gt_fctransfer_total, gt_return_rate ] writer.writerow(pad_row(r, num_cols)) writer.writerow(blank_row) # Write % calculation based on grand total percentages = [0.0125, 0.029, 0.06, 0.07, 0.08, 0.09, 0.15] for p in percentages: perc_fmt = "{} %".format(format_value(round(p * 100.0, 2))) writer.writerow(["", "", perc_fmt, format_value(gt_tx_total * p)] + [""] * len(rpt.sales_channels)) r = [ "", "", "100.000,00/GRAND TOTAL", format_value(100000.0 / gt_tx_total) if gt_tx_total > 0 else "N/A" ] writer.writerow(pad_row(r, num_cols)) # List any currency conversions if rpt.exhange_rates_used: writer.writerow(blank_row) for mp, er in rpt.exhange_rates_used.items(): for r in er: r = ["Exchange Rates", "", mp, r] writer.writerow(pad_row(r, num_cols)) # Show any missing exchange rate periods if rpt.bad_exchange_rate_txs: writer.writerow(blank_row) for x in rpt.bad_exchange_rate_txs: r = [ "ERROR: Umtauschkurs fehlt fuer Period {} im Markplatz {}". format(x.TransactionPeriod, x.MarketPlace) ] writer.writerow(pad_row(r, num_cols)) writer.writerow(blank_row) # # Write out VAT totals by country country_totals = rpt.get_dest_country_totals() r = ["Country", "Net amount", "VAT amount", "VAT rate", "Gross amount"] writer.writerow(pad_row(r, num_cols)) for country in sorted(country_totals.keys()): v = country_totals[country] tax_rates = format_tax_rates( list(rpt.sales_tax_rates.get(country, []))) vat_total = round(v.get("vat", 0.0), 2) amt_total = round(v.get("gross", 0.0), 2) net_total = round(v.get("net", 0.0), 2) if vat_total > 0: r = [country, net_total, vat_total, tax_rates, amt_total] writer.writerow(pad_row(r, num_cols)) def format_tax_rates(tax_rates: List[float]) -> str: """ Format a list of tax rates and return as string """ rates = [] for tr in sorted(tax_rates): rate = round(tr * 100.0, 2) fmt = "{} %".format(rate) if rate > 0 else "" rates.append(fmt) return ",".join(rates) def summarize_transactions( # pylint: disable=too-many-locals, too-many-branches, too-many-statements in_file: str) -> TransactionSummaryReport: """ Read input file and add transaction to report object """ header_read = False marketplace_index = -1 depart_country_index = -1 arrival_country_index = -1 sales_channel_index = -1 period_index = -1 amount_index = -1 tx_id_index = -1 vat_perc_index = -1 tx_type_index = -1 sku_index = -1 line = 0 report = TransactionSummaryReport() with codecs.open(in_file, "r", encoding="utf-8") as f: reader = csv.reader(f, delimiter="\t", quotechar="|") for row in reader: if header_read: line += 1 tx_type = row[tx_type_index] # e.g. FC_Transfer # tx_id = row[tx_id_index] # e.g. 303-7809307-0302734 market_place = row[marketplace_index] # e.g. "amazon.es" period = row[period_index] # e.g. 2019-DEC ship_from_country = row[depart_country_index] if row[ depart_country_index] else "DE" # e.g. ES ship_to_country = row[arrival_country_index] if row[ arrival_country_index] else "DE" # e.g. ES sales_channel = row[sales_channel_index] amt = round(parse_currency(row[amount_index]), 2) tax_rate = parse_currency(row[vat_perc_index], 2) sku = row[sku_index] tx = Transaction(market_place, sales_channel, ship_from_country, ship_to_country, amt, tax_rate, period, tx_type, sku) report.add_transaction(tx) else: if "MARKETPLACE" in row: marketplace_index = row.index("MARKETPLACE") if "SALE_DEPART_COUNTRY" in row: depart_country_index = row.index("SALE_DEPART_COUNTRY") if "SALE_ARRIVAL_COUNTRY" in row: arrival_country_index = row.index("SALE_ARRIVAL_COUNTRY") if "TOTAL_ACTIVITY_VALUE_AMT_VAT_INCL" in row: amount_index = row.index( "TOTAL_ACTIVITY_VALUE_AMT_VAT_INCL") if "SALES_CHANNEL" in row: sales_channel_index = row.index("SALES_CHANNEL") if "ACTIVITY_PERIOD" in row: period_index = row.index("ACTIVITY_PERIOD") if "PRICE_OF_ITEMS_VAT_RATE_PERCENT" in row: vat_perc_index = row.index( "PRICE_OF_ITEMS_VAT_RATE_PERCENT") if "TRANSACTION_EVENT_ID" in row: tx_id_index = row.index("TRANSACTION_EVENT_ID") if "TRANSACTION_TYPE" in row: tx_type_index = row.index("TRANSACTION_TYPE") if "SELLER_SKU" in row: sku_index = row.index("SELLER_SKU") if min([ marketplace_index, depart_country_index, arrival_country_index, amount_index, vat_perc_index, sales_channel_index, period_index, tx_id_index, tx_type_index, sku_index ]) < 0: raise Exception("Not all necessary columns found") header_read = True return report def process_files(data_root, tax_data_root): # pylint: disable=too-many-locals """ Main processing loop """ path_in = os.path.join(data_root, "in") path_out = os.path.join(data_root, "out") path_prc = os.path.join(data_root, "processed") path_err = os.path.join(data_root, "error") path_log = os.path.join(data_root, "log") path_ref = os.path.join(data_root, "ref") path_tax_ref = os.path.join(tax_data_root, "ref") files = os.listdir(path_in) num_files = len(files) print("Found {0} files...".format(num_files)) if num_files == 0: return read_currency_ref(path_ref) read_ean_lookup(path_tax_ref) for f in files: summary_filename_base = os.path.splitext(f)[0] in_file = os.path.join(path_in, f) prc_file = os.path.join(path_prc, f) err_file = os.path.join(path_err, f) if os.path.isfile(in_file): setup_logging(path_log) logging.debug("Summarizing Amazon Invoice '%s'...", in_file) try: report_data: TransactionSummaryReport = summarize_transactions( in_file) write_summary_reports(path_out, summary_filename_base, report_data) shutil.move(in_file, prc_file) except Exception as e: # pylint: disable=broad-except logging.exception(e) shutil.move(in_file, err_file) def main(): """ Main entry point """ data_root = get_data_root() invoice_data_root = os.path.join(data_root, "Matching_Abrechnung_4") tax_data_root = invoice_data_root process_files(invoice_data_root, tax_data_root) if __name__ == "__main__": main()