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 @@