add exim for export import

This commit is contained in:
bcomsugi 2025-02-19 06:58:18 +07:00
parent a110b9ae99
commit 86ff995133
4 changed files with 1882 additions and 1742 deletions

0
Exim/__init__.py Normal file
View File

132
Exim/exim.py Normal file
View File

@ -0,0 +1,132 @@
# from . import QBClasses
from pprint import pprint
from QBClass.QBClasses import InvoiceQuery, SalesOrderQuery
# import timeit
import time
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"Execution time: {duration} Total: {total}")
return result
total = 0
return wrapper
@timer
def get_all_so_from_invoice():
# iq = InvoiceQuery(MaxReturned= 20, IncludeLinkedTxns='true', IncludeLineItems='true')
start = time.time()
print('Get Invoice Query List. Processing..... wait for at minute(1 month=90secs)')
iq = InvoiceQuery(TxnDateRangeFilter_FromTxnDate='2024-09-1', TxnDateRangeFilter_ToTxnDate='2024-09-31', IncludeLinkedTxns='true', IncludeLineItems='true')
# 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)
print(f"Execution time SalesOrderQuery: {time.time()-start}")
for idx, txn in enumerate(so.all()):
so_dict[txn['RefNumber']] = txn
# pprint(so.all(), sort_dicts=False)
res = next(iter(so_dict))
print(f'{so_dict[res] = }')
print(f'{len(iq.all()) = } {len(so.all()) = } {len(so_list) = } {len(dup_so_list) = } {len(so_dict) = }')
@timer
def process():
# iq = InvoiceQuery(MaxReturned= 20, IncludeLinkedTxns='true', IncludeLineItems='true')
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())
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']]
else:
so_linkedTxn = so.all()['LinkedTxn']
# 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')
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 is one SO is fully invoiced
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
# print(timeit.repeat(process, repeat=1))
# process()
get_all_so_from_invoice()

View File

@ -975,6 +975,11 @@ class SalesOrderAdd(baseQBQuery):
class SalesOrderQuery(baseQBQuery):
def __init__(self, *args, **kwargs):
if 'debug' in kwargs and isinstance(kwargs['debug'], bool):
self.class_debug=kwargs["debug"]
else:
self.class_debug=False
if self.class_debug:
print(f'{args = }')
print(f'{kwargs = }')
super().__init__(*args, **kwargs)
@ -990,8 +995,7 @@ class SalesOrderQuery(baseQBQuery):
self.defaultFilterKey = "TxnID"
self.className = 'SalesOrderQuery' #Hardcoded because for check if self.__class__.__name__==self.className
self.classNameRq:str = self.__class__.__name__ + 'Rq'
if 'debug' in kwargs and isinstance(kwargs['debug'], bool):
self.class_debug=kwargs["debug"]
self.QBDict[self.classNameRq]={}
### End Required Variable. can put ENUM list below this (eg: self.ENUM_GeneralSummaryReportQuery=[])
@ -1057,6 +1061,7 @@ class SalesOrderQuery(baseQBQuery):
self.QBDict[self.classNameRq]["OwnerID"]=kwargs['OwnerID']
# print(self.classNameRq)
if self.class_debug:
print(self.QBDict)
if self.__class__.__name__==self.className:
self.runCheck() ### running the qbxml connection to get data ###

View File

@ -42,10 +42,12 @@ class baseQBQuery:
self.statusMessage = ""
self.statusSeverity = ""
self.statusOk = False
if self.class_debug:
if self.__class__.__name__=="baseQBQuery":
print("baseqbquey same with classname")
else:
print("accessed from child class")
print("basequery is accessed from ", self.__class__.__name__ + "Rq")
# @timing
@ -79,6 +81,7 @@ class baseQBQuery:
# print(f'{dataDict = }')
# # QBXML = '<?qbxml version="13.0"?>' + xmltodict.unparse(dataDict, pretty=True)
self.QBXML = xmltodict.unparse(dataDict, pretty=True).replace("</?qbxml>", "").replace(f'version="{version}"', f'version="{version}"?')
if self.class_debug:
print(self.QBXML, type(self.QBXML))
return self.QBXML
@ -128,7 +131,7 @@ class baseQBQuery:
print(f'isDataOK -> {self.listOfDict.firstValue() = }')
if self.listOfDict.firstValue().get(self.retName,None)==None:
return False
if self.class_debug:
print(f'{self.statusCode = }, {self.statusMessage = }, {self.statusSeverity = }')
varDict = self.varDict['QBXML']['QBXMLMsgsRs'][self.__class__.__name__+"Rs"]
return True