2002 lines
75 KiB
Python
2002 lines
75 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
QBO Excel Sync Web Application
|
|
A Flask-based web application for mapping and importing Excel data to QuickBooks Online.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
# Add src to path
|
|
sys.path.insert(0, str(Path(__file__).parent / "src"))
|
|
|
|
from flask import Flask, render_template, request, jsonify, session, redirect, url_for, send_file
|
|
from flask_session import Session
|
|
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
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler('app.log'),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.environ.get('SECRET_KEY', 'qbo-excel-sync-secret-key-change-in-production')
|
|
|
|
# Configure session
|
|
app.config['SESSION_TYPE'] = 'filesystem'
|
|
app.config['SESSION_FILE_DIR'] = '.flask_session'
|
|
app.config['SESSION_PERMANENT'] = False
|
|
Session(app)
|
|
|
|
# Configure upload folder
|
|
UPLOAD_FOLDER = Path(__file__).parent / 'uploads'
|
|
UPLOAD_FOLDER.mkdir(exist_ok=True)
|
|
app.config['UPLOAD_FOLDER'] = str(UPLOAD_FOLDER)
|
|
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB max file size
|
|
|
|
ALLOWED_EXTENSIONS = {'xlsx', 'xls'}
|
|
|
|
# Initialize settings
|
|
settings = Settings()
|
|
|
|
|
|
def allowed_file(filename):
|
|
"""Check if file extension is allowed."""
|
|
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:
|
|
creds = QBOCredentials(**creds_data)
|
|
if creds.access_token:
|
|
return QBOClient(creds)
|
|
|
|
# Fall back to saved credentials from settings
|
|
creds = settings.get_credentials()
|
|
if creds and creds.access_token:
|
|
# Also store in session for future requests
|
|
session['qbo_credentials'] = {
|
|
'client_id': creds.client_id,
|
|
'client_secret': creds.client_secret,
|
|
'redirect_uri': creds.redirect_uri,
|
|
'environment': creds.environment,
|
|
'access_token': creds.access_token,
|
|
'refresh_token': creds.refresh_token,
|
|
'realm_id': creds.realm_id,
|
|
'token_expiry': creds.token_expiry
|
|
}
|
|
return QBOClient(creds)
|
|
|
|
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."""
|
|
# Get connection type directly from session
|
|
connection_type = session.get('connection_type', 'qbo')
|
|
is_connected = False
|
|
company_name = ""
|
|
|
|
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:
|
|
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,
|
|
'connection_type': connection_type,
|
|
'qbd_available': qbd_available,
|
|
'qbd_running': qbd_running
|
|
}
|
|
|
|
|
|
def log_action(action: str, details: str = "", level: str = "info"):
|
|
"""Log user actions for audit trail."""
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
# Handle case when called outside request context (e.g., background threads)
|
|
try:
|
|
user_ip = request.remote_addr
|
|
except RuntimeError:
|
|
user_ip = "background"
|
|
|
|
log_message = f"[{timestamp}] [{user_ip}] [{action}] {details}"
|
|
|
|
if level == "error":
|
|
logger.error(log_message)
|
|
elif level == "warning":
|
|
logger.warning(log_message)
|
|
else:
|
|
logger.info(log_message)
|
|
|
|
|
|
# =============================================================================
|
|
# Routes - Pages
|
|
# =============================================================================
|
|
|
|
@app.route('/')
|
|
def index():
|
|
"""Home page / Dashboard."""
|
|
log_action("PAGE_VIEW", "Dashboard")
|
|
|
|
# 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',
|
|
environment=settings.qbo_environment)
|
|
|
|
|
|
@app.route('/connection')
|
|
def connection():
|
|
"""Connection management page."""
|
|
log_action("PAGE_VIEW", "Connection")
|
|
creds = settings.get_credentials()
|
|
|
|
# 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,
|
|
environment=settings.qbo_environment)
|
|
|
|
|
|
@app.route('/import')
|
|
def import_page():
|
|
"""Import page."""
|
|
log_action("PAGE_VIEW", "Import")
|
|
|
|
# Get available templates
|
|
templates = settings.get_templates()
|
|
|
|
# Don't pass is_connected - let the context processor handle it
|
|
return render_template('import.html',
|
|
templates=templates,
|
|
environment=settings.qbo_environment)
|
|
|
|
|
|
@app.route('/templates')
|
|
def templates_page():
|
|
"""Mapping templates page."""
|
|
log_action("PAGE_VIEW", "Templates")
|
|
templates = settings.get_templates()
|
|
default_templates = settings.get_default_templates()
|
|
|
|
return render_template('templates.html',
|
|
templates=templates,
|
|
default_templates=default_templates)
|
|
|
|
|
|
@app.route('/settings')
|
|
def settings_page():
|
|
"""Settings page."""
|
|
log_action("PAGE_VIEW", "Settings")
|
|
import_settings = settings.import_settings
|
|
|
|
return render_template('settings.html',
|
|
import_settings=import_settings,
|
|
environment=settings.qbo_environment)
|
|
|
|
|
|
# =============================================================================
|
|
# Routes - API Endpoints
|
|
# =============================================================================
|
|
|
|
@app.route('/api/credentials', methods=['POST'])
|
|
def save_credentials():
|
|
"""Save QBO credentials."""
|
|
try:
|
|
data = request.json
|
|
log_action("CREATE", f"Saving QBO credentials for environment: {data.get('environment', 'sandbox')}")
|
|
|
|
creds = QBOCredentials(
|
|
client_id=data.get('client_id', ''),
|
|
client_secret=data.get('client_secret', ''),
|
|
redirect_uri=data.get('redirect_uri', 'http://localhost:5000/callback'),
|
|
environment=data.get('environment', 'sandbox')
|
|
)
|
|
|
|
# Preserve existing tokens if any
|
|
existing = settings.get_credentials()
|
|
if existing.access_token:
|
|
creds.access_token = existing.access_token
|
|
creds.refresh_token = existing.refresh_token
|
|
creds.realm_id = existing.realm_id
|
|
creds.token_expiry = existing.token_expiry
|
|
|
|
settings.save_credentials(creds)
|
|
|
|
logger.info(f"Credentials saved successfully for environment: {creds.environment}")
|
|
return jsonify({'success': True, 'message': 'Credentials saved successfully'})
|
|
except Exception as e:
|
|
log_action("CREATE", f"Failed to save credentials: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/oauth/start', methods=['POST'])
|
|
def start_oauth():
|
|
"""Start OAuth flow - return authorization URL."""
|
|
try:
|
|
creds = settings.get_credentials()
|
|
|
|
if not creds.client_id or not creds.client_secret:
|
|
return jsonify({'success': False, 'error': 'Please save your credentials first'}), 400
|
|
|
|
client = QBOClient(creds)
|
|
auth_url = client.get_authorization_url(state='web_auth')
|
|
|
|
log_action("CREATE", "OAuth flow initiated")
|
|
return jsonify({'success': True, 'auth_url': auth_url})
|
|
except Exception as e:
|
|
log_action("CREATE", f"Failed to start OAuth: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/oauth/callback')
|
|
@app.route('/callback')
|
|
def oauth_callback():
|
|
"""Handle OAuth callback."""
|
|
code = request.args.get('code')
|
|
realm_id = request.args.get('realmId')
|
|
error = request.args.get('error')
|
|
|
|
if error:
|
|
log_action("CREATE", f"OAuth callback error: {error}", "error")
|
|
return render_template('oauth_result.html', success=False, error=error)
|
|
|
|
if not code:
|
|
log_action("CREATE", "OAuth callback missing authorization code", "error")
|
|
return render_template('oauth_result.html', success=False, error='Missing authorization code')
|
|
|
|
try:
|
|
creds = settings.get_credentials()
|
|
creds.realm_id = realm_id
|
|
|
|
client = QBOClient(creds)
|
|
|
|
# Exchange code for tokens
|
|
if client._exchange_code_for_tokens(code):
|
|
# Save updated credentials
|
|
settings.save_credentials(client.credentials)
|
|
|
|
# Store in session
|
|
session['qbo_credentials'] = {
|
|
'client_id': client.credentials.client_id,
|
|
'client_secret': client.credentials.client_secret,
|
|
'redirect_uri': client.credentials.redirect_uri,
|
|
'environment': client.credentials.environment,
|
|
'access_token': client.credentials.access_token,
|
|
'refresh_token': client.credentials.refresh_token,
|
|
'realm_id': client.credentials.realm_id,
|
|
'token_expiry': client.credentials.token_expiry
|
|
}
|
|
|
|
log_action("CREATE", f"OAuth successful for realm: {realm_id}")
|
|
return render_template('oauth_result.html', success=True)
|
|
else:
|
|
log_action("CREATE", "Failed to exchange authorization code", "error")
|
|
return render_template('oauth_result.html', success=False, error='Failed to exchange authorization code')
|
|
except Exception as e:
|
|
log_action("CREATE", f"OAuth callback exception: {str(e)}", "error")
|
|
return render_template('oauth_result.html', success=False, error=str(e))
|
|
|
|
|
|
@app.route('/api/oauth/manual', methods=['POST'])
|
|
def manual_oauth():
|
|
"""Save manually entered OAuth tokens."""
|
|
try:
|
|
data = request.json
|
|
log_action("CREATE", "Manual OAuth token entry")
|
|
|
|
creds = settings.get_credentials()
|
|
creds.access_token = data.get('access_token', '')
|
|
creds.refresh_token = data.get('refresh_token', '')
|
|
creds.realm_id = data.get('realm_id', '')
|
|
|
|
# Set token expiry to 1 hour from now
|
|
from datetime import timedelta
|
|
creds.token_expiry = (datetime.now() + timedelta(hours=1)).isoformat()
|
|
|
|
settings.save_credentials(creds)
|
|
|
|
# Store in session
|
|
session['qbo_credentials'] = {
|
|
'client_id': creds.client_id,
|
|
'client_secret': creds.client_secret,
|
|
'redirect_uri': creds.redirect_uri,
|
|
'environment': creds.environment,
|
|
'access_token': creds.access_token,
|
|
'refresh_token': creds.refresh_token,
|
|
'realm_id': creds.realm_id,
|
|
'token_expiry': creds.token_expiry
|
|
}
|
|
|
|
# Test connection
|
|
client = QBOClient(creds)
|
|
company_info = client.get_company_info()
|
|
|
|
logger.info(f"Manual OAuth successful for company: {company_info.get('CompanyName', 'Unknown')}")
|
|
return jsonify({
|
|
'success': True,
|
|
'company_name': company_info.get('CompanyName', 'Unknown')
|
|
})
|
|
except Exception as e:
|
|
log_action("CREATE", f"Manual OAuth failed: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/disconnect', methods=['POST'])
|
|
def disconnect():
|
|
"""Disconnect from QBO."""
|
|
try:
|
|
log_action("DELETE", "Disconnecting from QBO")
|
|
|
|
client = get_qbo_client()
|
|
if client:
|
|
client.disconnect()
|
|
|
|
# Clear credentials
|
|
creds = settings.get_credentials()
|
|
creds.access_token = None
|
|
creds.refresh_token = None
|
|
creds.realm_id = None
|
|
creds.token_expiry = None
|
|
settings.save_credentials(creds)
|
|
|
|
# Clear session
|
|
session.pop('qbo_credentials', None)
|
|
|
|
logger.info("Disconnected from QBO successfully")
|
|
return jsonify({'success': True})
|
|
except Exception as e:
|
|
log_action("DELETE", f"Disconnect failed: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@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:
|
|
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()
|
|
|
|
# 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()
|
|
|
|
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)")
|
|
|
|
# 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 = """<?xml version="1.0" encoding="utf-8"?>
|
|
<?qbxml version="13.0"?>
|
|
<QBXML>
|
|
<QBXMLMsgsRq onError="stopOnError">
|
|
<CompanyQueryRq></CompanyQueryRq>
|
|
</QBXMLMsgsRq>
|
|
</QBXML>"""
|
|
|
|
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 = """<?xml version="1.0" encoding="utf-8"?>
|
|
<?qbxml version="13.0"?>
|
|
<QBXML>
|
|
<QBXMLMsgsRq onError="stopOnError">
|
|
<CompanyQueryRq></CompanyQueryRq>
|
|
</QBXMLMsgsRq>
|
|
</QBXML>"""
|
|
|
|
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 = """<?xml version="1.0" encoding="utf-8"?>
|
|
<?qbxml version="13.0"?>
|
|
<QBXML>
|
|
<QBXMLMsgsRq onError="stopOnError">
|
|
<CustomerQueryRq><MaxReturned>1000</MaxReturned></CustomerQueryRq>
|
|
</QBXMLMsgsRq>
|
|
</QBXML>"""
|
|
response = rp.ProcessRequest(ticket, customer_request)
|
|
root = ET.fromstring(response)
|
|
customers = root.findall(".//CustomerRet")
|
|
|
|
# Query vendors count
|
|
vendor_request = """<?xml version="1.0" encoding="utf-8"?>
|
|
<?qbxml version="13.0"?>
|
|
<QBXML>
|
|
<QBXMLMsgsRq onError="stopOnError">
|
|
<VendorQueryRq><MaxReturned>1000</MaxReturned></VendorQueryRq>
|
|
</QBXMLMsgsRq>
|
|
</QBXML>"""
|
|
response = rp.ProcessRequest(ticket, vendor_request)
|
|
root = ET.fromstring(response)
|
|
vendors = root.findall(".//VendorRet")
|
|
|
|
# Query accounts count
|
|
account_request = """<?xml version="1.0" encoding="utf-8"?>
|
|
<?qbxml version="13.0"?>
|
|
<QBXML>
|
|
<QBXMLMsgsRq onError="stopOnError">
|
|
<AccountQueryRq><MaxReturned>1000</MaxReturned></AccountQueryRq>
|
|
</QBXMLMsgsRq>
|
|
</QBXML>"""
|
|
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': 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)})
|
|
|
|
|
|
@app.route('/api/upload', methods=['POST'])
|
|
def upload_file():
|
|
"""Upload Excel file."""
|
|
try:
|
|
if 'file' not in request.files:
|
|
return jsonify({'success': False, 'error': 'No file provided'}), 400
|
|
|
|
file = request.files['file']
|
|
|
|
if file.filename == '':
|
|
return jsonify({'success': False, 'error': 'No file selected'}), 400
|
|
|
|
if not allowed_file(file.filename):
|
|
return jsonify({'success': False, 'error': 'Invalid file type. Please upload .xlsx or .xls files'}), 400
|
|
|
|
filename = secure_filename(file.filename)
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
unique_filename = f"{timestamp}_{filename}"
|
|
filepath = os.path.join(app.config['UPLOAD_FOLDER'], unique_filename)
|
|
file.save(filepath)
|
|
|
|
# Parse file to get sheets and columns
|
|
parser = ExcelParser()
|
|
sheets = parser.get_sheet_names(filepath)
|
|
columns = parser.get_columns(filepath, sheets[0]) if sheets else []
|
|
|
|
# Preview data
|
|
preview_columns, preview_data = parser.preview(filepath, sheets[0] if sheets else None, max_rows=10)
|
|
|
|
# Store filepath in session
|
|
session['current_file'] = filepath
|
|
session['current_filename'] = filename
|
|
session['excel_columns'] = columns # Store columns for template creation
|
|
|
|
log_action("CREATE", f"File uploaded: {filename} ({len(sheets)} sheets, {len(columns)} columns)")
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'filename': filename,
|
|
'filepath': unique_filename,
|
|
'sheets': sheets,
|
|
'columns': columns,
|
|
'preview': {
|
|
'columns': preview_columns,
|
|
'data': preview_data
|
|
}
|
|
})
|
|
except Exception as e:
|
|
log_action("CREATE", f"File upload failed: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/file/sheets', methods=['GET'])
|
|
def get_sheets():
|
|
"""Get sheets from uploaded file."""
|
|
try:
|
|
filepath = session.get('current_file')
|
|
if not filepath or not os.path.exists(filepath):
|
|
return jsonify({'success': False, 'error': 'No file uploaded'}), 400
|
|
|
|
parser = ExcelParser()
|
|
sheets = parser.get_sheet_names(filepath)
|
|
|
|
return jsonify({'success': True, 'sheets': sheets})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/file/refresh', methods=['POST'])
|
|
def refresh_file():
|
|
"""Refresh the current file content from disk."""
|
|
try:
|
|
filepath = session.get('current_file')
|
|
if not filepath or not os.path.exists(filepath):
|
|
return jsonify({'success': False, 'error': 'No file uploaded or file not found'}), 400
|
|
|
|
log_action("EDIT", f"Refreshing file: {session.get('current_filename', 'Unknown')}")
|
|
|
|
parser = ExcelParser()
|
|
|
|
# Re-read sheets and columns
|
|
sheets = parser.get_sheet_names(filepath)
|
|
columns = parser.get_columns(filepath, sheets[0]) if sheets else []
|
|
|
|
# Update session with new columns
|
|
session['excel_columns'] = columns
|
|
|
|
# Get preview of first sheet
|
|
preview_columns, preview_data = parser.preview(filepath, sheets[0] if sheets else None, max_rows=10)
|
|
|
|
logger.info(f"File refreshed: {len(sheets)} sheets, {len(columns)} columns")
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'sheets': sheets,
|
|
'columns': columns,
|
|
'preview': {
|
|
'columns': preview_columns,
|
|
'data': preview_data
|
|
}
|
|
})
|
|
except Exception as e:
|
|
log_action("EDIT", f"File refresh failed: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/file/columns', methods=['GET'])
|
|
def get_columns():
|
|
"""Get columns from a specific sheet."""
|
|
try:
|
|
filepath = session.get('current_file')
|
|
sheet_name = request.args.get('sheet')
|
|
|
|
if not filepath or not os.path.exists(filepath):
|
|
return jsonify({'success': False, 'error': 'No file uploaded'}), 400
|
|
|
|
parser = ExcelParser()
|
|
columns = parser.get_columns(filepath, sheet_name)
|
|
|
|
return jsonify({'success': True, 'columns': columns})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/file/preview', methods=['GET'])
|
|
def preview_file():
|
|
"""Preview file data."""
|
|
try:
|
|
filepath = session.get('current_file')
|
|
sheet_name = request.args.get('sheet')
|
|
max_rows_param = request.args.get('max_rows', '20')
|
|
|
|
# If max_rows is 0 or empty, show all rows
|
|
max_rows = int(max_rows_param) if max_rows_param else 0
|
|
if max_rows == 0:
|
|
max_rows = 10000 # Set a reasonable upper limit
|
|
|
|
if not filepath or not os.path.exists(filepath):
|
|
return jsonify({'success': False, 'error': 'No file uploaded'}), 400
|
|
|
|
parser = ExcelParser()
|
|
columns, data = parser.preview(filepath, sheet_name, max_rows)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'columns': columns,
|
|
'data': data,
|
|
'row_count': len(data)
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/validate', methods=['POST'])
|
|
def validate_data():
|
|
"""Validate Excel data with mappings."""
|
|
try:
|
|
data = request.json
|
|
filepath = session.get('current_file')
|
|
sheet_name = data.get('sheet')
|
|
mappings_data = data.get('mappings', [])
|
|
|
|
if not filepath or not os.path.exists(filepath):
|
|
return jsonify({'success': False, 'error': 'No file uploaded'}), 400
|
|
|
|
# Convert mappings to FieldMapping objects
|
|
mappings = [
|
|
FieldMapping(
|
|
excel_column=m.get('excel_column', ''),
|
|
qbo_field=m.get('qbo_field', ''),
|
|
transform=m.get('transform') or None,
|
|
default_value=m.get('default_value') or None,
|
|
required=m.get('required', False)
|
|
)
|
|
for m in mappings_data
|
|
]
|
|
|
|
parser = ExcelParser()
|
|
parse_result = parser.parse(filepath, mappings, sheet_name)
|
|
|
|
# Store parse result in session for import
|
|
session['parse_result'] = {
|
|
'filepath': parse_result.filepath,
|
|
'sheet_name': parse_result.sheet_name,
|
|
'total_rows': parse_result.total_rows,
|
|
'valid_rows': parse_result.valid_rows,
|
|
'error_count': parse_result.error_count,
|
|
'warning_count': parse_result.warning_count
|
|
}
|
|
session['current_mappings'] = mappings_data
|
|
|
|
# Prepare issues for response
|
|
issues = []
|
|
for row in parse_result.rows:
|
|
for issue in row.issues:
|
|
issues.append({
|
|
'row': issue.row,
|
|
'column': issue.column,
|
|
'field': issue.field,
|
|
'message': issue.message,
|
|
'severity': issue.severity.value,
|
|
'value': str(issue.value) if issue.value else None
|
|
})
|
|
|
|
log_action("CREATE", f"Validation complete: {parse_result.valid_rows}/{parse_result.total_rows} valid rows")
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'total_rows': parse_result.total_rows,
|
|
'valid_rows': parse_result.valid_rows,
|
|
'error_count': parse_result.error_count,
|
|
'warning_count': parse_result.warning_count,
|
|
'success_rate': parse_result.success_rate,
|
|
'issues': issues
|
|
})
|
|
except Exception as e:
|
|
import traceback
|
|
error_traceback = traceback.format_exc()
|
|
log_action("CREATE", f"Validation failed: {str(e)}\n{error_traceback}", "error")
|
|
logger.error(f"Validation exception: {error_traceback}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e),
|
|
'details': type(e).__name__,
|
|
'traceback': error_traceback
|
|
}), 400
|
|
|
|
|
|
@app.route('/api/import', methods=['POST'])
|
|
def import_data():
|
|
"""Import data to QuickBooks Online."""
|
|
try:
|
|
data = request.json
|
|
data_type = data.get('data_type', 'check')
|
|
|
|
client = get_qbo_client()
|
|
if not client or not client.is_authenticated:
|
|
return jsonify({'success': False, 'error': 'Not connected to QuickBooks'}), 400
|
|
|
|
filepath = session.get('current_file')
|
|
mappings_data = session.get('current_mappings', data.get('mappings', []))
|
|
sheet_name = data.get('sheet')
|
|
|
|
if not filepath or not os.path.exists(filepath):
|
|
return jsonify({'success': False, 'error': 'No file uploaded'}), 400
|
|
|
|
# Convert mappings
|
|
mappings = [
|
|
FieldMapping(
|
|
excel_column=m.get('excel_column', ''),
|
|
qbo_field=m.get('qbo_field', ''),
|
|
transform=m.get('transform') or None,
|
|
default_value=m.get('default_value') or None,
|
|
required=m.get('required', False)
|
|
)
|
|
for m in mappings_data
|
|
]
|
|
|
|
# Parse the file
|
|
parser = ExcelParser()
|
|
parse_result = parser.parse(filepath, mappings, sheet_name)
|
|
|
|
# Create import processor
|
|
logs = []
|
|
def log_callback(level, message):
|
|
logs.append({'level': level, 'message': message})
|
|
|
|
processor = ImportProcessor(
|
|
client,
|
|
settings.import_settings,
|
|
log_callback=log_callback
|
|
)
|
|
|
|
# Run import
|
|
result = processor.process(parse_result, data_type)
|
|
|
|
log_action("CREATE", f"Import complete: {result.successful} successful, {result.failed} failed, {result.duplicates} duplicates")
|
|
|
|
# Build failed records details for the response
|
|
failed_records = []
|
|
for record in result.records:
|
|
if record.status.value == 'failed':
|
|
failed_records.append({
|
|
'row': record.row_number,
|
|
'error': record.error_message,
|
|
'data': {k: str(v)[:100] for k, v in (record.original_data or {}).items() if v}
|
|
})
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'total_records': result.total_records,
|
|
'successful': result.successful,
|
|
'failed': result.failed,
|
|
'duplicates': result.duplicates,
|
|
'skipped': result.skipped,
|
|
'success_rate': result.success_rate,
|
|
'duration_seconds': result.duration_seconds,
|
|
'logs': logs,
|
|
'failed_records': failed_records
|
|
})
|
|
except Exception as e:
|
|
import traceback
|
|
error_traceback = traceback.format_exc()
|
|
log_action("CREATE", f"Import failed: {str(e)}\n{error_traceback}", "error")
|
|
logger.error(f"Import exception: {error_traceback}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e),
|
|
'details': type(e).__name__,
|
|
'traceback': error_traceback
|
|
}), 400
|
|
|
|
|
|
# =============================================================================
|
|
# Import Job Storage (File-based for multi-worker support)
|
|
# =============================================================================
|
|
import json
|
|
import threading
|
|
|
|
JOBS_DIR = os.path.join(os.path.dirname(__file__), 'data', 'jobs')
|
|
os.makedirs(JOBS_DIR, exist_ok=True)
|
|
|
|
# Thread lock for file operations
|
|
job_lock = threading.Lock()
|
|
|
|
def save_job(job_id, job_data):
|
|
"""Save job data to file."""
|
|
with job_lock:
|
|
job_file = os.path.join(JOBS_DIR, f'{job_id}.json')
|
|
# Convert datetime to string for JSON serialization
|
|
job_copy = job_data.copy()
|
|
if 'start_time' in job_copy and hasattr(job_copy['start_time'], 'isoformat'):
|
|
job_copy['start_time'] = job_copy['start_time'].isoformat()
|
|
with open(job_file, 'w') as f:
|
|
json.dump(job_copy, f)
|
|
|
|
def load_job(job_id):
|
|
"""Load job data from file."""
|
|
job_file = os.path.join(JOBS_DIR, f'{job_id}.json')
|
|
if not os.path.exists(job_file):
|
|
return None
|
|
try:
|
|
with open(job_file, 'r') as f:
|
|
job_data = json.load(f)
|
|
# Convert start_time back to datetime if needed
|
|
if 'start_time' in job_data and isinstance(job_data['start_time'], str):
|
|
job_data['start_time'] = datetime.fromisoformat(job_data['start_time'])
|
|
return job_data
|
|
except Exception as e:
|
|
logger.error(f"Error loading job {job_id}: {e}")
|
|
return None
|
|
|
|
def update_job(job_id, updates):
|
|
"""Update specific fields in a job."""
|
|
with job_lock:
|
|
job_data = load_job(job_id)
|
|
if job_data:
|
|
job_data.update(updates)
|
|
save_job(job_id, job_data)
|
|
return job_data
|
|
return None
|
|
|
|
def cleanup_old_jobs():
|
|
"""Remove job files older than 1 hour."""
|
|
try:
|
|
now = datetime.now()
|
|
for filename in os.listdir(JOBS_DIR):
|
|
if filename.endswith('.json'):
|
|
filepath = os.path.join(JOBS_DIR, filename)
|
|
file_time = datetime.fromtimestamp(os.path.getmtime(filepath))
|
|
if (now - file_time).total_seconds() > 3600: # 1 hour
|
|
os.remove(filepath)
|
|
except Exception as e:
|
|
logger.warning(f"Error cleaning up old jobs: {e}")
|
|
|
|
|
|
@app.route('/api/import/start', methods=['POST'])
|
|
def start_import():
|
|
"""Start an async import job."""
|
|
import uuid
|
|
|
|
# Cleanup old jobs
|
|
cleanup_old_jobs()
|
|
|
|
try:
|
|
data = request.json
|
|
data_type = data.get('data_type', 'check')
|
|
|
|
client = get_qbo_client()
|
|
if not client or not client.is_authenticated:
|
|
return jsonify({'success': False, 'error': 'Not connected to QuickBooks'}), 400
|
|
|
|
filepath = session.get('current_file')
|
|
mappings_data = session.get('current_mappings', data.get('mappings', []))
|
|
sheet_name = data.get('sheet')
|
|
|
|
if not filepath or not os.path.exists(filepath):
|
|
return jsonify({'success': False, 'error': 'No file uploaded'}), 400
|
|
|
|
# Convert mappings
|
|
mappings = [
|
|
FieldMapping(
|
|
excel_column=m.get('excel_column', ''),
|
|
qbo_field=m.get('qbo_field', ''),
|
|
transform=m.get('transform') or None,
|
|
default_value=m.get('default_value') or None,
|
|
required=m.get('required', False)
|
|
)
|
|
for m in mappings_data
|
|
]
|
|
|
|
# Parse the file
|
|
parser = ExcelParser()
|
|
parse_result = parser.parse(filepath, mappings, sheet_name)
|
|
|
|
# Create job
|
|
job_id = str(uuid.uuid4())[:8]
|
|
job_data = {
|
|
'status': 'running',
|
|
'total_records': parse_result.total_rows,
|
|
'processed': 0,
|
|
'successful': 0,
|
|
'failed': 0,
|
|
'duplicates': 0,
|
|
'skipped': 0,
|
|
'recent_results': [],
|
|
'start_time': datetime.now(),
|
|
'error': None,
|
|
'duration_seconds': 0
|
|
}
|
|
save_job(job_id, job_data)
|
|
|
|
# Store credentials for thread
|
|
creds_data = session.get('qbo_credentials')
|
|
if not creds_data:
|
|
# Get from settings if not in session
|
|
creds = settings.get_credentials()
|
|
creds_data = asdict(creds) if creds else None
|
|
|
|
import_settings_data = asdict(settings.import_settings)
|
|
|
|
# Serialize parse_result for thread
|
|
parse_result_data = {
|
|
'filepath': parse_result.filepath,
|
|
'sheet_name': parse_result.sheet_name,
|
|
'columns': parse_result.columns,
|
|
'total_rows': parse_result.total_rows,
|
|
'valid_rows': parse_result.valid_rows,
|
|
'error_count': parse_result.error_count,
|
|
'warning_count': parse_result.warning_count,
|
|
'rows': []
|
|
}
|
|
for row in parse_result.rows:
|
|
parse_result_data['rows'].append({
|
|
'row_number': row.row_number,
|
|
'original_data': row.original_data,
|
|
'qbo_data': row.qbo_data,
|
|
'is_valid': row.is_valid
|
|
})
|
|
|
|
# Start background import
|
|
def run_import():
|
|
try:
|
|
from src.config.settings import QBOCredentials, ImportSettings
|
|
|
|
# Load job from file
|
|
job = load_job(job_id)
|
|
if not job:
|
|
logger.error(f"Job {job_id} not found in run_import")
|
|
return
|
|
|
|
# Recreate client in thread
|
|
creds = QBOCredentials(**creds_data) if creds_data else settings.get_credentials()
|
|
thread_client = QBOClient(creds)
|
|
thread_settings = ImportSettings(**import_settings_data)
|
|
|
|
# Create processor
|
|
processor = ImportProcessor(thread_client, thread_settings)
|
|
|
|
# Rebuild parse result for validation
|
|
from src.core.excel_parser import ParsedRow, ParseResult
|
|
rebuilt_rows = []
|
|
for rd in parse_result_data['rows']:
|
|
row = ParsedRow(
|
|
row_number=rd['row_number'],
|
|
original_data=rd['original_data'],
|
|
qbo_data=rd['qbo_data'],
|
|
is_valid=rd['is_valid']
|
|
)
|
|
rebuilt_rows.append(row)
|
|
|
|
rebuilt_parse_result = ParseResult(
|
|
filepath=parse_result_data['filepath'],
|
|
sheet_name=parse_result_data['sheet_name'],
|
|
columns=parse_result_data['columns'],
|
|
rows=rebuilt_rows,
|
|
total_rows=parse_result_data['total_rows'],
|
|
valid_rows=parse_result_data['valid_rows'],
|
|
error_count=parse_result_data['error_count'],
|
|
warning_count=parse_result_data['warning_count']
|
|
)
|
|
|
|
# Validate and resolve records
|
|
records = processor.validate_and_resolve(rebuilt_parse_result, data_type)
|
|
|
|
# Process each record
|
|
from src.core.import_processor import ImportStatus
|
|
|
|
for i, record in enumerate(records):
|
|
# Reload job to get latest state
|
|
job = load_job(job_id)
|
|
if not job:
|
|
logger.error(f"Job {job_id} disappeared during import")
|
|
return
|
|
|
|
result_entry = None
|
|
|
|
if record.status == ImportStatus.FAILED:
|
|
job['failed'] += 1
|
|
result_entry = {
|
|
'row': record.row_number,
|
|
'status': 'failed',
|
|
'error': record.error_message,
|
|
'key_fields': None,
|
|
'data_type': data_type
|
|
}
|
|
elif record.status == ImportStatus.DUPLICATE:
|
|
job['duplicates'] += 1
|
|
result_entry = {
|
|
'row': record.row_number,
|
|
'status': 'duplicate',
|
|
'error': None,
|
|
'data_type': data_type
|
|
}
|
|
elif record.status == ImportStatus.SKIPPED:
|
|
job['skipped'] += 1
|
|
result_entry = {
|
|
'row': record.row_number,
|
|
'status': 'skipped',
|
|
'error': record.error_message,
|
|
'data_type': data_type
|
|
}
|
|
elif record.status == ImportStatus.PENDING:
|
|
# Import to QBO
|
|
try:
|
|
if data_type == "check":
|
|
check_data = record.qbo_data.copy()
|
|
check_data["PaymentType"] = "Check"
|
|
qbo_entity = thread_client.create_check(check_data)
|
|
elif data_type == "invoice":
|
|
qbo_entity = thread_client.create_invoice(record.qbo_data)
|
|
elif data_type == "bill":
|
|
qbo_entity = thread_client.create_bill(record.qbo_data)
|
|
elif data_type == "customer":
|
|
qbo_entity = thread_client.create_customer(record.qbo_data)
|
|
elif data_type == "vendor":
|
|
qbo_entity = thread_client.create_vendor(record.qbo_data)
|
|
elif data_type == "account":
|
|
qbo_entity = thread_client.create_account(record.qbo_data)
|
|
else:
|
|
raise ValueError(f"Unknown data type: {data_type}")
|
|
|
|
qbo_id = qbo_entity.get("Id")
|
|
job['successful'] += 1
|
|
result_entry = {
|
|
'row': record.row_number,
|
|
'status': 'success',
|
|
'qbo_id': qbo_id,
|
|
'data_type': data_type
|
|
}
|
|
|
|
except Exception as e:
|
|
job['failed'] += 1
|
|
key_fields = processor._get_key_fields(record.qbo_data, data_type)
|
|
result_entry = {
|
|
'row': record.row_number,
|
|
'status': 'failed',
|
|
'error': str(e),
|
|
'key_fields': key_fields,
|
|
'data_type': data_type
|
|
}
|
|
|
|
# Small delay to avoid rate limiting
|
|
import time
|
|
time.sleep(0.1)
|
|
|
|
# Update job progress
|
|
if result_entry:
|
|
job['recent_results'].append(result_entry)
|
|
job['processed'] = i + 1
|
|
|
|
# Keep only last 50 results to limit file size
|
|
if len(job['recent_results']) > 50:
|
|
job['recent_results'] = job['recent_results'][-50:]
|
|
|
|
# Save job state
|
|
save_job(job_id, job)
|
|
|
|
# Final update
|
|
job = load_job(job_id)
|
|
if job:
|
|
job['status'] = 'completed'
|
|
start_time = job.get('start_time')
|
|
if isinstance(start_time, str):
|
|
start_time = datetime.fromisoformat(start_time)
|
|
job['duration_seconds'] = (datetime.now() - start_time).total_seconds()
|
|
save_job(job_id, job)
|
|
|
|
log_action("CREATE", f"Import complete: {job['successful']} successful, {job['failed']} failed")
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
logger.error(f"Import thread error: {traceback.format_exc()}")
|
|
job = load_job(job_id)
|
|
if job:
|
|
job['status'] = 'failed'
|
|
job['error'] = str(e)
|
|
start_time = job.get('start_time')
|
|
if isinstance(start_time, str):
|
|
start_time = datetime.fromisoformat(start_time)
|
|
job['duration_seconds'] = (datetime.now() - start_time).total_seconds()
|
|
save_job(job_id, job)
|
|
|
|
thread = threading.Thread(target=run_import)
|
|
thread.daemon = True
|
|
thread.start()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'job_id': job_id,
|
|
'total_records': parse_result.total_rows
|
|
})
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
error_traceback = traceback.format_exc()
|
|
log_action("CREATE", f"Import start failed: {str(e)}", "error")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e),
|
|
'traceback': error_traceback
|
|
}), 400
|
|
|
|
|
|
@app.route('/api/import/progress/<job_id>', methods=['GET'])
|
|
def get_import_progress(job_id):
|
|
"""Get import job progress."""
|
|
job = load_job(job_id)
|
|
if not job:
|
|
return jsonify({'success': False, 'error': 'Job not found'}), 404
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'status': job['status'],
|
|
'total_records': job['total_records'],
|
|
'processed': job['processed'],
|
|
'successful': job['successful'],
|
|
'failed': job['failed'],
|
|
'duplicates': job['duplicates'],
|
|
'skipped': job['skipped'],
|
|
'recent_results': job['recent_results'][-20:], # Last 20 results
|
|
'error': job['error']
|
|
})
|
|
|
|
|
|
@app.route('/api/import/result/<job_id>', methods=['GET'])
|
|
def get_import_result(job_id):
|
|
"""Get final import result."""
|
|
job = load_job(job_id)
|
|
if not job:
|
|
return jsonify({'success': False, 'error': 'Job not found'}), 404
|
|
|
|
total = job['successful'] + job['failed'] + job['duplicates'] + job['skipped']
|
|
success_rate = (job['successful'] / total * 100) if total > 0 else 0
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'status': job['status'],
|
|
'total_records': job['total_records'],
|
|
'successful': job['successful'],
|
|
'failed': job['failed'],
|
|
'duplicates': job['duplicates'],
|
|
'skipped': job['skipped'],
|
|
'success_rate': success_rate,
|
|
'duration_seconds': job['duration_seconds'],
|
|
'error': job['error']
|
|
})
|
|
|
|
|
|
@app.route('/api/file/clear', methods=['POST'])
|
|
def clear_file():
|
|
"""Clear the current file from session."""
|
|
try:
|
|
filepath = session.get('current_file')
|
|
|
|
# Remove file from uploads folder
|
|
if filepath and os.path.exists(filepath):
|
|
try:
|
|
os.remove(filepath)
|
|
logger.info(f"Removed uploaded file: {filepath}")
|
|
except Exception as e:
|
|
logger.warning(f"Could not remove file: {e}")
|
|
|
|
# Clear session data
|
|
session.pop('current_file', None)
|
|
session.pop('current_filename', None)
|
|
session.pop('excel_columns', None)
|
|
session.pop('current_mappings', None)
|
|
|
|
log_action("DELETE", "File cleared from session")
|
|
|
|
return jsonify({'success': True})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
# =============================================================================
|
|
# Routes - Templates API
|
|
# =============================================================================
|
|
|
|
@app.route('/api/templates', methods=['GET'])
|
|
def get_templates():
|
|
"""Get all mapping templates."""
|
|
try:
|
|
data_type = request.args.get('data_type')
|
|
templates = settings.get_templates(data_type)
|
|
|
|
templates_list = []
|
|
for t in templates:
|
|
templates_list.append({
|
|
'name': t.name,
|
|
'data_type': t.data_type,
|
|
'mappings': [
|
|
{
|
|
'excel_column': m.excel_column,
|
|
'qbo_field': m.qbo_field,
|
|
'transform': m.transform,
|
|
'default_value': m.default_value,
|
|
'required': m.required
|
|
}
|
|
for m in t.mappings
|
|
],
|
|
'created_at': t.created_at,
|
|
'updated_at': t.updated_at
|
|
})
|
|
|
|
return jsonify({'success': True, 'templates': templates_list})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/qbo-fields/<data_type>', methods=['GET'])
|
|
def get_qbo_fields(data_type):
|
|
"""Get available QBO fields for a data type."""
|
|
try:
|
|
# Define QBO fields for each data type with descriptions
|
|
qbo_fields = {
|
|
"check": [
|
|
{"field": "EntityRef.value", "label": "Payee (Entity)", "required": True, "description": "Vendor or Employee name"},
|
|
{"field": "AccountRef.value", "label": "Bank Account", "required": True, "description": "Bank account for the check"},
|
|
{"field": "TxnDate", "label": "Transaction Date", "required": True, "transform": "date"},
|
|
{"field": "Line.Amount", "label": "Line Amount", "required": True, "transform": "currency"},
|
|
{"field": "DocNumber", "label": "Check Number", "required": False},
|
|
{"field": "Line.AccountBasedExpenseLineDetail.AccountRef.value", "label": "Expense Account", "required": False},
|
|
{"field": "PrivateNote", "label": "Private Note/Memo", "required": False},
|
|
{"field": "Line.Description", "label": "Line Description", "required": False},
|
|
{"field": "DepartmentRef.value", "label": "Department", "required": False},
|
|
{"field": "ClassRef.value", "label": "Class", "required": False},
|
|
],
|
|
"invoice": [
|
|
{"field": "CustomerRef.value", "label": "Customer", "required": True, "description": "Customer name"},
|
|
{"field": "TxnDate", "label": "Invoice Date", "required": True, "transform": "date"},
|
|
{"field": "DueDate", "label": "Due Date", "required": False, "transform": "date"},
|
|
{"field": "DocNumber", "label": "Invoice Number", "required": False},
|
|
{"field": "Line.SalesItemLineDetail.ItemRef.value", "label": "Item/Product", "required": False},
|
|
{"field": "Line.Description", "label": "Line Description", "required": False},
|
|
{"field": "Line.SalesItemLineDetail.Qty", "label": "Quantity", "required": False, "transform": "number"},
|
|
{"field": "Line.SalesItemLineDetail.UnitPrice", "label": "Unit Price", "required": False, "transform": "currency"},
|
|
{"field": "Line.Amount", "label": "Line Amount", "required": False, "transform": "currency"},
|
|
{"field": "PrivateNote", "label": "Private Note", "required": False},
|
|
{"field": "CustomerMemo.value", "label": "Customer Memo", "required": False},
|
|
{"field": "BillEmail.Address", "label": "Email Address", "required": False},
|
|
{"field": "DepartmentRef.value", "label": "Department", "required": False},
|
|
{"field": "ClassRef.value", "label": "Class", "required": False},
|
|
],
|
|
"bill": [
|
|
{"field": "VendorRef.value", "label": "Vendor", "required": True, "description": "Vendor name"},
|
|
{"field": "TxnDate", "label": "Bill Date", "required": True, "transform": "date"},
|
|
{"field": "DueDate", "label": "Due Date", "required": False, "transform": "date"},
|
|
{"field": "DocNumber", "label": "Bill Number", "required": False},
|
|
{"field": "Line.AccountBasedExpenseLineDetail.AccountRef.value", "label": "Expense Account", "required": False},
|
|
{"field": "Line.Description", "label": "Line Description", "required": False},
|
|
{"field": "Line.Amount", "label": "Line Amount", "required": False, "transform": "currency"},
|
|
{"field": "PrivateNote", "label": "Private Note/Memo", "required": False},
|
|
{"field": "DepartmentRef.value", "label": "Department", "required": False},
|
|
{"field": "ClassRef.value", "label": "Class", "required": False},
|
|
],
|
|
"customer": [
|
|
{"field": "DisplayName", "label": "Display Name", "required": True, "description": "Name shown in lists"},
|
|
{"field": "CompanyName", "label": "Company Name", "required": False},
|
|
{"field": "GivenName", "label": "First Name", "required": False},
|
|
{"field": "FamilyName", "label": "Last Name", "required": False},
|
|
{"field": "PrimaryEmailAddr.Address", "label": "Email", "required": False},
|
|
{"field": "PrimaryPhone.FreeFormNumber", "label": "Phone", "required": False},
|
|
{"field": "Mobile.FreeFormNumber", "label": "Mobile", "required": False},
|
|
{"field": "BillAddr.Line1", "label": "Billing Street", "required": False},
|
|
{"field": "BillAddr.City", "label": "Billing City", "required": False},
|
|
{"field": "BillAddr.CountrySubDivisionCode", "label": "Billing State", "required": False},
|
|
{"field": "BillAddr.PostalCode", "label": "Billing Zip", "required": False},
|
|
{"field": "BillAddr.Country", "label": "Billing Country", "required": False},
|
|
{"field": "ShipAddr.Line1", "label": "Shipping Street", "required": False},
|
|
{"field": "ShipAddr.City", "label": "Shipping City", "required": False},
|
|
{"field": "ShipAddr.CountrySubDivisionCode", "label": "Shipping State", "required": False},
|
|
{"field": "ShipAddr.PostalCode", "label": "Shipping Zip", "required": False},
|
|
{"field": "Notes", "label": "Notes", "required": False},
|
|
],
|
|
"vendor": [
|
|
{"field": "DisplayName", "label": "Display Name", "required": True, "description": "Name shown in lists"},
|
|
{"field": "CompanyName", "label": "Company Name", "required": False},
|
|
{"field": "GivenName", "label": "First Name", "required": False},
|
|
{"field": "FamilyName", "label": "Last Name", "required": False},
|
|
{"field": "PrimaryEmailAddr.Address", "label": "Email", "required": False},
|
|
{"field": "PrimaryPhone.FreeFormNumber", "label": "Phone", "required": False},
|
|
{"field": "BillAddr.Line1", "label": "Address Street", "required": False},
|
|
{"field": "BillAddr.City", "label": "City", "required": False},
|
|
{"field": "BillAddr.CountrySubDivisionCode", "label": "State", "required": False},
|
|
{"field": "BillAddr.PostalCode", "label": "Zip Code", "required": False},
|
|
{"field": "BillAddr.Country", "label": "Country", "required": False},
|
|
{"field": "TaxIdentifier", "label": "Tax ID", "required": False},
|
|
{"field": "AcctNum", "label": "Account Number", "required": False},
|
|
{"field": "Notes", "label": "Notes", "required": False},
|
|
],
|
|
"account": [
|
|
{"field": "Name", "label": "Account Name", "required": True},
|
|
{"field": "AccountType", "label": "Account Type", "required": True, "description": "Bank, Expense, Income, etc."},
|
|
{"field": "AccountSubType", "label": "Account Sub-Type", "required": False},
|
|
{"field": "AcctNum", "label": "Account Number", "required": False},
|
|
{"field": "Description", "label": "Description", "required": False},
|
|
{"field": "CurrentBalance", "label": "Opening Balance", "required": False, "transform": "currency"},
|
|
],
|
|
}
|
|
|
|
if data_type not in qbo_fields:
|
|
return jsonify({'success': False, 'error': 'Unknown data type'}), 400
|
|
|
|
# Get Excel columns from session if available (from recent upload)
|
|
excel_columns = session.get('excel_columns', [])
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'qbo_fields': qbo_fields[data_type],
|
|
'excel_columns': excel_columns
|
|
})
|
|
except Exception as e:
|
|
log_action("CREATE", f"Failed to get QBO fields: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/templates/default/<data_type>', methods=['GET'])
|
|
def get_default_template(data_type):
|
|
"""Get default template for a data type."""
|
|
try:
|
|
defaults = settings.get_default_templates()
|
|
|
|
if data_type not in defaults:
|
|
return jsonify({'success': False, 'error': 'Unknown data type'}), 400
|
|
|
|
template = defaults[data_type]
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'template': {
|
|
'name': template.name,
|
|
'data_type': template.data_type,
|
|
'mappings': [
|
|
{
|
|
'excel_column': m.excel_column,
|
|
'qbo_field': m.qbo_field,
|
|
'transform': m.transform,
|
|
'default_value': m.default_value,
|
|
'required': m.required
|
|
}
|
|
for m in template.mappings
|
|
]
|
|
}
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/templates', methods=['POST'])
|
|
def save_template():
|
|
"""Save a mapping template."""
|
|
try:
|
|
data = request.json
|
|
log_action("CREATE", f"Creating/updating template: {data.get('name')}")
|
|
|
|
mappings = [
|
|
FieldMapping(
|
|
excel_column=m.get('excel_column', ''),
|
|
qbo_field=m.get('qbo_field', ''),
|
|
transform=m.get('transform') or None,
|
|
default_value=m.get('default_value') or None,
|
|
required=m.get('required', False)
|
|
)
|
|
for m in data.get('mappings', [])
|
|
]
|
|
|
|
template = MappingTemplate(
|
|
name=data.get('name', ''),
|
|
data_type=data.get('data_type', 'check'),
|
|
mappings=mappings
|
|
)
|
|
|
|
settings.save_template(template)
|
|
|
|
logger.info(f"Template saved: {template.name}")
|
|
return jsonify({'success': True, 'message': 'Template saved successfully'})
|
|
except Exception as e:
|
|
log_action("CREATE", f"Failed to save template: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/templates/<path:template_name>', methods=['DELETE'])
|
|
def delete_template(template_name):
|
|
"""Delete a mapping template."""
|
|
try:
|
|
log_action("DELETE", f"Deleting template: {template_name}")
|
|
settings.delete_template(template_name)
|
|
logger.info(f"Template deleted: {template_name}")
|
|
return jsonify({'success': True, 'message': 'Template deleted successfully'})
|
|
except Exception as e:
|
|
log_action("DELETE", f"Failed to delete template: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
# =============================================================================
|
|
# Routes - Settings API
|
|
# =============================================================================
|
|
|
|
@app.route('/api/settings', methods=['GET'])
|
|
def get_settings():
|
|
"""Get import settings."""
|
|
try:
|
|
import_settings = settings.import_settings
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'settings': {
|
|
'skip_duplicates': import_settings.skip_duplicates,
|
|
'duplicate_check_fields': import_settings.duplicate_check_fields,
|
|
'batch_size': import_settings.batch_size,
|
|
'validate_before_import': import_settings.validate_before_import,
|
|
'create_missing_references': import_settings.create_missing_references,
|
|
'date_format': import_settings.date_format,
|
|
'decimal_separator': import_settings.decimal_separator,
|
|
'thousand_separator': import_settings.thousand_separator
|
|
},
|
|
'environment': settings.qbo_environment
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/settings', methods=['POST'])
|
|
def update_settings():
|
|
"""Update import settings."""
|
|
try:
|
|
data = request.json
|
|
log_action("EDIT", "Updating import settings")
|
|
|
|
from src.config.settings import ImportSettings
|
|
|
|
import_settings = ImportSettings(
|
|
skip_duplicates=data.get('skip_duplicates', True),
|
|
duplicate_check_fields=data.get('duplicate_check_fields', ['DocNumber']),
|
|
batch_size=data.get('batch_size', 50),
|
|
validate_before_import=data.get('validate_before_import', True),
|
|
create_missing_references=data.get('create_missing_references', False),
|
|
date_format=data.get('date_format', '%Y-%m-%d'),
|
|
decimal_separator=data.get('decimal_separator', '.'),
|
|
thousand_separator=data.get('thousand_separator', ',')
|
|
)
|
|
|
|
settings.import_settings = import_settings
|
|
|
|
if 'environment' in data:
|
|
settings.qbo_environment = data['environment']
|
|
|
|
logger.info("Settings updated successfully")
|
|
return jsonify({'success': True, 'message': 'Settings saved successfully'})
|
|
except Exception as e:
|
|
log_action("EDIT", f"Failed to update settings: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
@app.route('/api/settings/clear-credentials', methods=['POST'])
|
|
def clear_credentials():
|
|
"""Clear saved credentials."""
|
|
try:
|
|
log_action("DELETE", "Clearing saved credentials")
|
|
settings.clear_credentials()
|
|
session.pop('qbo_credentials', None)
|
|
logger.info("Credentials cleared successfully")
|
|
return jsonify({'success': True, 'message': 'Credentials cleared successfully'})
|
|
except Exception as e:
|
|
log_action("DELETE", f"Failed to clear credentials: {str(e)}", "error")
|
|
return jsonify({'success': False, 'error': str(e)}), 400
|
|
|
|
|
|
# =============================================================================
|
|
# Error Handlers
|
|
# =============================================================================
|
|
|
|
@app.errorhandler(404)
|
|
def not_found(e):
|
|
"""Handle 404 errors."""
|
|
return render_template('404.html'), 404
|
|
|
|
|
|
@app.errorhandler(500)
|
|
def server_error(e):
|
|
"""Handle 500 errors."""
|
|
logger.error(f"Server error: {e}")
|
|
return render_template('500.html'), 500
|
|
|
|
|
|
# =============================================================================
|
|
# Main
|
|
# =============================================================================
|
|
|
|
if __name__ == '__main__':
|
|
# Ensure upload directory exists
|
|
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, threaded=False) |