mirror of
https://github.com/bcomsugi/dasaproject.git
synced 2026-01-08 18:42:37 +07:00
588 lines
23 KiB
Python
588 lines
23 KiB
Python
# from . import QBClasses
|
|
from pprint import pprint
|
|
# from QBClass.QBClasses import InvoiceQuery, SalesOrderQuery
|
|
from QBClass.QBClasses import InvoiceQuery, SalesOrderQuery, InvoiceAdd
|
|
# import timeit
|
|
import time
|
|
import json
|
|
import pandas as pd
|
|
import numpy as np
|
|
import datetime
|
|
|
|
print('succes Loading modules')
|
|
|
|
|
|
|
|
def timer(func):
|
|
def wrapper(*args, **kwargs):
|
|
nonlocal total
|
|
start = time.time()
|
|
result = func(*args, **kwargs)
|
|
duration = time.time() - start
|
|
total += duration
|
|
print(f"Function:{func.__name__} Execution time: {duration} Total: {total}")
|
|
return result
|
|
|
|
total = 0
|
|
return wrapper
|
|
|
|
|
|
@timer
|
|
def get_all_so_from_invoice( FromTxnDate=None, ToTxnDate=None, MaxReturned=None,):
|
|
print(MaxReturned, FromTxnDate, ToTxnDate)
|
|
start = time.time()
|
|
print('Get Invoice Query List. Processing..... wait for at minute(1 month=90secs)')
|
|
if MaxReturned:
|
|
iq = InvoiceQuery(MaxReturned= MaxReturned, IncludeLinkedTxns='true', IncludeLineItems='true', debug=False)
|
|
elif FromTxnDate and ToTxnDate:
|
|
print('here')
|
|
iq = InvoiceQuery(TxnDateRangeFilter_FromTxnDate=FromTxnDate, TxnDateRangeFilter_ToTxnDate=ToTxnDate, IncludeLinkedTxns='true', IncludeLineItems='true', debug=False)
|
|
else:
|
|
iq = InvoiceQuery(TxnDateRangeFilter_FromTxnDate='2021-01-01', TxnDateRangeFilter_ToTxnDate='2021-03-30', IncludeLinkedTxns='true', IncludeLineItems='true', debug=False)
|
|
# pprint(iq.all(), sort_dicts=False)
|
|
# print(iq.all())
|
|
print(f"Execution time InvoiceQuery: {time.time()-start} {len(iq.all()) = }")
|
|
so_list = []
|
|
iq_list = []
|
|
dup_so_list = []
|
|
for idx, txn in enumerate(iq.all()):
|
|
# iq_list.append(txn)
|
|
# print(f"{idx = } {txn['RefNumber'] = } {txn['TxnDate'] = } {txn['Subtotal'] = } ")
|
|
if 'LinkedTxn' in txn:
|
|
# pprint(txn['LinkedTxn'], sort_dicts=False)
|
|
if not isinstance(txn['LinkedTxn'], list): #if there is no receive payment and only 1 linked traction, need to change to a list. RECORD it
|
|
txn_linkedTxn = [txn['LinkedTxn']]
|
|
else:
|
|
txn_linkedTxn = txn['LinkedTxn']
|
|
for linkedtxn in txn_linkedTxn:
|
|
if linkedtxn['TxnType']=='SalesOrder':
|
|
if linkedtxn['RefNumber'] not in so_list:
|
|
so_list.append(linkedtxn['RefNumber'])
|
|
else:
|
|
dup_so_list.append(linkedtxn['RefNumber'])
|
|
|
|
print(f'{dup_so_list = }')
|
|
print()
|
|
so_dict = {}
|
|
print(f"Execution time before SO: {time.time()-start}")
|
|
print('Get Sales Order Query List. Processing..... wait for at minute(1 month=130 secs)')
|
|
so = SalesOrderQuery(RefNumber = so_list, IncludeLinkedTxns='true', IncludeLineItems='true', debug=False)
|
|
duplicateSO = []
|
|
soWithNoLinkedTxn = []
|
|
print(f"Execution time SalesOrderQuery: {time.time()-start}")
|
|
for idx, txn in enumerate(so.all()):
|
|
if 'LinkedTxn' in txn:
|
|
if txn['RefNumber'] not in so_dict:
|
|
so_dict[txn['RefNumber']] = txn
|
|
else:
|
|
duplicateSO.append(txn['RefNumber'])
|
|
else:
|
|
soWithNoLinkedTxn.append(txn)
|
|
# pprint(so.all(), sort_dicts=False)
|
|
res = next(iter(so_dict))
|
|
# print(f'{so_dict[res] = }')
|
|
# pprint(so_dict[res])
|
|
print(f'{soWithNoLinkedTxn = }')
|
|
print(f'{len(iq.all()) = } {len(so.all()) = } {len(so_list) = } {len(dup_so_list) = } {len(so_dict) = }')
|
|
print(f'{duplicateSO = }')
|
|
print()
|
|
### start processing like below module ###
|
|
pass
|
|
return iq.all(), so_dict
|
|
|
|
|
|
@timer
|
|
def process():
|
|
iq = InvoiceQuery(MaxReturned= 20, IncludeLinkedTxns='true', IncludeLineItems='true', debug=False)
|
|
# iq = InvoiceQuery(TxnDateRangeFilter_FromTxnDate='2024-02-01', TxnDateRangeFilter_ToTxnDate='2024-02-29', IncludeLinkedTxns='true', IncludeLineItems='true')
|
|
pprint(iq.all(), sort_dicts=False)
|
|
# print(iq.all())
|
|
nolinkInv = []
|
|
soNotInOneInv = []
|
|
for idx, txn in enumerate(iq.all()):
|
|
print(f"{idx = } {txn['RefNumber'] = } {txn['TxnDate'] = } {txn['Subtotal'] = } ")
|
|
# print(f"{txn['Subtotal'] = }")
|
|
if 'LinkedTxn' in txn:
|
|
# pprint(txn['LinkedTxn'], sort_dicts=False)
|
|
|
|
if not isinstance(txn['LinkedTxn'], list): #if there is no receive payment and only 1 linked traction, need to change to a list. RECORD it
|
|
txn_linkedTxn = [txn['LinkedTxn']]
|
|
else:
|
|
txn_linkedTxn = txn['LinkedTxn']
|
|
|
|
for linkedtxn in txn_linkedTxn:
|
|
try:
|
|
if linkedtxn['TxnType']=='SalesOrder':
|
|
so = SalesOrderQuery(RefNumber = linkedtxn['RefNumber'], IncludeLinkedTxns='true', debug=False)
|
|
is_soLinkedToOneInvoice = False
|
|
if 'LinkedTxn' in so.all():
|
|
if not isinstance(so.all()['LinkedTxn'], list):
|
|
# print(so.all())
|
|
so_linkedTxn = [so.all()['LinkedTxn']] #make a list
|
|
else:
|
|
so_linkedTxn = so.all()['LinkedTxn'] #just copy, already list
|
|
# print(so.all())
|
|
for solinkedtxn in so_linkedTxn:
|
|
# print(len(so_linkedTxn))
|
|
if solinkedtxn['TxnType']=='Invoice' and len(so_linkedTxn)==1:
|
|
# print(so.all()['RefNumber'], 'the only one SO')
|
|
is_soLinkedToOneInvoice=True
|
|
# pass
|
|
else:
|
|
is_soLinkedToOneInvoice=False
|
|
print(so.all()['RefNumber'], 'NOT the only One, this SO have other Invoice number') #make append to a list
|
|
soNotInOneInv.append(so.all()['RefNumber'])
|
|
|
|
if float(linkedtxn['Amount'])<0:
|
|
if so.all()['TotalAmount']!=linkedtxn['Amount'][1:]:
|
|
if is_soLinkedToOneInvoice: #maybe the SO is manually closed, check it item by item, find which item is not in invoice
|
|
if so.all()['IsManuallyClosed'] == 'true':
|
|
pass
|
|
print(f"{so.all()['TxnID'] = } {so.all()['RefNumber'] = }")
|
|
else:
|
|
print('SO TotalAmount<>Amount in Invoice. not Manually closed and not fully Invoiced')
|
|
pprint(f'{linkedtxn = }', sort_dicts=False)
|
|
print(f"{so.all()['TxnID'] = } {so.all()['RefNumber'] = }")
|
|
print(so.all())
|
|
else:
|
|
pass # this SO is fully invoiced, starting process to export the details
|
|
else:
|
|
print('Linkedtxn amount is positif(should be negatif')
|
|
except Exception as e:
|
|
print('ERROR')
|
|
pprint(linkedtxn, sort_dicts=False)
|
|
print(f"{so.all()['TxnID'] = }")
|
|
print(so.all())
|
|
print(e)
|
|
break
|
|
else:
|
|
nolinkInv.append(txn)
|
|
print(f'{len(nolinkInv) = }')
|
|
|
|
|
|
@timer
|
|
def process_data(iq_list, so_dict):
|
|
print('process_data')
|
|
# iq = InvoiceQuery(MaxReturned= 20, IncludeLinkedTxns='true', IncludeLineItems='true', debug=False)
|
|
# iq = InvoiceQuery(TxnDateRangeFilter_FromTxnDate='2024-02-01', TxnDateRangeFilter_ToTxnDate='2024-02-29', IncludeLinkedTxns='true', IncludeLineItems='true')
|
|
# pprint(iq_list, sort_dicts=False)
|
|
# print(iq_list)
|
|
nolinkInv = []
|
|
soNotInOneInv = []
|
|
manuallyClosedSO = []
|
|
openSO = []
|
|
fullyInvoicedSO = []
|
|
so_list = [x for x in so_dict]
|
|
print(f'{len(so_list) = }')
|
|
result_iq_list = []
|
|
for idx, txn in enumerate(iq_list):
|
|
# print(f"{idx = } {txn['RefNumber'] = } {txn['TxnDate'] = } {txn['Subtotal'] = } ")
|
|
# print(f"{txn['Subtotal'] = }")
|
|
|
|
if 'LinkedTxn' in txn:
|
|
# pprint(txn['LinkedTxn'], sort_dicts=False)
|
|
if not isinstance(txn['LinkedTxn'], list): #if there is no receive payment and only 1 linked traction, need to change to a list. RECORD it
|
|
txn_linkedTxn = [txn['LinkedTxn']]
|
|
else:
|
|
txn_linkedTxn = txn['LinkedTxn']
|
|
|
|
for linkedtxn in txn_linkedTxn:
|
|
# try:
|
|
if linkedtxn['TxnType']=='SalesOrder':
|
|
try:
|
|
# so_list.remove(linkedtxn['RefNumber'])
|
|
# so = SalesOrderQuery(RefNumber = linkedtxn['RefNumber'], IncludeLinkedTxns='true', debug=False)
|
|
sodt = so_dict[linkedtxn['RefNumber']]
|
|
is_soLinkedToOneInvoice = False
|
|
if sodt['IsManuallyClosed']=='true':
|
|
manuallyClosedSO.append(sodt['RefNumber'])
|
|
if sodt['IsFullyInvoiced']=='true':
|
|
fullyInvoicedSO.append(sodt['RefNumber'])
|
|
if 'LinkedTxn' in sodt:
|
|
if not isinstance(sodt['LinkedTxn'], list):
|
|
# print(sodt)
|
|
so_linkedTxn = [sodt['LinkedTxn']] #make a list
|
|
else:
|
|
so_linkedTxn = sodt['LinkedTxn'] #just copy, already list
|
|
# print(sodt)
|
|
for solinkedtxn in so_linkedTxn:
|
|
# print(len(so_linkedTxn))
|
|
if solinkedtxn['TxnType']=='Invoice' and len(so_linkedTxn)==1:
|
|
# print(sodt['RefNumber'], 'the only one SO')
|
|
is_soLinkedToOneInvoice=True
|
|
# pass
|
|
else:
|
|
is_soLinkedToOneInvoice=False
|
|
print(sodt['RefNumber'], 'NOT the only One, this SO have other Invoice number') #make append to a list
|
|
soNotInOneInv.append(sodt['RefNumber'])
|
|
|
|
if float(linkedtxn['Amount'])<0:
|
|
# if sodt['TotalAmount']!=linkedtxn['Amount'][1:]:
|
|
# pass
|
|
# if is_soLinkedToOneInvoice: #maybe the SO is manually closed, check it item by item, find which item is not in invoice
|
|
# if sodt['IsManuallyClosed'] == 'true':
|
|
# pass
|
|
# print(f"{sodt['TxnID'] = } {sodt['RefNumber'] = }")
|
|
# manuallyClosedSO.append(sodt['RefNumber'])
|
|
# else:
|
|
# print('SO TotalAmount<>Amount in Invoice. not Manually closed and not fully Invoiced')
|
|
# pprint(f'{linkedtxn = }', sort_dicts=False)
|
|
# print(f"{sodt['TxnID'] = } {sodt['RefNumber'] = }")
|
|
# print(sodt)
|
|
# openSO.append(sodt['RefNumber'])
|
|
# else:
|
|
pass # this SO is fully invoiced, starting process to export the details
|
|
try:
|
|
if not isinstance(sodt['SalesOrderLineRet'],list):
|
|
sodt['SalesOrderLineRet'] = [sodt['SalesOrderLineRet']]
|
|
if not isinstance(txn['InvoiceLineRet'],list):
|
|
txn['InvoiceLineRet'] = [txn['InvoiceLineRet']]
|
|
for so_line in sodt['SalesOrderLineRet']:
|
|
checklist = ['ItemRef', ]
|
|
if len([ i for i in checklist if i in so_line])==0:
|
|
continue
|
|
sotxnlineid = so_line['TxnLineID']
|
|
soitemref_fullname = so_line.get('ItemRef',{}).get('FullName')
|
|
soquantity:str = so_line.get('Quantity')
|
|
# print(f'{soquantity = }')
|
|
sounitofmeasure = so_line.get('UnitOfMeasure')
|
|
# sooverrideuomsetref_fullname = so_line['OverrideUOMSetRef']['FullName']
|
|
sorate = so_line.get('Rate')
|
|
soamount = so_line.get('Amount')
|
|
soinvoiced = so_line.get('Invoiced')
|
|
soismanuallyclosed = so_line['IsManuallyClosed']
|
|
soother2 = so_line.get('Other2')
|
|
|
|
#check compare to the invoicelineret
|
|
for inv_line in txn['InvoiceLineRet']: #loop start from top in order
|
|
is_inv_so_line_ok = True
|
|
if 'soline' not in inv_line: #this line has no so link yet
|
|
if inv_line.get('ItemRef',{}).get('FullName') != soitemref_fullname:
|
|
continue
|
|
if inv_line.get('UnitOfMeasure') != sounitofmeasure:
|
|
print(f"{inv_line['ItemRef']['FullName'] = } {inv_line['UnitOfMeasure']=} != {sounitofmeasure=}")
|
|
is_inv_so_line_ok = False
|
|
#do convertion????
|
|
# continue
|
|
if soinvoiced and inv_line.get('Quantity',0)!=soinvoiced: #compre with the invoiced
|
|
print(f"{inv_line['ItemRef']['FullName'] = } {inv_line['Quantity']=} != {soquantity=}")
|
|
is_inv_so_line_ok = False
|
|
if sorate:
|
|
if float(inv_line.get('Rate',0))!=float(sorate):
|
|
print(f"{txn['RefNumber'] = } {inv_line['ItemRef']['FullName'] = } {inv_line['Rate']=} <> {sorate=} ; {float(inv_line.get('Rate',0)) = } {float(sorate) = }")
|
|
is_inv_so_line_ok = False
|
|
# if soamount:
|
|
# if float(inv_line.get('Amount',0))!=float(soamount):
|
|
# print(f"{txn['RefNumber']} {inv_line['ItemRef']['FullName'] = } {inv_line['Amount']=} != {soamount=}")
|
|
# is_inv_so_line_ok = False
|
|
if not is_inv_so_line_ok:
|
|
print(f"{inv_line['ItemRef']['FullName'] = } Some detail not equal")
|
|
continue
|
|
#add to the spesific invoiceline
|
|
inv_line['soline']= {'TxnID':sodt['TxnID'],
|
|
'RefNumber':sodt['RefNumber'],
|
|
'TxnDate':sodt['TxnDate'],
|
|
'TxnLineID': sotxnlineid,
|
|
'ItemRef_FullName':soitemref_fullname,
|
|
'Quantity':soquantity,
|
|
'UnitOfMeasure':sounitofmeasure,
|
|
'Rate':sorate,
|
|
'Amount':soamount,
|
|
'Invoiced':soinvoiced,
|
|
'IsManuallyClosed':soismanuallyclosed,
|
|
'Other2':soother2}
|
|
break
|
|
except Exception as e:
|
|
print(f"SO {sodt['RefNumber'] = } {txn['RefNumber'] = }")
|
|
print(f'ERROR: {e}')
|
|
else:
|
|
print('Linkedtxn amount is positif(should be negatif')
|
|
except Exception as e:
|
|
# print('ERROR')
|
|
# pprint(linkedtxn, sort_dicts=False)
|
|
# print(f"{sodt['TxnID'] = }")
|
|
print(f"{sodt['RefNumber'] = } {txn['RefNumber'] = }")
|
|
print(f'ERROR: {e}')
|
|
break
|
|
else:
|
|
nolinkInv.append(txn)
|
|
# c =next(iter(iq_list))
|
|
# print('TEST RESULT:')
|
|
# pprint(c, sort_dicts=False)
|
|
# print(f'{len(nolinkInv) = }\n{nolinkInv = }\n{soNotInOneInv = }\n{manuallyClosedSO = }\n{openSO = }\n{len(fullyInvoicedSO) = }')
|
|
c = [item for item in so_list if item not in fullyInvoicedSO]
|
|
print(f'not fuuly invoice, leftover SO: {c}')
|
|
return iq_list
|
|
|
|
|
|
def checking_iqwith_so(iq_list):
|
|
if not iq_list:
|
|
return False
|
|
if not isinstance(iq_list,list):
|
|
iq_list = [iq_list]
|
|
inv_nolineret = []
|
|
inv_line_no_soline = []
|
|
inv_line_no_itemref = []
|
|
for txn in iq_list:
|
|
if 'InvoiceLineRet' not in txn:
|
|
print(f"{txn['RefNumber'] = } doesnt have InvoiceLineRet")
|
|
inv_nolineret.append(txn['RefNumber'])
|
|
continue
|
|
# if 'LinkedTxn'
|
|
if not isinstance(txn['InvoiceLineRet'], list):
|
|
txn['InvoiceLineRet'] = [txn['InvoiceLineRet']]
|
|
for idx, inv_line in enumerate(txn['InvoiceLineRet']):
|
|
if 'ItemRef' not in inv_line:
|
|
inv_line_no_itemref.append({'RefNumber':txn['RefNumber'],
|
|
'idx':idx,})
|
|
continue
|
|
if 'soline' in inv_line:
|
|
continue
|
|
|
|
if '400_Sales' not in inv_line['ItemRef']['FullName'] and 'Sales Promo Discount' not in inv_line['ItemRef']['FullName']:
|
|
inv_line_no_soline.append({'RefNumber':txn['RefNumber'],
|
|
'idx':idx,
|
|
'ItemRef_FullName':inv_line['ItemRef']['FullName'],
|
|
'Amount':inv_line['Amount']})
|
|
print(f'{inv_line_no_soline = }')
|
|
print(f'{inv_line_no_itemref = }')
|
|
print(f'{inv_nolineret = }')
|
|
if inv_line_no_soline or inv_line_no_itemref or inv_nolineret:
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
def make_invoiceadd_dict(iq_list):
|
|
print('make_invoiceadd_dict')
|
|
# status = checking_iqwith_so(iq_list)
|
|
# print(status)
|
|
# if not status:
|
|
# return False
|
|
inv_nolineret = []
|
|
inv_line_no_soline = []
|
|
inv_line_no_itemref = []
|
|
invoiceaddlist = []
|
|
invadddict = {}
|
|
for inv_line in iq_list:
|
|
# if 'InvoiceLineRet' not in txn:
|
|
# print(f"{txn['RefNumber'] = } doesnt have InvoiceLineRet")
|
|
# inv_nolineret.append(txn['RefNumber'])
|
|
# continue
|
|
# if not isinstance(txn['InvoiceLineRet'], list):
|
|
# txn['InvoiceLineRet'] = [txn['InvoiceLineRet']]
|
|
# for idx, inv_line in enumerate(txn['InvoiceLineRet']):
|
|
invadddict = {}
|
|
# pprint(inv_line, sort_dicts=False)
|
|
invadddict['CustomerRef_FullName'] = inv_line['CustomerRef']['FullName']
|
|
invadddict['ARAccountRef_FullName']= inv_line['ARAccountRef']['FullName']
|
|
invadddict['TemplateRef_FullName'] = inv_line['TemplateRef']['FullName']
|
|
invadddict['TxnDate'] = inv_line['TxnDate']
|
|
invadddict['RefNumber'] = inv_line['RefNumber']
|
|
if 'BillAddress' in inv_line:
|
|
invadddict['BillAddress_Addr1'] = inv_line['BillAddress'].get('Addr1')
|
|
invadddict['BillAddress_Addr2'] = inv_line['BillAddress'].get('Addr2')
|
|
invadddict['BillAddress_Addr3'] = inv_line['BillAddress'].get('Addr3')
|
|
invadddict['BillAddress_Addr4'] = inv_line['BillAddress'].get('Addr4')
|
|
invadddict['BillAddress_Addr5'] = inv_line['BillAddress'].get('Addr5')
|
|
invadddict['BillAddress_City'] = inv_line['BillAddress'].get('City')
|
|
invadddict['BillAddress_State'] = inv_line['BillAddress'].get('State')
|
|
invadddict['BillAddress_PostalCode'] = inv_line['BillAddress'].get('PostalCode')
|
|
invadddict['BillAddress_Country'] = inv_line['BillAddress'].get('Country')
|
|
invadddict['BillAddress_Note'] = inv_line['BillAddress'].get('Note')
|
|
if 'ShipAddress' in inv_line:
|
|
invadddict['ShipAddress_Addr1'] = inv_line['ShipAddress'].get('Addr1')
|
|
invadddict['ShipAddress_Addr2'] = inv_line['ShipAddress'].get('Addr2')
|
|
invadddict['ShipAddress_Addr3'] = inv_line['ShipAddress'].get('Addr3')
|
|
invadddict['ShipAddress_Addr4'] = inv_line['ShipAddress'].get('Addr4')
|
|
invadddict['ShipAddress_Addr5'] = inv_line['ShipAddress'].get('Addr5')
|
|
invadddict['ShipAddress_City'] = inv_line['ShipAddress'].get('City')
|
|
invadddict['ShipAddress_State'] = inv_line['ShipAddress'].get('State')
|
|
invadddict['ShipAddress_PostalCode'] = inv_line['ShipAddress'].get('PostalCode')
|
|
invadddict['ShipAddress_Country'] = inv_line['ShipAddress'].get('Country')
|
|
invadddict['ShipAddress_Note'] = inv_line['ShipAddress'].get('Note')
|
|
if inv_line.get('PONumber'):
|
|
invadddict['PONumber'] = inv_line.get('PONumber')
|
|
if 'TermsRef' in inv_line:
|
|
invadddict['TermsRef_FullName'] = inv_line['TermsRef']['FullName']
|
|
invadddict['DueDate'] = inv_line['DueDate']
|
|
if 'SalesRepRef' in inv_line:
|
|
invadddict['SalesRepRef_FullName'] = inv_line['SalesRepRef']['FullName']
|
|
invadddict['ShipDate'] = inv_line['ShipDate']
|
|
if inv_line.get('Memo'):
|
|
invadddict['Memo'] = inv_line['Memo']
|
|
invadddict['IsToBePrinted'] = "false"
|
|
invadddict['IsToBeEmailed'] = "false"
|
|
if 'Other' in inv_line :
|
|
invadddict['Other'] = inv_line['Other']
|
|
|
|
#InvoiceLineAdd BEGIN
|
|
invoicelineadd_list = []
|
|
txnid_list = []
|
|
if not isinstance(inv_line['InvoiceLineRet'], list):
|
|
inv_line['InvoiceLineRet'] = [inv_line['InvoiceLineRet']]
|
|
|
|
for idx, invlineret in enumerate(inv_line['InvoiceLineRet']):
|
|
# print(f'{invlineret = }')
|
|
if 'ItemRef' not in invlineret:
|
|
inv_line_no_itemref.append({'RefNumber':inv_line['RefNumber'],
|
|
'idx':idx,})
|
|
continue
|
|
invlineadd={}
|
|
if 'soline' not in invlineret:
|
|
invlineadd['ItemRef_FullName'] = invlineret['ItemRef']['FullName']
|
|
# invlineadd['Desc'] = invlineret['Desc']
|
|
if 'Quantity' in invlineret:
|
|
invlineadd['Quantity'] = invlineret['Quantity']
|
|
if 'UnitOfMeasure' in invlineret:
|
|
invlineadd['UnitOfMeasure'] = invlineret['UnitOfMeasure']
|
|
invlineadd['Rate'] = invlineret['Rate']
|
|
invlineadd['Amount'] = invlineret['Amount']
|
|
if 'Other1' in invlineret:
|
|
invlineadd['Other1'] = invlineret['Other1']
|
|
if 'Other2' in invlineret:
|
|
invlineadd['Other2'] = invlineret['Other2']
|
|
# print(f"{invlineadd = } {invlineret['ItemRef']['FullName']=}")
|
|
### WARNING, check with the real TxnID in new QB. Replace the soline TxnID and TxnLineID with New QB SalesOrder
|
|
if 'soline' in invlineret: ### WARNING, check with the real TxnID in new QB
|
|
invlineadd['LinkToTxn_TxnID'] = invlineret['soline']['TxnID']
|
|
invlineadd['LinkToTxn_TxnLineID'] = invlineret['soline']['TxnLineID']
|
|
txnid_list.append(invlineret['soline']['TxnID'])
|
|
invoicelineadd_list.append(invlineadd)
|
|
# print(invlineadd)
|
|
# print(invoicelineadd_list)
|
|
txnid = list(set(txnid_list))
|
|
# if txnid:
|
|
# invadddict['LinkToTxnID']=txnid
|
|
invadddict['InvoiceLineAdd'] = invoicelineadd_list
|
|
invoiceaddlist.append(invadddict)
|
|
|
|
# pprint(invadddict, sort_dicts=False)
|
|
|
|
# pprint(invoiceaddlist, sort_dicts=False)
|
|
|
|
for x in invoiceaddlist:
|
|
if '10671' in x['RefNumber']:
|
|
pprint(x, sort_dicts=False)
|
|
break
|
|
# for y in x['InvoiceLineAdd']:
|
|
# if '10671' in y['ItemRef_FullName']:
|
|
# pprint(x, sort_dicts=False)
|
|
# break
|
|
print(f'make_invoiceadd_dict - >{len(invoiceaddlist) = }')
|
|
return invoiceaddlist
|
|
|
|
def writeToFile(iq_list=None, so_dict=None, filename = "", suffix="", indent=2):
|
|
if not filename:
|
|
return False
|
|
try:
|
|
if iq_list:
|
|
if indent==None:
|
|
iq_list_json = json.dumps(iq_list)
|
|
else:
|
|
iq_list_json = json.dumps(iq_list, indent=indent)
|
|
with open(f"{filename}{suffix}.json", "w") as outfile:
|
|
outfile.write(iq_list_json)
|
|
if so_dict:
|
|
if indent==None:
|
|
so_dict_json = json.dumps(so_dict)
|
|
else:
|
|
so_dict_json = json.dumps(so_dict, indent=indent)
|
|
with open(f"{filename}_so_dict{suffix}.json", "w") as outfile:
|
|
outfile.write(so_dict_json)
|
|
return True
|
|
except Exception as e:
|
|
print(e)
|
|
return False
|
|
|
|
def readJsonFromFile(filename):
|
|
try:
|
|
with open(filename, "r") as infile:
|
|
_list = json.load(infile)
|
|
return _list
|
|
except Exception as e:
|
|
print(e)
|
|
return []
|
|
|
|
|
|
def get_last_date_of_month(stryearmonth:str):
|
|
# Get Last date of Month
|
|
# Using replace() + timedelta()
|
|
|
|
# initializing date
|
|
test_date = datetime.datetime.fromisoformat(stryearmonth)
|
|
# test_date = datetime.datetime(2018, 6, 4)
|
|
|
|
# printing original date
|
|
print("The original date is : " + str(test_date))
|
|
|
|
# getting next month
|
|
# using replace to get to last day + offset
|
|
# to reach next month
|
|
nxt_mnth = test_date.replace(day=28) + datetime.timedelta(days=4)
|
|
|
|
# subtracting the days from next month date to
|
|
# get last date of current Month
|
|
res = nxt_mnth - datetime.timedelta(days=nxt_mnth.day)
|
|
|
|
# printing result
|
|
datesplit = stryearmonth.split('-')
|
|
print("Last date of month : " + str(res.day), f"{datesplit[0]}-{datesplit[1]}-{str(res.day)}")
|
|
|
|
return f"{datesplit[0]}-{datesplit[1]}-{str(res.day)}"
|
|
|
|
@timer
|
|
def main(fromtxndate, totxndate, maxreturned:int=None):
|
|
# print(timeit.repeat(process, repeat=1))
|
|
# process()
|
|
|
|
invqueryfilename = f'exim\Data\{fromtxndate}iq'
|
|
|
|
# if maxreturned:
|
|
# iq_list, so_dict = get_all_so_from_invoice(MaxReturned=maxreturned)
|
|
|
|
# else:
|
|
# try:
|
|
# _fromdate = datetime.datetime.fromisoformat(fromtxndate)
|
|
# _todate = datetime.datetime.fromisoformat(totxndate)
|
|
# except Exception as e:
|
|
# print('date format should be yyyy-mm-dd example: "2024-03-09"')
|
|
# return False
|
|
|
|
# ## Reading from QB and write to a file depends on txndate
|
|
# iq_list, so_dict = get_all_so_from_invoice(FromTxnDate=fromtxndate, ToTxnDate=totxndate)
|
|
# writeToFile(iq_list, so_dict, filename=invqueryfilename)
|
|
|
|
### reading from existing file the iq_list and so_dict jsonfile
|
|
|
|
iq_list = readJsonFromFile(f"{invqueryfilename}.json")
|
|
so_dict = readJsonFromFile(f"{invqueryfilename}_so_dict.json")
|
|
print(f'{len(iq_list) = } {len(so_dict) = }')
|
|
iq_list = process_data(iq_list, so_dict)
|
|
suffix = '_withsoindent2'
|
|
writeToFile(iq_list, filename=f'{invqueryfilename}', suffix=suffix, indent=2)
|
|
iq_list = readJsonFromFile(f'{invqueryfilename}{suffix}.json')
|
|
print(len(iq_list))
|
|
print(f'Seems {invqueryfilename}{suffix}.json is {checking_iqwith_so(iq_list)}') #checking the iq_list. is it good to import to new QB?
|
|
|
|
return make_invoiceadd_dict(iq_list)
|
|
|
|
if __name__=='__main__':
|
|
# print(np.arange('2021-02', '2021-03', dtype='datetime64[D]'))
|
|
fromtxndate = '2024-07-01'
|
|
totxndate = get_last_date_of_month(fromtxndate)
|
|
_fromdate = datetime.datetime.fromisoformat(fromtxndate)
|
|
|
|
print(fromtxndate)
|
|
|
|
readydata = main(fromtxndate, totxndate)
|
|
# pprint(readydata, sort_dicts=False)
|
|
|
|
IA = InvoiceAdd(*readydata, debug=False)
|
|
# print(IA.create_QBXML())
|
|
filename = f'exim\Data\{fromtxndate}_qbxml.txt'
|
|
|
|
with open(f"{filename}", "w") as outfile:
|
|
outfile.write(IA.create_QBXML())
|
|
# IA.all()
|