Release 1.0.0
This commit is contained in:
+358
@@ -0,0 +1,358 @@
|
||||
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):
|
||||
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"
|
||||
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))
|
||||
Reference in New Issue
Block a user