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 = ""
# close_tag = ""
# rq_open = ""
# rq_close = ""
# elif type == "Inventory Assembly":
# open_tag = ""
# close_tag = ""
# rq_open = ""
# rq_close = ""
# else:
# raise ValueError("Unsupported item type")
# mpn = f"{data['mpn']}" if data.get("mpn") else ""
# xml = f"""
#
#
#
# {rq_open}
# {open_tag}
# {data['name']}
# {mpn}
# {data['sdesc']}
# {data['sprice']}
#
# {data['income_acc']}
#
# {data['pdesc']}
# {data['cost']}
#
# {data['cogs_acc']}
#
#
# {data['pref_vendor']}
#
#
# {data['asset_acc']}
#
# {data['min']}
# {data['max']}
# {data['on_hand']}
# {data['tot_value']}
# {data['as_of']}
# {'false' if data.get('inactive') else 'true'}
# {close_tag}
# {rq_close}
#
#
# """
# 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()