from collections import defaultdict import xml.etree.ElementTree as ET import os, re from pprint import pprint from localconfig import * from invoice_as_vrbl import * # lib offices also has a reference to Seller from stores import Stores from offices import Offices from repairedbuyers import RepairedBuyers from datetime import datetime import base64 SORTED_PATH = os.path.join(BASEPATH, "Sorted") till_seq_from_filename = r'^[^_]+_T([^_]+)_[^_]+_([^_.]+)' storenames = Stores() offices = Offices() repairedbuyers = RepairedBuyers() def etree_to_dict(t): t_tag = t.tag if '}' in t_tag: t_tag = t_tag.split('}',1)[1] d = {t_tag: {} if t.attrib else None} children = list(t) if children: dd = defaultdict(list) for dc in map(etree_to_dict, children): for k, v in dc.items(): dd[k].append(v) d = {t_tag: {k:v[0] if len(v) == 1 else v for k, v in dd.items()}} if t.attrib: d[t_tag].update(('@' + k, v) for k, v in t.attrib.items()) if t.text: text = t.text.strip() if children or t.attrib: if text: d[t_tag]['#text'] = text else: d[t_tag] = text return d def search_dictlist(d, k, v): return next((sub for sub in d if sub[k] == v), None) def escape_quotes(txt): if txt and '"' in txt: return f'"{txt.replace('"','\'')}"' return txt def get_value_from_addon(treex, key): result = 'XX' for ao in treex['Addon']: if ao['Key']==key: result = ao['Value'] return result def get_value_from_binary_data(treex, key): result = '' for ao in treex['BinaryData']: if ao['Name']==key: result = ao['Content'] if len(result)>1: result = base64.b64decode(result).decode("utf-8") return result def escape(xmlcontent): if xmlcontent: return xmlcontent.replace('&','&').replace('>','>').replace('<','<') return '' def get_elm_value(tree, name, default=''): names = name.split('/') if names[0] in tree: if len(names) > 1: return get_elm_value(tree[names[0]], '/'.join(names[1:]), default) return escape(tree[name]) return default def dump_buyer_info(buyer, invoice_ref, receipt_ref, p_country): print("dump buyers") if buyer: message = f"{p_country},{invoice_ref},{receipt_ref},{buyer.orgid},{buyer.taxID},{buyer.name},{buyer.city},{buyer.street1},{buyer.zip},{buyer.country}\n" print(message) with open(BUYERS_BROKEN, "a", encoding="utf-8") as lfile: lfile.write(message) def get_seller_from_country(countrycode): row = offices.get_office_details(countrycode) return Seller(row['PartyId'], row['Name'], row['City'], taxid=row['TaxId'], street1=row['Street1'], street2=row['Street2'], zip=row['Zip'], country=row['Country'], scheme=row['PartyScheme'], legalid=row['LegalId'], vrbl_receiver=row['VRBL'],minimum=row['MinimumAmount'], orgidlength=row['orgidlength'], taxidlength=row['taxidlength'], roundingAllowed=row['roundingAllowed']) def get_customer_from_dict(trx): cID = cfullname = ccity = ctaxID = cstreet1 = czip = ccountry = cphone = cemail = '' cFound = False if 'Customer' in trx: cID = get_elm_value(trx['Customer'],'CustomerID') cfullname = get_elm_value(trx['Customer'],'CustomerName/FullName') ccity = get_elm_value(trx['Customer'],'Address/City') cstreet1 = get_elm_value(trx['Customer'],'Address/AddressLine/#text') czip = get_elm_value(trx['Customer'],'Address/PostalCode') ccountry = get_elm_value(trx['Customer'],'Address/Country') ctaxID = get_elm_value(trx['Customer'],'CustomerTaxID') cphone = get_elm_value(trx['Customer'],'Telephone/FullTelephoneNumber') if not cphone: cphone = get_elm_value(trx['Customer'],'Telephone/0/FullTelephoneNumber') cemail = get_elm_value(trx['Customer'],'EMail/EMailAddress') # if a @ in the companyname, then the taxid has to come from the fulname if len(ctaxID)<1 and '@' in cfullname: cfullname, ctaxID = cfullname.split('@', 1) cust = Buyer(cID, cfullname, ccity, taxid=ctaxID, street1=cstreet1, zip=czip, country=ccountry, phone=cphone, email=cemail) return cust return None def get_trxlink_from_dict(trx): link = {} if 'TransactionLink' in trx: if 'RetailStoreID' in trx['TransactionLink']: link['store'] = trx['TransactionLink']['RetailStoreID'] if 'WorkstationID' in trx['TransactionLink']: link['till'] = trx['TransactionLink']['WorkstationID'] if 'SequenceNumber' in trx['TransactionLink']: link['seq'] = trx['TransactionLink']['SequenceNumber'] if 'BusinessDayDate' in trx['TransactionLink']: link['date'] = trx['TransactionLink']['BusinessDayDate'] if len(link) == 4: link['year'], link['month'], link['day'], dummy = link['date'].replace('-','_').replace('+','_').split('_') return link def process_receipt(invoice): receipt_filename = f"{invoice.receipt.store}_T{invoice.receipt.till}_{invoice.receipt.year[2:]}{invoice.receipt.month}{invoice.receipt.day}_{invoice.receipt.seq}.xml" p = os.path.join(SORTED_PATH, invoice.receipt.year, invoice.receipt.month, invoice.receipt.day, invoice.receipt.store, receipt_filename) if os.path.exists(p): e = ET.parse(p) root = e.getroot() tree = etree_to_dict(root) trx = tree['POSLog']['Transaction']['RetailTransaction'] # 2026-01-10T15:03:18.594+01:00" dt_iso = get_elm_value(trx, 'ReceiptDateTime') dt = datetime.fromisoformat(dt_iso) # timezone-aware datetime invoice.receipt.add_time(dt.astimezone().strftime("%H:%M:%S")) # add header brand, store, currency, debitnote country = get_value_from_addon(tree['POSLog']['Transaction']['ReceiptHeaderAddonList'], 'PVHCountry') brand = get_value_from_addon(tree['POSLog']['Transaction']['ReceiptHeaderAddonList'], 'PVHBrand') curr = get_elm_value(tree['POSLog']['Transaction'],'CurrencyCode') isreturn = get_elm_value(tree['POSLog']['Transaction'],'ReceiptReturnedFlag') postvoided = get_elm_value(tree['POSLog']['Transaction'],'PostVoidedFlag') loyalty = get_elm_value(tree['POSLog']['Transaction'],'LoyaltyAccount/CustomerID') fiscalseq = get_elm_value(tree['POSLog']['Transaction'],'FiscalSequenceNumber') isfiscal = get_elm_value(tree['POSLog']['Transaction'],'FiscalFlag') fiscalprinter = get_elm_value(tree['POSLog']['Transaction'],'FiscalPrinterID') is_debit = False if 'NegativeTotalFlag' in trx: if trx['NegativeTotalFlag'] == 'true': is_debit = True invoice.add_header(Header(country, brand, curr, is_debit, isreturn, postvoided, loyalty, fiscalseq, isfiscal, fiscalprinter)) # for some countries you may need to add tax-references if country == 'HR': invoice.header.add_taxref('ZKI',get_value_from_binary_data(tree['POSLog']['Transaction']['TransactionBinaryDataList'], 'FISCALIZATION_RECEIPT_SECURITY')) invoice.header.add_taxref('JIR',get_value_from_binary_data(tree['POSLog']['Transaction']['TransactionBinaryDataList'], 'FISCALIZATION_RECEIPT_FISCAL_CODE')) # add totals if 'Total' in trx: grand = net = vat = 0.0 for e in trx['Total']: if e['@TotalType'] == 'TransactionGrandAmount': grand = e['#text'] elif e['@TotalType'] == 'TransactionNetAmount': net = e['#text'] elif e['@TotalType'] == 'TransactionTaxAmount': vat = e['#text'] if 'NegativeTotalFlag' in trx and trx['NegativeTotalFlag'] == 'true': grand = '-' + grand net = '-' + net vat = '-' + vat invoice.add_totals(Totals(grand, net, vat)) # add buyer, if noit there yet if not invoice.buyer: invoice.add_buyer(get_customer_from_dict(trx), 'receipt') # add seller invoice.add_seller(get_seller_from_country(invoice.process_country)) # add salesitesm, tenders and totalstaxes if 'LineItem' in trx: for i in trx['LineItem']: sequence = 0 if 'SequenceNumber' in i: sequence = i['SequenceNumber'] if i['@VoidFlag']=='false': if 'Sale' in i or 'Return' in i: returnflag = False if 'Sale' in i: s = i['Sale'] else: s = i['Return'] returnflag = True positemid = '' if type(s['POSIdentity']) is list: positemid = s['POSIdentity'][-1]['POSItemID'] else: positemid = s['POSIdentity']['POSItemID'] itemidtext = '' if '#text' in s['ItemID']: itemidtext = s['ItemID']['#text'] elif 'SpecialOrderNumber' in s: itemidtext = 'Order ' + s['SpecialOrderNumber'] article = Article(sequence, returnflag, itemidtext, positemid, escape(s['Description']), s['Quantity']['#text'], s['Quantity']['@UnitOfMeasureCode'], s['RegularSalesUnitPrice']['#text'], s['ExtendedAmount']) # discount if 'RetailPriceModifier' in s: discElm = s['RetailPriceModifier'] if type(discElm) is list: for d in discElm: article.add_discount(Discount(get_elm_value(d,'PromotionID'),d['Amount']['@Action'],d['Amount']['#text'],d['ReasonCode'])) else: article.add_discount(Discount(get_elm_value(discElm, 'PromotionID'),discElm['Amount']['@Action'],discElm['Amount']['#text'],discElm['ReasonCode'])) if 'Tax' in s: t = s['Tax'] article.add_tax(Tax(f"{t['@TaxSubType']}.{t['@TaxType']}",t['@TypeCode'],t['Amount'],t['Percent'],t['TaxGroupID'],t['TaxableAmount'])) invoice.add_article(article) elif 'Tender' in i: t = i['Tender'] invoice.add_tender(Tender(escape(t['@TenderDescription']), t['Amount'], t['@TenderType'], t['@TypeCode'])) elif 'Tax' in i: t = i['Tax'] invoice.add_tax(TotalTax(f"{t['@TaxSubType']}.{t['@TaxType']}",t['@TypeCode'],t['Amount'],t['Percent'],t['TaxGroupID'],t['TaxableAmount'])) else: log(f"No linked receipt file found for {receipt_filename}",2) return None return invoice def process_invoice_request(filepath): # csv-line(str) invoice = None customer = {} receipt = {} invoiceNumber = '' e = ET.parse(filepath) root = e.getroot() tree = etree_to_dict(root) trx = tree['POSLog']['Transaction']['RetailTransaction'] # this incoice request can have a customer or not (if not , the customer info comes from the receipt) # get the receipt-id store = date = till = seq = '' if 'InvoiceNumber' in trx: trx_link = get_trxlink_from_dict(trx) if len(trx_link) > 1: country = storenames.country_for(trx_link['store']) invoice = Invoice(trx['InvoiceNumber'], country) invoice.poslogid = os.path.basename(filepath).replace('.xml', '') invoice.add_receipt(Receipt(trx_link['store'], trx_link['year'], trx_link['month'], trx_link['day'], trx_link['till'], trx_link['seq'])) invoice.add_buyer(get_customer_from_dict(trx), 'invoice') invoice = process_receipt(invoice) if invoice is not None: # repair the buyerdata repairedBuyerData = repairedbuyers.get_buyer_for(invoice.poslogid) if repairedBuyerData: invoice.buyer.update(repairedBuyerData) else: log(f"Invoice reuest {filepath} does not have a related sales",2) else: log(f"Invoice Request {filepath} doen NOT have an invoice number",2) return invoice def file_has_invoice(filename): transactiontytpe = '' with open(filename, 'r', encoding='utf-8') as f: while line := f.readline(): if transactiontytpe in line: return True return False def file_has_sales(filename): transactiontytpe = '' with open(filename, 'r', encoding='utf-8') as f: while line := f.readline(): if transactiontytpe in line: return True return False def find_invoices_in_storeday(store, year, month, day): invoices_fine = [] invoices_broken = [] broken_buyer = 0 invoices_found = False # find files existing in NEW, but not in PRD folder = os.path.join(SORTED_PATH, year, month, day, store) for root, dirs, files in os.walk(folder): for file in files: if (store in file) and '_T000_' in file and file.endswith(".xml"): filepath = os.path.join(root, file) if file_has_invoice(filepath): # print(f"found file {root} {file}") m = re.search(till_seq_from_filename, file) if m: till, seq = m.groups() invoices_found = True invoice = process_invoice_request(filepath) # ok we have an invoice, now lets check if we should/can send it to ecosio if invoice is not None: if invoice.buyer is not None: if abs(invoice.totals.grandAmount) > invoice.seller.minimum: # only invoice if seller and buyer are fron the same country. also try to invoice if there is an non-recognized country if invoice.buyer.country == invoice.country() or invoice.buyer.country not in EU_COUNTRIES: if invoice.buyer.repair_and_check_if_data_is_broken(): # add to broken buyerdata broken_buyer += 1 log(f"receipt {invoice.receipt.id()}: buyer has insufficient details - repair before export",2) dump_buyer_info(invoice.buyer, invoice.poslogid, invoice.receipt.id(), invoice.country()) invoices_broken.append(invoice) else: invoices_fine.append(invoice) else: log(f"Invoice {invoice.poslogid}: Buyer country {invoice.buyer.country} does not match company country {invoice.country()}",1) else: log(f"Invoice {invoice.poslogid}: grandTotal {invoice.totals.grandAmount} not enough for invoicing.") else: log(f"Invoice {invoice.poslogid} has not a valid buyer - cannot export",2) # if not invoices_found: # log(f"No invoices found for {store}, {year}-{month}-{day}") return invoices_fine, invoices_broken if __name__ == '__main__': # several_i, broken_i = find_invoices_in_storeday("AN03","2026","02","06") # several_i, broken_i = find_invoices_in_storeday("A100","2026","02","06") several_i, broken_i = find_invoices_in_storeday("AN19","2026","02","10") for i in several_i: # print(str(i)) a = str(i) # for i in broken_i: # print(str(i))