360 lines
16 KiB
Python
360 lines
16 KiB
Python
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 = '<RetailTransaction TransactionStatus="SES:Invoice">'
|
|
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 = '<RetailTransaction TransactionStatus="Finished">'
|
|
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)) |