Update with QB Desktop connection
This commit is contained in:
@@ -0,0 +1,758 @@
|
||||
"""
|
||||
QuickBooks Desktop API Client
|
||||
Handles COM-based communication with QuickBooks Desktop via QBXMLRP2.
|
||||
Requires pywin32 and QuickBooks Desktop SDK installed on Windows.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import logging
|
||||
import xml.etree.ElementTree as ET
|
||||
from typing import Optional, Dict, List, Any
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Check if we're on Windows
|
||||
IS_WINDOWS = sys.platform == 'win32'
|
||||
|
||||
if IS_WINDOWS:
|
||||
try:
|
||||
import pythoncom
|
||||
from win32com.client import Dispatch
|
||||
HAS_WIN32COM = True
|
||||
except ImportError:
|
||||
HAS_WIN32COM = False
|
||||
logger.warning("pywin32 not installed. QuickBooks Desktop integration unavailable.")
|
||||
else:
|
||||
HAS_WIN32COM = False
|
||||
logger.info("Not running on Windows. QuickBooks Desktop integration unavailable.")
|
||||
|
||||
|
||||
# Connection type constants (from QBXMLRP2 SDK)
|
||||
class QBConnectionType:
|
||||
localQBD = 1 # Local QuickBooks Desktop
|
||||
remoteQBD = 2 # Remote QuickBooks Desktop
|
||||
localQBDLaunchUI = 3 # Launch QuickBooks if not running
|
||||
|
||||
|
||||
# File mode constants
|
||||
class QBFileMode:
|
||||
qbFileOpenDoNotCare = 0 # Open any file
|
||||
qbFileOpenSingleUser = 1 # Open in single-user mode
|
||||
qbFileOpenMultiUser = 2 # Open in multi-user mode
|
||||
|
||||
|
||||
@dataclass
|
||||
class QBDCredentials:
|
||||
"""QuickBooks Desktop connection settings."""
|
||||
application_name: str = "QBO Excel Sync"
|
||||
application_id: str = ""
|
||||
company_file: str = "" # Empty string = currently open company file
|
||||
connection_type: int = QBConnectionType.localQBD
|
||||
file_mode: int = QBFileMode.qbFileOpenDoNotCare
|
||||
|
||||
|
||||
class QBDesktopError(Exception):
|
||||
"""QuickBooks Desktop specific error."""
|
||||
def __init__(self, message: str, code: str = None, detail: str = None):
|
||||
self.message = message
|
||||
self.code = code
|
||||
self.detail = detail
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class QBDesktopClient:
|
||||
"""QuickBooks Desktop API Client using COM interface."""
|
||||
|
||||
QBXML_VERSION = "13.0"
|
||||
COUNTRY = "US"
|
||||
|
||||
def __init__(self, credentials: QBDCredentials = None):
|
||||
self.credentials = credentials or QBDCredentials()
|
||||
self.request_processor = None
|
||||
self.session_ticket = None
|
||||
self._connected = False
|
||||
self._company_name = None
|
||||
|
||||
@property
|
||||
def is_available(self) -> bool:
|
||||
"""Check if QBD integration is available on this system."""
|
||||
return IS_WINDOWS and HAS_WIN32COM
|
||||
|
||||
@property
|
||||
def is_connected(self) -> bool:
|
||||
"""Check if connected to QuickBooks Desktop."""
|
||||
return self._connected and self.session_ticket is not None
|
||||
|
||||
@property
|
||||
def company_name(self) -> Optional[str]:
|
||||
"""Get the connected company name."""
|
||||
return self._company_name
|
||||
|
||||
def connect(self) -> bool:
|
||||
"""Connect to QuickBooks Desktop."""
|
||||
if not self.is_available:
|
||||
raise QBDesktopError("QuickBooks Desktop integration not available on this system")
|
||||
|
||||
try:
|
||||
# Initialize COM for this thread
|
||||
# CRITICAL: Flask runs in multi-threaded mode, so we need proper COM initialization
|
||||
# Use CoInitializeEx with COINIT_APARTMENTTHREADED for COM automation objects
|
||||
try:
|
||||
# First, try to uninitialize any previous COM state
|
||||
try:
|
||||
pythoncom.CoUninitialize()
|
||||
except:
|
||||
pass
|
||||
|
||||
# Initialize with single-threaded apartment (STA) - required for UI automation objects
|
||||
pythoncom.CoInitializeEx(pythoncom.COINIT_APARTMENTTHREADED)
|
||||
logger.info("COM initialized with COINIT_APARTMENTTHREADED")
|
||||
except pythoncom.com_error as e:
|
||||
# If already initialized, that's OK
|
||||
if e.hresult == -2147417850: # RPC_E_CHANGED_MODE
|
||||
logger.info("COM already initialized in different mode, continuing...")
|
||||
else:
|
||||
logger.warning(f"COM initialization warning: {e}")
|
||||
except Exception as e:
|
||||
logger.warning(f"COM initialization warning: {e}")
|
||||
|
||||
# Create the request processor using Dispatch (not DispatchEx)
|
||||
# Dispatch works better with apartment-threaded COM objects
|
||||
try:
|
||||
self.request_processor = Dispatch("QBXMLRP2.RequestProcessor")
|
||||
logger.info("QBXMLRP2.RequestProcessor created successfully")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create RequestProcessor: {e}")
|
||||
raise QBDesktopError(
|
||||
f"Cannot create QuickBooks RequestProcessor. "
|
||||
f"Make sure QuickBooks Desktop SDK is installed. Error: {str(e)}",
|
||||
code="SDK_NOT_FOUND"
|
||||
)
|
||||
|
||||
# Open connection
|
||||
try:
|
||||
self.request_processor.OpenConnection2(
|
||||
self.credentials.application_id,
|
||||
self.credentials.application_name,
|
||||
self.credentials.connection_type
|
||||
)
|
||||
logger.info("OpenConnection2 successful")
|
||||
except Exception as e:
|
||||
logger.error(f"OpenConnection2 failed: {e}")
|
||||
raise QBDesktopError(
|
||||
f"Failed to open connection to QuickBooks. Error: {str(e)}",
|
||||
code="CONNECTION_FAILED"
|
||||
)
|
||||
|
||||
# Begin session
|
||||
# NOTE: This may trigger an authorization dialog in QuickBooks Desktop
|
||||
# The user must click "Yes, always" in QuickBooks for this to succeed
|
||||
logger.info("Attempting BeginSession - check QuickBooks for authorization dialog...")
|
||||
try:
|
||||
self.session_ticket = self.request_processor.BeginSession(
|
||||
self.credentials.company_file,
|
||||
self.credentials.file_mode
|
||||
)
|
||||
logger.info(f"BeginSession successful, ticket: {self.session_ticket[:20] if self.session_ticket else 'None'}...")
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
logger.error(f"BeginSession failed: {e}")
|
||||
|
||||
# Try to close the connection
|
||||
try:
|
||||
self.request_processor.CloseConnection()
|
||||
except:
|
||||
pass
|
||||
|
||||
# Parse common session errors
|
||||
if "80040408" in error_msg:
|
||||
raise QBDesktopError(
|
||||
"QuickBooks is not running or no company file is open. "
|
||||
"Please open QuickBooks Desktop with a company file first.",
|
||||
code="QB_NOT_RUNNING"
|
||||
)
|
||||
elif "80040416" in error_msg or "80040417" in error_msg:
|
||||
raise QBDesktopError(
|
||||
"Access denied. When QuickBooks prompts you to allow access, "
|
||||
"please click 'Yes, always allow access'. You may need to "
|
||||
"run QuickBooks as Administrator.",
|
||||
code="ACCESS_DENIED"
|
||||
)
|
||||
elif "80040422" in error_msg:
|
||||
raise QBDesktopError(
|
||||
"No company file is open in QuickBooks. "
|
||||
"Please open a company file and try again.",
|
||||
code="NO_COMPANY_FILE"
|
||||
)
|
||||
elif "80040423" in error_msg:
|
||||
raise QBDesktopError(
|
||||
"The company file is in use in single-user mode. "
|
||||
"Switch to multi-user mode or close other connections.",
|
||||
code="SINGLE_USER_MODE"
|
||||
)
|
||||
else:
|
||||
raise QBDesktopError(
|
||||
f"Failed to start QuickBooks session: {error_msg}",
|
||||
code="SESSION_FAILED"
|
||||
)
|
||||
|
||||
self._connected = True
|
||||
|
||||
# Get company info
|
||||
self._fetch_company_info()
|
||||
|
||||
logger.info(f"Connected to QuickBooks Desktop: {self._company_name}")
|
||||
return True
|
||||
|
||||
except QBDesktopError:
|
||||
self._connected = False
|
||||
raise
|
||||
except Exception as e:
|
||||
self._connected = False
|
||||
error_msg = str(e)
|
||||
logger.error(f"Unexpected error connecting to QBD: {error_msg}")
|
||||
|
||||
# Parse common errors
|
||||
if "80040408" in error_msg:
|
||||
raise QBDesktopError(
|
||||
"QuickBooks is not running. Please open QuickBooks Desktop first.",
|
||||
code="QB_NOT_RUNNING"
|
||||
)
|
||||
elif "80040416" in error_msg:
|
||||
raise QBDesktopError(
|
||||
"Access denied. Please authorize this application in QuickBooks.",
|
||||
code="ACCESS_DENIED"
|
||||
)
|
||||
elif "80040422" in error_msg:
|
||||
raise QBDesktopError(
|
||||
"No company file is open in QuickBooks.",
|
||||
code="NO_COMPANY_FILE"
|
||||
)
|
||||
else:
|
||||
raise QBDesktopError(f"Failed to connect to QuickBooks Desktop: {error_msg}")
|
||||
|
||||
def disconnect(self) -> bool:
|
||||
"""Disconnect from QuickBooks Desktop."""
|
||||
try:
|
||||
if self.session_ticket:
|
||||
self.request_processor.EndSession(self.session_ticket)
|
||||
self.session_ticket = None
|
||||
|
||||
if self.request_processor:
|
||||
self.request_processor.CloseConnection()
|
||||
self.request_processor = None
|
||||
|
||||
self._connected = False
|
||||
self._company_name = None
|
||||
|
||||
# Uninitialize COM
|
||||
pythoncom.CoUninitialize()
|
||||
|
||||
logger.info("Disconnected from QuickBooks Desktop")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error disconnecting from QuickBooks Desktop: {e}")
|
||||
return False
|
||||
|
||||
def _fetch_company_info(self):
|
||||
"""Fetch company information from QuickBooks."""
|
||||
try:
|
||||
response = self._send_request("CompanyQuery", {})
|
||||
if response:
|
||||
company = response.get("CompanyRet", {})
|
||||
self._company_name = company.get("CompanyName", "Unknown Company")
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not fetch company info: {e}")
|
||||
self._company_name = "Connected"
|
||||
|
||||
def _build_qbxml_request(self, request_type: str, request_data: Dict = None) -> str:
|
||||
"""Build a QBXML request string."""
|
||||
request_data = request_data or {}
|
||||
|
||||
# Build the request XML
|
||||
xml_parts = [
|
||||
'<?xml version="1.0" encoding="utf-8"?>',
|
||||
'<?qbxml version="{}">'.format(self.QBXML_VERSION),
|
||||
'<QBXML>',
|
||||
'<QBXMLMsgsRq onError="stopOnError">'
|
||||
]
|
||||
|
||||
# Add the specific request
|
||||
xml_parts.append(f'<{request_type}Rq>')
|
||||
|
||||
if request_type.endswith("Query"):
|
||||
# Query request
|
||||
xml_parts.append(self._dict_to_xml(request_data))
|
||||
elif request_type.endswith("Add"):
|
||||
# Add request
|
||||
base_type = request_type[:-3] # Remove "Add"
|
||||
xml_parts.append(f'<{base_type}Add>')
|
||||
xml_parts.append(self._dict_to_xml(request_data))
|
||||
xml_parts.append(f'</{base_type}Add>')
|
||||
elif request_type.endswith("Mod"):
|
||||
# Modify request
|
||||
base_type = request_type[:-3] # Remove "Mod"
|
||||
xml_parts.append(f'<{base_type}Mod>')
|
||||
xml_parts.append(self._dict_to_xml(request_data))
|
||||
xml_parts.append(f'</{base_type}Mod>')
|
||||
else:
|
||||
xml_parts.append(self._dict_to_xml(request_data))
|
||||
|
||||
xml_parts.append(f'</{request_type}Rq>')
|
||||
xml_parts.append('</QBXMLMsgsRq>')
|
||||
xml_parts.append('</QBXML>')
|
||||
|
||||
return ''.join(xml_parts)
|
||||
|
||||
def _dict_to_xml(self, data: Dict, indent: int = 0) -> str:
|
||||
"""Convert a dictionary to XML string."""
|
||||
xml_parts = []
|
||||
|
||||
for key, value in data.items():
|
||||
if value is None:
|
||||
continue
|
||||
elif isinstance(value, dict):
|
||||
xml_parts.append(f'<{key}>')
|
||||
xml_parts.append(self._dict_to_xml(value, indent + 1))
|
||||
xml_parts.append(f'</{key}>')
|
||||
elif isinstance(value, list):
|
||||
for item in value:
|
||||
if isinstance(item, dict):
|
||||
xml_parts.append(f'<{key}>')
|
||||
xml_parts.append(self._dict_to_xml(item, indent + 1))
|
||||
xml_parts.append(f'</{key}>')
|
||||
else:
|
||||
xml_parts.append(f'<{key}>{self._escape_xml(str(item))}</{key}>')
|
||||
else:
|
||||
xml_parts.append(f'<{key}>{self._escape_xml(str(value))}</{key}>')
|
||||
|
||||
return ''.join(xml_parts)
|
||||
|
||||
def _escape_xml(self, text: str) -> str:
|
||||
"""Escape special XML characters."""
|
||||
return (text
|
||||
.replace('&', '&')
|
||||
.replace('<', '<')
|
||||
.replace('>', '>')
|
||||
.replace('"', '"')
|
||||
.replace("'", '''))
|
||||
|
||||
def _send_request(self, request_type: str, request_data: Dict = None) -> Dict:
|
||||
"""Send a request to QuickBooks and parse the response."""
|
||||
if not self.is_connected:
|
||||
raise QBDesktopError("Not connected to QuickBooks Desktop")
|
||||
|
||||
request_xml = self._build_qbxml_request(request_type, request_data)
|
||||
|
||||
logger.debug(f"Sending QBXML request: {request_type}")
|
||||
logger.debug(f"Request XML: {request_xml[:500]}...")
|
||||
|
||||
try:
|
||||
response_xml = self.request_processor.ProcessRequest(
|
||||
self.session_ticket,
|
||||
request_xml
|
||||
)
|
||||
|
||||
logger.debug(f"Response XML: {response_xml[:500]}...")
|
||||
|
||||
return self._parse_response(response_xml, request_type)
|
||||
|
||||
except Exception as e:
|
||||
raise QBDesktopError(f"Request failed: {str(e)}")
|
||||
|
||||
def _parse_response(self, response_xml: str, request_type: str) -> Dict:
|
||||
"""Parse QBXML response into a dictionary."""
|
||||
try:
|
||||
root = ET.fromstring(response_xml)
|
||||
|
||||
# Find the response element
|
||||
msgs_rs = root.find(".//QBXMLMsgsRs")
|
||||
if msgs_rs is None:
|
||||
raise QBDesktopError("Invalid QBXML response")
|
||||
|
||||
# Get the specific response
|
||||
response_tag = f"{request_type}Rs"
|
||||
response_elem = msgs_rs.find(response_tag)
|
||||
|
||||
if response_elem is None:
|
||||
raise QBDesktopError(f"No response found for {request_type}")
|
||||
|
||||
# Check status
|
||||
status_code = response_elem.get("statusCode", "0")
|
||||
status_message = response_elem.get("statusMessage", "")
|
||||
|
||||
if status_code != "0":
|
||||
raise QBDesktopError(
|
||||
f"QuickBooks error: {status_message}",
|
||||
code=status_code,
|
||||
detail=status_message
|
||||
)
|
||||
|
||||
# Parse response data
|
||||
return self._xml_to_dict(response_elem)
|
||||
|
||||
except ET.ParseError as e:
|
||||
raise QBDesktopError(f"Failed to parse response XML: {e}")
|
||||
|
||||
def _xml_to_dict(self, element: ET.Element) -> Dict:
|
||||
"""Convert XML element to dictionary."""
|
||||
result = {}
|
||||
|
||||
for child in element:
|
||||
if len(child) > 0:
|
||||
# Has children - recurse
|
||||
child_dict = self._xml_to_dict(child)
|
||||
if child.tag in result:
|
||||
# Multiple elements with same tag - make a list
|
||||
if not isinstance(result[child.tag], list):
|
||||
result[child.tag] = [result[child.tag]]
|
||||
result[child.tag].append(child_dict)
|
||||
else:
|
||||
result[child.tag] = child_dict
|
||||
else:
|
||||
# Leaf node
|
||||
if child.tag in result:
|
||||
if not isinstance(result[child.tag], list):
|
||||
result[child.tag] = [result[child.tag]]
|
||||
result[child.tag].append(child.text)
|
||||
else:
|
||||
result[child.tag] = child.text
|
||||
|
||||
return result
|
||||
|
||||
# ==========================================================================
|
||||
# Query Methods
|
||||
# ==========================================================================
|
||||
|
||||
def get_customers(self, max_results: int = 1000) -> List[Dict]:
|
||||
"""Get all customers."""
|
||||
response = self._send_request("CustomerQuery", {
|
||||
"MaxReturned": str(max_results)
|
||||
})
|
||||
customers = response.get("CustomerRet", [])
|
||||
if isinstance(customers, dict):
|
||||
customers = [customers]
|
||||
return customers
|
||||
|
||||
def get_vendors(self, max_results: int = 1000) -> List[Dict]:
|
||||
"""Get all vendors."""
|
||||
response = self._send_request("VendorQuery", {
|
||||
"MaxReturned": str(max_results)
|
||||
})
|
||||
vendors = response.get("VendorRet", [])
|
||||
if isinstance(vendors, dict):
|
||||
vendors = [vendors]
|
||||
return vendors
|
||||
|
||||
def get_accounts(self, max_results: int = 1000) -> List[Dict]:
|
||||
"""Get all accounts."""
|
||||
response = self._send_request("AccountQuery", {
|
||||
"MaxReturned": str(max_results)
|
||||
})
|
||||
accounts = response.get("AccountRet", [])
|
||||
if isinstance(accounts, dict):
|
||||
accounts = [accounts]
|
||||
return accounts
|
||||
|
||||
def get_items(self, max_results: int = 1000) -> List[Dict]:
|
||||
"""Get all items/products."""
|
||||
response = self._send_request("ItemQuery", {
|
||||
"MaxReturned": str(max_results)
|
||||
})
|
||||
# Items can be different types
|
||||
items = []
|
||||
for item_type in ["ItemServiceRet", "ItemInventoryRet", "ItemNonInventoryRet",
|
||||
"ItemOtherChargeRet", "ItemGroupRet", "ItemDiscountRet"]:
|
||||
type_items = response.get(item_type, [])
|
||||
if isinstance(type_items, dict):
|
||||
type_items = [type_items]
|
||||
items.extend(type_items)
|
||||
return items
|
||||
|
||||
def get_employees(self, max_results: int = 1000) -> List[Dict]:
|
||||
"""Get all employees."""
|
||||
response = self._send_request("EmployeeQuery", {
|
||||
"MaxReturned": str(max_results)
|
||||
})
|
||||
employees = response.get("EmployeeRet", [])
|
||||
if isinstance(employees, dict):
|
||||
employees = [employees]
|
||||
return employees
|
||||
|
||||
# ==========================================================================
|
||||
# Create Methods
|
||||
# ==========================================================================
|
||||
|
||||
def create_check(self, data: Dict) -> Dict:
|
||||
"""Create a check in QuickBooks Desktop."""
|
||||
# Build the check request
|
||||
check_data = {
|
||||
"AccountRef": data.get("AccountRef"),
|
||||
"PayeeEntityRef": data.get("EntityRef"),
|
||||
"TxnDate": data.get("TxnDate"),
|
||||
"RefNumber": data.get("DocNumber"),
|
||||
"Memo": data.get("PrivateNote", ""),
|
||||
}
|
||||
|
||||
# Add expense lines
|
||||
if "Line" in data:
|
||||
expense_lines = []
|
||||
for line in data["Line"]:
|
||||
if line.get("DetailType") == "AccountBasedExpenseLineDetail":
|
||||
detail = line.get("AccountBasedExpenseLineDetail", {})
|
||||
expense_lines.append({
|
||||
"AccountRef": detail.get("AccountRef"),
|
||||
"Amount": line.get("Amount"),
|
||||
"Memo": line.get("Description", "")
|
||||
})
|
||||
if expense_lines:
|
||||
check_data["ExpenseLineAdd"] = expense_lines
|
||||
|
||||
response = self._send_request("CheckAdd", check_data)
|
||||
return response.get("CheckRet", {})
|
||||
|
||||
def create_invoice(self, data: Dict) -> Dict:
|
||||
"""Create an invoice in QuickBooks Desktop."""
|
||||
invoice_data = {
|
||||
"CustomerRef": data.get("CustomerRef"),
|
||||
"TxnDate": data.get("TxnDate"),
|
||||
"RefNumber": data.get("DocNumber"),
|
||||
"DueDate": data.get("DueDate"),
|
||||
"Memo": data.get("PrivateNote", ""),
|
||||
}
|
||||
|
||||
# Add invoice lines
|
||||
if "Line" in data:
|
||||
invoice_lines = []
|
||||
for line in data["Line"]:
|
||||
if line.get("DetailType") == "SalesItemLineDetail":
|
||||
detail = line.get("SalesItemLineDetail", {})
|
||||
invoice_lines.append({
|
||||
"ItemRef": detail.get("ItemRef"),
|
||||
"Quantity": detail.get("Qty"),
|
||||
"Rate": detail.get("UnitPrice"),
|
||||
"Amount": line.get("Amount"),
|
||||
"Desc": line.get("Description", "")
|
||||
})
|
||||
if invoice_lines:
|
||||
invoice_data["InvoiceLineAdd"] = invoice_lines
|
||||
|
||||
response = self._send_request("InvoiceAdd", invoice_data)
|
||||
return response.get("InvoiceRet", {})
|
||||
|
||||
def create_bill(self, data: Dict) -> Dict:
|
||||
"""Create a bill in QuickBooks Desktop."""
|
||||
bill_data = {
|
||||
"VendorRef": data.get("VendorRef"),
|
||||
"TxnDate": data.get("TxnDate"),
|
||||
"RefNumber": data.get("DocNumber"),
|
||||
"DueDate": data.get("DueDate"),
|
||||
"Memo": data.get("PrivateNote", ""),
|
||||
}
|
||||
|
||||
# Add expense lines
|
||||
if "Line" in data:
|
||||
expense_lines = []
|
||||
for line in data["Line"]:
|
||||
if line.get("DetailType") == "AccountBasedExpenseLineDetail":
|
||||
detail = line.get("AccountBasedExpenseLineDetail", {})
|
||||
expense_lines.append({
|
||||
"AccountRef": detail.get("AccountRef"),
|
||||
"Amount": line.get("Amount"),
|
||||
"Memo": line.get("Description", "")
|
||||
})
|
||||
if expense_lines:
|
||||
bill_data["ExpenseLineAdd"] = expense_lines
|
||||
|
||||
response = self._send_request("BillAdd", bill_data)
|
||||
return response.get("BillRet", {})
|
||||
|
||||
def create_customer(self, data: Dict) -> Dict:
|
||||
"""Create a customer in QuickBooks Desktop."""
|
||||
customer_data = {
|
||||
"Name": data.get("DisplayName"),
|
||||
"CompanyName": data.get("CompanyName"),
|
||||
"FirstName": data.get("GivenName"),
|
||||
"LastName": data.get("FamilyName"),
|
||||
"Phone": data.get("PrimaryPhone", {}).get("FreeFormNumber") if isinstance(data.get("PrimaryPhone"), dict) else None,
|
||||
"Email": data.get("PrimaryEmailAddr", {}).get("Address") if isinstance(data.get("PrimaryEmailAddr"), dict) else None,
|
||||
}
|
||||
|
||||
# Add billing address
|
||||
if "BillAddr" in data:
|
||||
addr = data["BillAddr"]
|
||||
customer_data["BillAddress"] = {
|
||||
"Addr1": addr.get("Line1"),
|
||||
"City": addr.get("City"),
|
||||
"State": addr.get("CountrySubDivisionCode"),
|
||||
"PostalCode": addr.get("PostalCode"),
|
||||
"Country": addr.get("Country")
|
||||
}
|
||||
|
||||
response = self._send_request("CustomerAdd", customer_data)
|
||||
return response.get("CustomerRet", {})
|
||||
|
||||
def create_vendor(self, data: Dict) -> Dict:
|
||||
"""Create a vendor in QuickBooks Desktop."""
|
||||
vendor_data = {
|
||||
"Name": data.get("DisplayName"),
|
||||
"CompanyName": data.get("CompanyName"),
|
||||
"FirstName": data.get("GivenName"),
|
||||
"LastName": data.get("FamilyName"),
|
||||
"Phone": data.get("PrimaryPhone", {}).get("FreeFormNumber") if isinstance(data.get("PrimaryPhone"), dict) else None,
|
||||
"Email": data.get("PrimaryEmailAddr", {}).get("Address") if isinstance(data.get("PrimaryEmailAddr"), dict) else None,
|
||||
}
|
||||
|
||||
# Add address
|
||||
if "BillAddr" in data:
|
||||
addr = data["BillAddr"]
|
||||
vendor_data["VendorAddress"] = {
|
||||
"Addr1": addr.get("Line1"),
|
||||
"City": addr.get("City"),
|
||||
"State": addr.get("CountrySubDivisionCode"),
|
||||
"PostalCode": addr.get("PostalCode"),
|
||||
"Country": addr.get("Country")
|
||||
}
|
||||
|
||||
response = self._send_request("VendorAdd", vendor_data)
|
||||
return response.get("VendorRet", {})
|
||||
|
||||
def create_account(self, data: Dict) -> Dict:
|
||||
"""Create an account in QuickBooks Desktop."""
|
||||
# Map QBO account types to QBD account types
|
||||
account_type_map = {
|
||||
"Bank": "Bank",
|
||||
"Accounts Receivable": "AccountsReceivable",
|
||||
"Other Current Asset": "OtherCurrentAsset",
|
||||
"Fixed Asset": "FixedAsset",
|
||||
"Other Asset": "OtherAsset",
|
||||
"Accounts Payable": "AccountsPayable",
|
||||
"Credit Card": "CreditCard",
|
||||
"Other Current Liability": "OtherCurrentLiability",
|
||||
"Long Term Liability": "LongTermLiability",
|
||||
"Equity": "Equity",
|
||||
"Income": "Income",
|
||||
"Cost of Goods Sold": "CostOfGoodsSold",
|
||||
"Expense": "Expense",
|
||||
"Other Income": "OtherIncome",
|
||||
"Other Expense": "OtherExpense"
|
||||
}
|
||||
|
||||
qbo_type = data.get("AccountType", "Expense")
|
||||
qbd_type = account_type_map.get(qbo_type, "Expense")
|
||||
|
||||
account_data = {
|
||||
"Name": data.get("Name"),
|
||||
"AccountType": qbd_type,
|
||||
"Desc": data.get("Description"),
|
||||
"AccountNumber": data.get("AcctNum"),
|
||||
}
|
||||
|
||||
response = self._send_request("AccountAdd", account_data)
|
||||
return response.get("AccountRet", {})
|
||||
|
||||
def __enter__(self):
|
||||
"""Context manager entry."""
|
||||
self.connect()
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
"""Context manager exit."""
|
||||
self.disconnect()
|
||||
return False
|
||||
|
||||
|
||||
def check_qbd_availability() -> Dict[str, Any]:
|
||||
"""Check if QuickBooks Desktop integration is available."""
|
||||
result = {
|
||||
"available": False,
|
||||
"is_windows": IS_WINDOWS,
|
||||
"has_pywin32": HAS_WIN32COM,
|
||||
"qb_running": False,
|
||||
"sdk_installed": False,
|
||||
"error": None,
|
||||
"details": ""
|
||||
}
|
||||
|
||||
if not IS_WINDOWS:
|
||||
result["error"] = "QuickBooks Desktop is only available on Windows"
|
||||
result["details"] = "This feature requires Windows operating system"
|
||||
return result
|
||||
|
||||
if not HAS_WIN32COM:
|
||||
result["error"] = "pywin32 is not installed"
|
||||
result["details"] = "Run: pip install pywin32"
|
||||
return result
|
||||
|
||||
# Try to create the RequestProcessor to check if SDK is installed
|
||||
rp = None
|
||||
try:
|
||||
pythoncom.CoInitialize()
|
||||
|
||||
# Try to create RequestProcessor
|
||||
try:
|
||||
rp = Dispatch("QBXMLRP2.RequestProcessor")
|
||||
result["sdk_installed"] = True
|
||||
result["available"] = True
|
||||
logger.info("QBXMLRP2.RequestProcessor created successfully in availability check")
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
logger.warning(f"Failed to create RequestProcessor: {error_msg}")
|
||||
if "80040154" in error_msg or "Class not registered" in error_msg:
|
||||
result["error"] = "QuickBooks Desktop SDK not installed"
|
||||
result["details"] = (
|
||||
"The QBXMLRP2 component is not registered. "
|
||||
"This usually means QuickBooks Desktop is not installed, "
|
||||
"or the SDK components are missing. "
|
||||
"Try repairing your QuickBooks installation."
|
||||
)
|
||||
else:
|
||||
result["error"] = f"Cannot create QuickBooks interface: {error_msg}"
|
||||
return result
|
||||
|
||||
# Try to open a connection to see if QB is running
|
||||
try:
|
||||
rp.OpenConnection2("", "QBD Availability Check", QBConnectionType.localQBD)
|
||||
result["qb_running"] = True
|
||||
logger.info("OpenConnection2 successful - QuickBooks is running")
|
||||
|
||||
# Close the connection
|
||||
try:
|
||||
rp.CloseConnection()
|
||||
except:
|
||||
pass
|
||||
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
logger.info(f"OpenConnection2 check result: {error_msg}")
|
||||
|
||||
# QB SDK is available but QB might not be running
|
||||
if "80040408" in error_msg:
|
||||
result["qb_running"] = False
|
||||
result["error"] = "QuickBooks Desktop is not running"
|
||||
result["details"] = "Please open QuickBooks Desktop with a company file"
|
||||
elif "80040402" in error_msg:
|
||||
# This can happen when QB is running but locked
|
||||
result["qb_running"] = True
|
||||
result["error"] = "QuickBooks is busy or locked"
|
||||
result["details"] = "QuickBooks may be showing a dialog. Please check QuickBooks."
|
||||
else:
|
||||
# Unknown error but SDK is there
|
||||
result["qb_running"] = False
|
||||
result["error"] = f"QuickBooks not accessible: {error_msg}"
|
||||
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
logger.error(f"Unexpected error in availability check: {error_msg}")
|
||||
result["error"] = f"Error checking QuickBooks: {error_msg}"
|
||||
finally:
|
||||
try:
|
||||
pythoncom.CoUninitialize()
|
||||
except:
|
||||
pass
|
||||
|
||||
return result
|
||||
Reference in New Issue
Block a user