1318 lines
52 KiB
Python
1318 lines
52 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.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_qbo_client():
|
|
"""Get or create QBO client from session or saved credentials."""
|
|
# 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
|
|
|
|
|
|
@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
|
|
company_name = ""
|
|
|
|
if is_connected:
|
|
try:
|
|
company_info = client.get_company_info()
|
|
company_name = company_info.get('CompanyName', '')
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
'is_connected': is_connected,
|
|
'company_name': company_name,
|
|
'environment': settings.qbo_environment
|
|
}
|
|
|
|
|
|
def log_action(action: str, details: str = "", level: str = "info"):
|
|
"""Log user actions for audit trail."""
|
|
timestamp = datetime.now().isoformat()
|
|
user_ip = request.remote_addr
|
|
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")
|
|
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}")
|
|
|
|
return render_template('index.html',
|
|
is_connected=is_connected,
|
|
company_name=company_name,
|
|
environment=settings.qbo_environment)
|
|
|
|
|
|
@app.route('/connection')
|
|
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
|
|
|
|
return render_template('connection.html',
|
|
credentials=creds,
|
|
is_connected=is_connected,
|
|
company_name=company_name,
|
|
environment=settings.qbo_environment)
|
|
|
|
|
|
@app.route('/import')
|
|
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()
|
|
|
|
return render_template('import.html',
|
|
is_connected=is_connected,
|
|
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/connection/test', methods=['GET'])
|
|
def test_connection():
|
|
"""Test QBO connection."""
|
|
try:
|
|
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"Connection test successful: {len(customers)} customers, {len(vendors)} vendors, {len(accounts)} accounts")
|
|
|
|
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)
|
|
})
|
|
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 (in-memory for simplicity)
|
|
import_jobs = {}
|
|
|
|
@app.route('/api/import/start', methods=['POST'])
|
|
def start_import():
|
|
"""Start an async import job."""
|
|
import uuid
|
|
import threading
|
|
|
|
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]
|
|
import_jobs[job_id] = {
|
|
'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
|
|
}
|
|
|
|
# Store credentials for thread
|
|
creds_data = session.get('qbo_credentials')
|
|
import_settings_data = asdict(settings.import_settings)
|
|
|
|
# Start background import
|
|
def run_import():
|
|
try:
|
|
from src.config.settings import QBOCredentials, ImportSettings
|
|
|
|
# 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)
|
|
|
|
job = import_jobs[job_id]
|
|
|
|
# Create processor
|
|
processor = ImportProcessor(thread_client, thread_settings)
|
|
|
|
# Validate and resolve records
|
|
records = processor.validate_and_resolve(parse_result, data_type)
|
|
|
|
# Process each record
|
|
from src.core.import_processor import ImportStatus
|
|
|
|
for i, record in enumerate(records):
|
|
if record.status == ImportStatus.FAILED:
|
|
job['failed'] += 1
|
|
job['recent_results'].append({
|
|
'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
|
|
job['recent_results'].append({
|
|
'row': record.row_number,
|
|
'status': 'duplicate',
|
|
'error': None,
|
|
'data_type': data_type
|
|
})
|
|
elif record.status == ImportStatus.SKIPPED:
|
|
job['skipped'] += 1
|
|
job['recent_results'].append({
|
|
'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
|
|
job['recent_results'].append({
|
|
'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)
|
|
job['recent_results'].append({
|
|
'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)
|
|
|
|
job['processed'] = i + 1
|
|
|
|
# Keep only last 50 results to limit memory
|
|
if len(job['recent_results']) > 50:
|
|
job['recent_results'] = job['recent_results'][-50:]
|
|
|
|
job['status'] = 'completed'
|
|
job['duration_seconds'] = (datetime.now() - job['start_time']).total_seconds()
|
|
|
|
log_action("CREATE", f"Import complete: {job['successful']} successful, {job['failed']} failed")
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
job = import_jobs.get(job_id)
|
|
if job:
|
|
job['status'] = 'failed'
|
|
job['error'] = str(e)
|
|
job['duration_seconds'] = (datetime.now() - job['start_time']).total_seconds()
|
|
logger.error(f"Import thread error: {traceback.format_exc()}")
|
|
|
|
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 = import_jobs.get(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 = import_jobs.get(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
|
|
debug_mode = os.environ.get('FLASK_DEBUG', 'true').lower() == 'true'
|
|
port = int(os.environ.get('PORT', 5000))
|
|
|
|
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) |