mirror of
https://github.com/bcomsugi/dasaproject.git
synced 2026-01-08 18:42:37 +07:00
126 lines
5.0 KiB
Python
126 lines
5.0 KiB
Python
import pythoncom
|
|
import win32com.client
|
|
|
|
class QBSDK_NewItem:
|
|
def __init__(self):
|
|
pythoncom.CoInitialize()
|
|
print("[DEBUG} ✅ COM Initialized")
|
|
self.qb = None
|
|
print("[DEBUG] ✅ QBXMLRP2 Dispatch OK")
|
|
self.session_id = None
|
|
self.response_str = ""
|
|
# self.open_connection()
|
|
def open_connection(self):
|
|
print(self.qb)
|
|
self.qb = win32com.client.Dispatch("QBXMLRP2.RequestProcessor")
|
|
print("after dispatch")
|
|
# self.qb.OpenConnection2("DASA2", "DASA2", 1)
|
|
# self.qb.OpenConnection2("", "DASA2", 1)
|
|
# self.qb.OpenConnection2("", "Test qbXML Request", 1)
|
|
op = self.qb.OpenConnection('', 'DASA2')
|
|
#self.qb.OpenConnection('', 'Test qbXML Request')
|
|
print("[DEBUG] ✅ OpenConnection2 completed")
|
|
self.session_id = self.qb.BeginSession("", 2)
|
|
print("ok")
|
|
print("[DEBUG] ✅ BeginSession completed")
|
|
|
|
def close_connection(self):
|
|
if self.session_id:
|
|
self.qb.EndSession(self.session_id)
|
|
self.qb.CloseConnection()
|
|
pythoncom.CoUninitialize()
|
|
print("[DEBUG] ✅ QuickBooks connection closed")
|
|
|
|
def response_request(self, xml: str) -> str:
|
|
try:
|
|
print("\n---SENDING REQUEST TO QUICKBOOKS---\n")
|
|
print(xml)
|
|
|
|
response = self.qb.ProcessRequest(self.session_id, xml)
|
|
|
|
print("\n---RESPONSE RECEIVED FROM QUICKBOOKS---\n")
|
|
print(response if response else "No Response Received!")
|
|
|
|
return response
|
|
except Exception as e:
|
|
print("[DEBUG] ❌ Error Processing Request: ", str(e))
|
|
return ""
|
|
|
|
# def qbxml_itemAdd(self, data):
|
|
# type = data['type'].item_type if hasattr(data['type'], 'item_type') else data['type']
|
|
# purchase_from_vendor = data.get("purchase_from_vendor", False)
|
|
|
|
# if type == "Inventory Part":
|
|
# open_tag = "<ItemInventoryAdd>"
|
|
# close_tag = "</ItemInventoryAdd>"
|
|
# rq_open = "<ItemInventoryAddRq>"
|
|
# rq_close = "</ItemInventoryAddRq>"
|
|
# elif type == "Inventory Assembly":
|
|
# open_tag = "<ItemInventoryAssemblyAdd>"
|
|
# close_tag = "</ItemInventoryAssemblyAdd>"
|
|
# rq_open = "<ItemInventoryAssemblyAddRq>"
|
|
# rq_close = "</ItemInventoryAssemblyAddRq>"
|
|
# else:
|
|
# raise ValueError("Unsupported item type")
|
|
|
|
# mpn = f"<ManufacturerPartNumber >{data['mpn']}</ManufacturerPartNumber>" if data.get("mpn") else ""
|
|
|
|
# xml = f"""<?xml version="1.0" encoding="utf-8"?>
|
|
# <?qbxml version="13.0"?>
|
|
# <QBXML>
|
|
# <QBXMLMsgsRq onError="stopOnError">
|
|
# {rq_open}
|
|
# {open_tag}
|
|
# <Name >{data['name']}</Name>
|
|
# {mpn}
|
|
# <SalesDesc >{data['sdesc']}</SalesDesc>
|
|
# <SalesPrice >{data['sprice']}</SalesPrice>
|
|
# <IncomeAccountRef>
|
|
# <ListID >{data['income_acc']}</ListID>
|
|
# </IncomeAccountRef>
|
|
# <PurchaseDesc >{data['pdesc']}</PurchaseDesc>
|
|
# <PurchaseCost >{data['cost']}</PurchaseCost>
|
|
# <COGSAccountRef>
|
|
# <ListID >{data['cogs_acc']}</ListID>
|
|
# </COGSAccountRef>
|
|
# <PrefVendorRef>
|
|
# <FullName >{data['pref_vendor']}</FullName>
|
|
# </PrefVendorRef>
|
|
# <AssetAccountRef>
|
|
# <ListID >{data['asset_acc']}</ListID>
|
|
# </AssetAccountRef>
|
|
# <ReorderPoint >{data['min']}</ReorderPoint>
|
|
# <Max >{data['max']}</Max>
|
|
# <QuantityOnHand >{data['on_hand']}</QuantityOnHand>
|
|
# <TotalValue >{data['tot_value']}</TotalValue>
|
|
# <InventoryDate >{data['as_of']}</InventoryDate>
|
|
# <IsActive >{'false' if data.get('inactive') else 'true'}</IsActive>
|
|
# {close_tag}
|
|
# {rq_close}
|
|
# </QBXMLMsgsRq>
|
|
# </QBXML>
|
|
# """
|
|
# print("\n🔍 CHECKING QBXML FORMAT BEFORE SENDING:")
|
|
# qbxml = xml.strip()
|
|
# response = self.response_request(qbxml)
|
|
# return response
|
|
|
|
def send_rq_to_qb(self, xmlstr:str):
|
|
qbxml = xmlstr.strip()
|
|
self.response_str = self.response_request(qbxml)
|
|
return self.response_str
|
|
|
|
def send_new_item(dtstr:str):
|
|
print("start send new item")
|
|
qbdt = QBSDK_NewItem()
|
|
print(qbdt)
|
|
qbdt.open_connection()
|
|
qbdt.send_rq_to_qb(dtstr)
|
|
return qbdt.response_str
|
|
|
|
# if __name__=="__main__":
|
|
# test = QBSDK_NewItem()
|
|
# test.open_connection()
|
|
|
|
|
|
|