From dedb2db026e548e3083c7450aa073b1791935257 Mon Sep 17 00:00:00 2001 From: NguyenND Date: Mon, 16 Feb 2026 12:15:16 -0500 Subject: [PATCH] Update with QB Desktop connection --- app.py | 635 ++++++++++++++++++++++-- requirements.txt | 1 + src/api/qbd_client.py | 758 +++++++++++++++++++++++++++++ templates/connection.html | 989 ++++++++++++++++++++++++++++---------- test_qbd_connection.py | 188 ++++++++ 5 files changed, 2256 insertions(+), 315 deletions(-) create mode 100644 src/api/qbd_client.py create mode 100644 test_qbd_connection.py diff --git a/app.py b/app.py index 7d36a57..edc1c38 100644 --- a/app.py +++ b/app.py @@ -19,6 +19,7 @@ from werkzeug.utils import secure_filename from src.config.settings import Settings, QBOCredentials, MappingTemplate, FieldMapping from src.api.qbo_client import QBOClient +from src.api.qbd_client import QBDesktopClient, QBDCredentials, QBDesktopError, check_qbd_availability from src.core.excel_parser import ExcelParser, ParseResult from src.core.import_processor import ImportProcessor, ImportResult, ImportStatus from dataclasses import asdict @@ -60,8 +61,17 @@ def allowed_file(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS +def get_connection_type(): + """Get the current connection type (qbo or qbd).""" + return session.get('connection_type', 'qbo') + + def get_qbo_client(): """Get or create QBO client from session or saved credentials.""" + # Check if using QBD instead + if get_connection_type() == 'qbd': + return None + # First try session creds_data = session.get('qbo_credentials') if creds_data: @@ -88,24 +98,89 @@ def get_qbo_client(): return None +def get_qbd_client(): + """Get QBD client from session.""" + if get_connection_type() != 'qbd': + return None + + qbd_data = session.get('qbd_connection') + if qbd_data and qbd_data.get('connected'): + # Create new client with saved credentials + creds = QBDCredentials( + application_name=qbd_data.get('application_name', 'QBO Excel Sync'), + company_file=qbd_data.get('company_file', '') + ) + client = QBDesktopClient(creds) + # We need to reconnect each request since COM objects don't persist + try: + client.connect() + return client + except Exception as e: + logger.error(f"Failed to reconnect to QBD: {e}") + return None + return None + + +def get_active_client(): + """Get the active QB client (either QBO or QBD).""" + conn_type = get_connection_type() + if conn_type == 'qbd': + return get_qbd_client() + else: + return get_qbo_client() + + @app.context_processor def inject_common_variables(): """Inject common variables into all templates.""" - client = get_qbo_client() - is_connected = client.is_authenticated if client else False + # Get connection type directly from session + connection_type = session.get('connection_type', 'qbo') + is_connected = False company_name = "" - if is_connected: + if connection_type == 'qbd': + # Get QBD connection data directly + qbd_data = session.get('qbd_connection') + if qbd_data: + is_connected = qbd_data.get('connected', False) + company_name = qbd_data.get('company_name', '') + else: + client = get_qbo_client() + is_connected = client.is_authenticated if client else False + + if is_connected: + try: + company_info = client.get_company_info() + company_name = company_info.get('CompanyName', '') + except Exception: + pass + + # For QBD availability - just check if we're on Windows and have pywin32 + qbd_available = False + qbd_running = False + + import sys + if sys.platform == 'win32': try: - company_info = client.get_company_info() - company_name = company_info.get('CompanyName', '') - except Exception: + import pythoncom + from win32com.client import Dispatch + qbd_available = True + # If we're connected, QB is obviously running + if connection_type == 'qbd' and is_connected: + qbd_running = True + else: + # Assume running if available - actual check happens via API + qbd_running = True + except ImportError: pass return { 'is_connected': is_connected, 'company_name': company_name, - 'environment': settings.qbo_environment + 'environment': settings.qbo_environment, + 'connection_type': connection_type, + 'qbd_available': qbd_available, + 'qbd_running': qbd_running } @@ -137,20 +212,11 @@ def log_action(action: str, details: str = "", level: str = "info"): def index(): """Home page / Dashboard.""" log_action("PAGE_VIEW", "Dashboard") - client = get_qbo_client() - is_connected = client.is_authenticated if client else False - company_name = "" - if is_connected: - try: - company_info = client.get_company_info() - company_name = company_info.get('CompanyName', '') - except Exception as e: - logger.warning(f"Failed to get company info: {e}") + # Don't pass is_connected and company_name - let the context processor handle it + # This allows both QBO and QBD connection status to be shown correctly return render_template('index.html', - is_connected=is_connected, - company_name=company_name, environment=settings.qbo_environment) @@ -159,21 +225,12 @@ def connection(): """Connection management page.""" log_action("PAGE_VIEW", "Connection") creds = settings.get_credentials() - client = get_qbo_client() - is_connected = client.is_authenticated if client else False - company_name = "" - if is_connected: - try: - company_info = client.get_company_info() - company_name = company_info.get('CompanyName', '') - except Exception: - pass + # Don't pass is_connected and company_name - let the context processor handle it + # This allows QBD connection status to be shown correctly return render_template('connection.html', credentials=creds, - is_connected=is_connected, - company_name=company_name, environment=settings.qbo_environment) @@ -181,14 +238,12 @@ def connection(): def import_page(): """Import page.""" log_action("PAGE_VIEW", "Import") - client = get_qbo_client() - is_connected = client.is_authenticated if client else False # Get available templates templates = settings.get_templates() + # Don't pass is_connected - let the context processor handle it return render_template('import.html', - is_connected=is_connected, templates=templates, environment=settings.qbo_environment) @@ -391,30 +446,512 @@ def disconnect(): return jsonify({'success': False, 'error': str(e)}), 400 -@app.route('/api/connection/test', methods=['GET']) -def test_connection(): - """Test QBO connection.""" +@app.route('/api/session/debug', methods=['GET']) +def session_debug(): + """Debug endpoint to check session contents.""" + # Get session ID if using filesystem sessions + session_id = None try: - client = get_qbo_client() + from flask import request + session_id = request.cookies.get('session') + except: + pass + + return jsonify({ + 'session_id': session_id, + 'connection_type': session.get('connection_type', 'not set'), + 'qbd_connection': session.get('qbd_connection', 'not set'), + 'qbo_credentials': 'present' if session.get('qbo_credentials') else 'not set', + 'session_keys': list(session.keys()) + }) + + +# ============================================================================= +# Routes - QuickBooks Desktop Connection +# ============================================================================= + +@app.route('/api/qbd/status', methods=['GET']) +def qbd_status(): + """Check QuickBooks Desktop availability and status.""" + try: + status = check_qbd_availability() - if not client or not client.is_authenticated: - return jsonify({'success': False, 'connected': False, 'error': 'Not connected'}) + # Also check current connection + qbd_data = session.get('qbd_connection', {}) + status['connected'] = qbd_data.get('connected', False) + status['company_name'] = qbd_data.get('company_name', '') + status['connection_type'] = get_connection_type() - company_info = client.get_company_info() - customers = client.get_customers() - vendors = client.get_vendors() - accounts = client.get_accounts() + return jsonify({'success': True, **status}) + except Exception as e: + return jsonify({'success': False, 'error': str(e)}), 400 + + +@app.route('/api/qbd/connect', methods=['POST']) +def qbd_connect(): + """Connect to QuickBooks Desktop using subprocess to avoid COM threading issues.""" + import subprocess + import json + + try: + log_action("CREATE", "Attempting to connect to QuickBooks Desktop (via subprocess)") - log_action("CREATE", f"Connection test successful: {len(customers)} customers, {len(vendors)} vendors, {len(accounts)} accounts") + # Create a small Python script to run the connection in a separate process + # This avoids all Flask threading/COM issues + connect_script = ''' +import sys +import json + +try: + import pythoncom + from win32com.client import Dispatch + + pythoncom.CoInitialize() + + rp = Dispatch("QBXMLRP2.RequestProcessor") + rp.OpenConnection2("", "QBO Excel Sync", 1) + ticket = rp.BeginSession("", 0) + + # Query company name + request = """ + + + + + +""" + + response = rp.ProcessRequest(ticket, request) + + # Parse company name + import xml.etree.ElementTree as ET + root = ET.fromstring(response) + company_elem = root.find(".//CompanyName") + company_name = company_elem.text if company_elem is not None else "QuickBooks Desktop" + + rp.EndSession(ticket) + rp.CloseConnection() + pythoncom.CoUninitialize() + + print(json.dumps({"success": True, "company_name": company_name})) + +except Exception as e: + print(json.dumps({"success": False, "error": str(e)})) + sys.exit(1) +''' + + # Run the script in a separate Python process + result = subprocess.run( + [sys.executable, '-c', connect_script], + capture_output=True, + text=True, + timeout=60 # 60 second timeout + ) + + logger.info(f"Subprocess stdout: {result.stdout}") + logger.info(f"Subprocess stderr: {result.stderr}") + + if result.returncode != 0: + error_msg = result.stderr or result.stdout or "Unknown error" + log_action("CREATE", f"QBD subprocess failed: {error_msg}", "error") + return jsonify({ + 'success': False, + 'error': f"Connection failed: {error_msg}" + }), 400 + + # Parse the JSON output + try: + output = json.loads(result.stdout.strip()) + except json.JSONDecodeError: + log_action("CREATE", f"QBD subprocess invalid output: {result.stdout}", "error") + return jsonify({ + 'success': False, + 'error': f"Invalid response from connection script: {result.stdout}" + }), 400 + + if not output.get('success'): + error_msg = output.get('error', 'Unknown error') + log_action("CREATE", f"QBD connection failed: {error_msg}", "error") + return jsonify({ + 'success': False, + 'error': error_msg + }), 400 + + company_name = output.get('company_name', 'QuickBooks Desktop') + + # Store connection info in session + session['connection_type'] = 'qbd' + session['qbd_connection'] = { + 'connected': True, + 'company_name': company_name, + 'company_file': '', + 'application_name': 'QBO Excel Sync', + 'connected_at': datetime.now().isoformat() + } + session.modified = True # Force session to be saved + + log_action("CREATE", f"Connected to QuickBooks Desktop: {company_name}") + + return jsonify({ + 'success': True, + 'company_name': company_name, + 'message': 'Connected to QuickBooks Desktop' + }) + + except subprocess.TimeoutExpired: + log_action("CREATE", "QBD connection timed out after 60 seconds", "error") + return jsonify({ + 'success': False, + 'error': "Connection timed out. QuickBooks may be waiting for user input or not responding." + }), 400 + except Exception as e: + import traceback + error_trace = traceback.format_exc() + logger.error(f"QBD connection failed: {str(e)}\nTraceback: {error_trace}") + log_action("CREATE", f"QBD connection failed: {str(e)}", "error") + return jsonify({ + 'success': False, + 'error': str(e) + }), 400 + + +@app.route('/api/qbd/debug', methods=['GET']) +def qbd_debug(): + """Debug endpoint to check QBD availability with detailed info.""" + import sys + + result = { + 'platform': sys.platform, + 'is_windows': sys.platform == 'win32', + 'python_version': sys.version, + } + + # Check pywin32 + try: + import pythoncom + from win32com.client import Dispatch + result['pywin32_installed'] = True + result['pythoncom_available'] = True + except ImportError as e: + result['pywin32_installed'] = False + result['pywin32_error'] = str(e) + return jsonify(result) + + # Try to create RequestProcessor + try: + pythoncom.CoInitialize() + result['com_initialized'] = True + except Exception as e: + result['com_initialized'] = False + result['com_error'] = str(e) + return jsonify(result) + + try: + rp = Dispatch("QBXMLRP2.RequestProcessor") + result['request_processor_created'] = True + except Exception as e: + result['request_processor_created'] = False + result['request_processor_error'] = str(e) + try: + pythoncom.CoUninitialize() + except: + pass + return jsonify(result) + + # Try OpenConnection2 + try: + rp.OpenConnection2("", "QBD Debug Test", 1) # 1 = localQBD + result['connection_opened'] = True + except Exception as e: + result['connection_opened'] = False + result['connection_error'] = str(e) + try: + pythoncom.CoUninitialize() + except: + pass + return jsonify(result) + + # Try BeginSession + try: + ticket = rp.BeginSession("", 0) # 0 = qbFileOpenDoNotCare + result['session_started'] = True + result['session_ticket'] = ticket[:20] + "..." if ticket else None + + # End session + rp.EndSession(ticket) + result['session_ended'] = True + except Exception as e: + result['session_started'] = False + result['session_error'] = str(e) + + # Close connection + try: + rp.CloseConnection() + result['connection_closed'] = True + except Exception as e: + result['connection_closed'] = False + result['close_error'] = str(e) + + try: + pythoncom.CoUninitialize() + except: + pass + + return jsonify(result) + + +@app.route('/api/qbd/disconnect', methods=['POST']) +def qbd_disconnect(): + """Disconnect from QuickBooks Desktop.""" + try: + log_action("DELETE", "Disconnecting from QuickBooks Desktop") + + # Clear QBD session data + session.pop('qbd_connection', None) + session['connection_type'] = 'qbo' # Reset to QBO mode + + logger.info("Disconnected from QuickBooks Desktop") + return jsonify({'success': True}) + except Exception as e: + log_action("DELETE", f"QBD disconnect failed: {str(e)}", "error") + return jsonify({'success': False, 'error': str(e)}), 400 + + +@app.route('/api/qbd/test', methods=['GET']) +def qbd_test(): + """Test QuickBooks Desktop connection using subprocess.""" + import subprocess + import json as json_module + + try: + qbd_data = session.get('qbd_connection', {}) + if not qbd_data.get('connected'): + return jsonify({ + 'success': False, + 'connected': False, + 'error': 'Not connected to QuickBooks Desktop' + }) + + # Run test in subprocess + test_script = ''' +import sys +import json + +try: + import pythoncom + from win32com.client import Dispatch + + pythoncom.CoInitialize() + + rp = Dispatch("QBXMLRP2.RequestProcessor") + rp.OpenConnection2("", "QBO Excel Sync", 1) + ticket = rp.BeginSession("", 0) + + # Query company + company_request = """ + + + + + +""" + + response = rp.ProcessRequest(ticket, company_request) + + import xml.etree.ElementTree as ET + root = ET.fromstring(response) + company_elem = root.find(".//CompanyName") + company_name = company_elem.text if company_elem is not None else "QuickBooks Desktop" + + # Query customers count + customer_request = """ + + + + 1000 + +""" + response = rp.ProcessRequest(ticket, customer_request) + root = ET.fromstring(response) + customers = root.findall(".//CustomerRet") + + # Query vendors count + vendor_request = """ + + + + 1000 + +""" + response = rp.ProcessRequest(ticket, vendor_request) + root = ET.fromstring(response) + vendors = root.findall(".//VendorRet") + + # Query accounts count + account_request = """ + + + + 1000 + +""" + response = rp.ProcessRequest(ticket, account_request) + root = ET.fromstring(response) + accounts = root.findall(".//AccountRet") + + rp.EndSession(ticket) + rp.CloseConnection() + pythoncom.CoUninitialize() + + print(json.dumps({ + "success": True, + "company_name": company_name, + "customers_count": len(customers), + "vendors_count": len(vendors), + "accounts_count": len(accounts) + })) + +except Exception as e: + print(json.dumps({"success": False, "error": str(e)})) + sys.exit(1) +''' + + result = subprocess.run( + [sys.executable, '-c', test_script], + capture_output=True, + text=True, + timeout=60 + ) + + if result.returncode != 0: + session.pop('qbd_connection', None) + return jsonify({ + 'success': False, + 'connected': False, + 'error': result.stderr or result.stdout or "Test failed" + }) + + output = json_module.loads(result.stdout.strip()) + + if not output.get('success'): + session.pop('qbd_connection', None) + return jsonify({ + 'success': False, + 'connected': False, + 'error': output.get('error', 'Unknown error') + }) + + # Update session + session['qbd_connection']['company_name'] = output.get('company_name') + + log_action("CREATE", f"QBD test successful: {output.get('customers_count')} customers") return jsonify({ 'success': True, 'connected': True, - 'company_name': company_info.get('CompanyName', 'Unknown'), - 'customers_count': len(customers), - 'vendors_count': len(vendors), - 'accounts_count': len(accounts) + 'company_name': output.get('company_name'), + 'customers_count': output.get('customers_count', 0), + 'vendors_count': output.get('vendors_count', 0), + 'accounts_count': output.get('accounts_count', 0) }) + + except subprocess.TimeoutExpired: + session.pop('qbd_connection', None) + return jsonify({ + 'success': False, + 'connected': False, + 'error': 'Test timed out' + }) + except Exception as e: + return jsonify({ + 'success': False, + 'connected': False, + 'error': str(e) + }) + + +@app.route('/api/connection/switch', methods=['POST']) +def switch_connection(): + """Switch between QBO and QBD connection modes.""" + try: + data = request.json or {} + new_type = data.get('type', 'qbo') + + if new_type not in ['qbo', 'qbd']: + return jsonify({'success': False, 'error': 'Invalid connection type'}), 400 + + session['connection_type'] = new_type + log_action("EDIT", f"Switched connection type to {new_type.upper()}") + + return jsonify({ + 'success': True, + 'connection_type': new_type + }) + except Exception as e: + return jsonify({'success': False, 'error': str(e)}), 400 + + +@app.route('/api/connection/test', methods=['GET']) +def test_connection(): + """Test QB connection (QBO or QBD based on current mode).""" + try: + connection_type = get_connection_type() + + if connection_type == 'qbd': + # Test QBD connection + qbd_data = session.get('qbd_connection', {}) + if not qbd_data.get('connected'): + return jsonify({'success': False, 'connected': False, 'error': 'Not connected'}) + + creds = QBDCredentials( + application_name=qbd_data.get('application_name', 'QBO Excel Sync'), + company_file=qbd_data.get('company_file', '') + ) + + client = QBDesktopClient(creds) + client.connect() + + customers = client.get_customers() + vendors = client.get_vendors() + accounts = client.get_accounts() + company_name = client.company_name + + client.disconnect() + + log_action("CREATE", f"QBD connection test successful: {len(customers)} customers") + + return jsonify({ + 'success': True, + 'connected': True, + 'connection_type': 'qbd', + 'company_name': company_name, + 'customers_count': len(customers), + 'vendors_count': len(vendors), + 'accounts_count': len(accounts) + }) + else: + # Test QBO connection + client = get_qbo_client() + + if not client or not client.is_authenticated: + return jsonify({'success': False, 'connected': False, 'error': 'Not connected'}) + + company_info = client.get_company_info() + customers = client.get_customers() + vendors = client.get_vendors() + accounts = client.get_accounts() + + log_action("CREATE", f"QBO connection test successful: {len(customers)} customers") + + return jsonify({ + 'success': True, + 'connected': True, + 'connection_type': 'qbo', + 'company_name': company_info.get('CompanyName', 'Unknown'), + 'customers_count': len(customers), + 'vendors_count': len(vendors), + 'accounts_count': len(accounts) + }) except Exception as e: log_action("CREATE", f"Connection test failed: {str(e)}", "error") return jsonify({'success': False, 'connected': False, 'error': str(e)}) @@ -1456,8 +1993,10 @@ if __name__ == '__main__': UPLOAD_FOLDER.mkdir(exist_ok=True) # Run the app + # NOTE: threaded=False is important for COM/QuickBooks Desktop compatibility + # COM objects don't work well with Flask's default multi-threaded mode debug_mode = os.environ.get('FLASK_DEBUG', 'true').lower() == 'true' port = int(os.environ.get('PORT', 9000)) logger.info(f"Starting QBO Excel Sync Web App on port {port} (debug={debug_mode})") - app.run(host='0.0.0.0', port=port, debug=debug_mode) \ No newline at end of file + app.run(host='0.0.0.0', port=port, debug=debug_mode, threaded=False) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index c1fcbbb..8b83aec 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ Flask>=2.3.0 Flask-Session>=0.5.0 openpyxl>=3.1.0 requests>=2.31.0 +pywin32>=306; sys_platform == 'win32' \ No newline at end of file diff --git a/src/api/qbd_client.py b/src/api/qbd_client.py new file mode 100644 index 0000000..91c3c48 --- /dev/null +++ b/src/api/qbd_client.py @@ -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 = [ + '', + ''.format(self.QBXML_VERSION), + '', + '' + ] + + # 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'') + 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'') + else: + xml_parts.append(self._dict_to_xml(request_data)) + + xml_parts.append(f'') + xml_parts.append('') + xml_parts.append('') + + 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'') + 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'') + else: + xml_parts.append(f'<{key}>{self._escape_xml(str(item))}') + else: + xml_parts.append(f'<{key}>{self._escape_xml(str(value))}') + + 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 \ No newline at end of file diff --git a/templates/connection.html b/templates/connection.html index f411f09..70a10ff 100644 --- a/templates/connection.html +++ b/templates/connection.html @@ -6,307 +6,647 @@

QuickBooks Connection

-

Connect to your QuickBooks Online account to start importing data.

+

Connect to QuickBooks Online or QuickBooks Desktop to start importing data.

- -
-
-

API Credentials

- - - - - + +
+
+

Connection Type

-
-
-
- - -
- -
- - -
- -
- - - This must match exactly in your Intuit Developer app settings. -
- -
- - -
- - -
-
-
- - -
-
-

Connection Status

- - - - - -
-
-
-
- {% if is_connected %} - - - - - {% else %} - - - - - - {% endif %} -
-
-

{% if is_connected %}Connected{% else %}Not Connected{% endif %}

- {% if is_connected and company_name %} -

{{ company_name }}

- {% endif %} -
-
- -
- {% if is_connected %} - - - {% else %} - - {% endif %} +
+
+ +
- - + + +
+
+

QuickBooks Online Status

+
+
+
+
+ {% if is_connected and connection_type != 'qbd' %} + + + + + {% else %} + + + + + + {% endif %} +
+
+

{% if is_connected and connection_type != 'qbd' %}Connected{% else %}Not Connected{% endif %}

+ {% if is_connected and company_name and connection_type != 'qbd' %} +

{{ company_name }}

+ {% endif %} +
-
- - +
+ {% if is_connected and connection_type != 'qbd' %} + + + {% else %} + + {% endif %}
+
+
-
- - + + +
+ + +
+
+
+

QuickBooks Desktop Connection

+
+
+ {% if not qbd_available %} +
+ + + + - Save Tokens & Connect - - -
-
+
+ QuickBooks Desktop not available +

QuickBooks Desktop integration requires:

+
    +
  • Windows operating system
    • +
  • QuickBooks Desktop installed and running
    • +
  • pywin32 Python package (pip install pywin32)
    • +
+
+
+ {% else %} +
+
+ Status: + + {% if qbd_running %} + QuickBooks is running + {% else %} + QuickBooks not detected + {% endif %} + +
+
- - +
+
{% endblock %} {% block extra_js %}