nngo/QB-Import-from-Excel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0f913ae3c936cb36b4cd0974c3e98746da780a63
QB-Import-from-Excel/desktop_app.py
T
nngo 0f913ae3c9 Update desktop app
2026-02-18 10:23:36 -05:00

4503 lines
183 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
QBO Excel Sync - Desktop Application
A Tkinter-based desktop application for importing Excel data to QuickBooks.
Supports both QuickBooks Online (QBO) and QuickBooks Desktop (QBD).
"""
import os
import sys
import json
import logging
import threading
import webbrowser
import subprocess
from pathlib import Path
from datetime import datetime
from typing import Optional, Dict, Any, List, Callable
from dataclasses import dataclass
import tkinter as tk
from tkinter import ttk, filedialog, messagebox, scrolledtext
# Add the project root to path
PROJECT_ROOT = Path(__file__).parent
sys.path.insert(0, str(PROJECT_ROOT))
# Import project modules
from src.config.settings import Settings, ImportSettings, QBOCredentials, ConnectionProfile
from src.api.qbo_client import QBOClient
from src.core.excel_parser import ExcelParser, ParseResult
from src.core.import_processor import ImportProcessor, ImportResult
# Try to import QBD client (Windows only)
try:
from src.api.qbd_client import QBDesktopClient, QBDCredentials, QBDesktopError, check_qbd_availability
HAS_QBD = True
except ImportError:
HAS_QBD = False
QBDesktopClient = None
QBDCredentials = None
QBDesktopError = Exception
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('desktop_app.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# =============================================================================
# Application Constants
# =============================================================================
APP_NAME = "QBO Excel Sync"
APP_VERSION = "1.0.0"
WINDOW_WIDTH = 1200
WINDOW_HEIGHT = 800
MIN_WIDTH = 900
MIN_HEIGHT = 600
# Supported data types
DATA_TYPES = [
("Check", "check"),
("Invoice", "invoice"),
("Bill", "bill"),
("Customer", "customer"),
("Vendor", "vendor"),
("Account", "account"),
]
# Color scheme
COLORS = {
'primary': '#2563eb',
'primary_dark': '#1d4ed8',
'success': '#22c55e',
'warning': '#f59e0b',
'danger': '#ef4444',
'bg': '#f8fafc',
'card_bg': '#ffffff',
'text': '#1e293b',
'text_secondary': '#64748b',
'border': '#e2e8f0',
}
# =============================================================================
# Custom Widgets
# =============================================================================
class ModernButton(ttk.Button):
"""A modern styled button."""
pass
class StatusBar(ttk.Frame):
"""Status bar widget showing connection status."""
def __init__(self, parent, **kwargs):
super().__init__(parent, **kwargs)
self.status_label = ttk.Label(self, text="Not Connected", font=('Segoe UI', 9))
self.status_label.pack(side=tk.LEFT, padx=10)
self.connection_indicator = ttk.Label(self, text="●", foreground='red', font=('Segoe UI', 12))
self.connection_indicator.pack(side=tk.LEFT)
self.company_label = ttk.Label(self, text="", font=('Segoe UI', 9, 'italic'))
self.company_label.pack(side=tk.LEFT, padx=10)
self.version_label = ttk.Label(self, text=f"v{APP_VERSION}", font=('Segoe UI', 8))
self.version_label.pack(side=tk.RIGHT, padx=10)
def set_connected(self, company_name: str, connection_type: str = 'qbo'):
"""Update status to connected."""
type_text = "Desktop" if connection_type == 'qbd' else "Online"
self.status_label.config(text=f"Connected ({type_text})")
self.connection_indicator.config(foreground='green')
self.company_label.config(text=company_name)
def set_disconnected(self):
"""Update status to disconnected."""
self.status_label.config(text="Not Connected")
self.connection_indicator.config(foreground='red')
self.company_label.config(text="")
class LogViewer(ttk.Frame):
"""A log viewer widget with scrollable text."""
def __init__(self, parent, **kwargs):
super().__init__(parent, **kwargs)
self.text = scrolledtext.ScrolledText(
self,
wrap=tk.WORD,
font=('Consolas', 9),
state=tk.DISABLED,
height=8
)
self.text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
# Configure tags for different log levels
self.text.tag_configure('INFO', foreground='black')
self.text.tag_configure('SUCCESS', foreground='green')
self.text.tag_configure('WARNING', foreground='orange')
self.text.tag_configure('ERROR', foreground='red')
def log(self, message: str, level: str = 'INFO'):
"""Add a log message."""
self.text.config(state=tk.NORMAL)
timestamp = datetime.now().strftime('%H:%M:%S')
self.text.insert(tk.END, f"[{timestamp}] {message}\n", level)
self.text.see(tk.END)
self.text.config(state=tk.DISABLED)
def clear(self):
"""Clear all log messages."""
self.text.config(state=tk.NORMAL)
self.text.delete(1.0, tk.END)
self.text.config(state=tk.DISABLED)
class ProgressFrame(ttk.Frame):
"""A frame showing import progress."""
def __init__(self, parent, **kwargs):
super().__init__(parent, **kwargs)
self.label = ttk.Label(self, text="Ready", font=('Segoe UI', 10))
self.label.pack(fill=tk.X, padx=5, pady=2)
self.progress = ttk.Progressbar(self, mode='determinate', length=400)
self.progress.pack(fill=tk.X, padx=5, pady=2)
self.detail_label = ttk.Label(self, text="", font=('Segoe UI', 9), foreground='gray')
self.detail_label.pack(fill=tk.X, padx=5, pady=2)
def set_progress(self, value: int, maximum: int, text: str = "", detail: str = ""):
"""Update progress."""
self.progress['maximum'] = maximum
self.progress['value'] = value
if text:
self.label.config(text=text)
if detail:
self.detail_label.config(text=detail)
def reset(self):
"""Reset progress."""
self.progress['value'] = 0
self.label.config(text="Ready")
self.detail_label.config(text="")
# =============================================================================
# Main Application
# =============================================================================
class QBOExcelSyncApp:
"""Main application class."""
def __init__(self):
self.root = tk.Tk()
self.root.title(APP_NAME)
self.root.geometry(f"{WINDOW_WIDTH}x{WINDOW_HEIGHT}")
self.root.minsize(MIN_WIDTH, MIN_HEIGHT)
# Maximize window on startup (works on Windows)
try:
self.root.state('zoomed') # Windows
except:
try:
# Linux/Mac alternative
self.root.attributes('-zoomed', True)
except:
pass # Fall back to default size
# Application state
self.settings = Settings()
self.qbo_client: Optional[QBOClient] = None
self.qbd_client: Optional[QBDesktopClient] = None
self.connection_type: str = 'qbo' # 'qbo' or 'qbd'
self.is_connected: bool = False
self.company_name: str = ""
# Current import state
self.current_file: Optional[Path] = None
self.parse_result: Optional[ParseResult] = None
self.field_mappings: Dict[str, str] = {}
self.excel_columns: List[str] = []
self.excel_data: List[Dict[str, Any]] = []
self.parser: Optional[ExcelParser] = None
# Setup UI
self._setup_styles()
self._create_menu()
self._create_main_layout()
self._create_status_bar()
# Don't auto-connect - let user select connection type manually
# self._try_restore_connection()
# Bind window close event
self.root.protocol("WM_DELETE_WINDOW", self._on_close)
logger.info("Application started")
def _setup_styles(self):
"""Configure ttk styles."""
style = ttk.Style()
# Use a modern theme if available
available_themes = style.theme_names()
if 'clam' in available_themes:
style.theme_use('clam')
elif 'vista' in available_themes:
style.theme_use('vista')
# Configure custom styles
style.configure('Title.TLabel', font=('Segoe UI', 16, 'bold'))
style.configure('Subtitle.TLabel', font=('Segoe UI', 11))
style.configure('Header.TLabel', font=('Segoe UI', 12, 'bold'))
style.configure('Card.TFrame', background='white', relief='solid', borderwidth=1)
style.configure('Primary.TButton', font=('Segoe UI', 10))
style.configure('Success.TLabel', foreground='green')
style.configure('Error.TLabel', foreground='red')
style.configure('Warning.TLabel', foreground='orange')
def _create_menu(self):
"""Create the application menu bar."""
menubar = tk.Menu(self.root)
self.root.config(menu=menubar)
# File menu
file_menu = tk.Menu(menubar, tearoff=0)
menubar.add_cascade(label="File", menu=file_menu)
file_menu.add_command(label="Open Excel File...", command=self._open_file, accelerator="Ctrl+O")
file_menu.add_separator()
file_menu.add_command(label="Exit", command=self._on_close, accelerator="Alt+F4")
# Connection menu
conn_menu = tk.Menu(menubar, tearoff=0)
menubar.add_cascade(label="Connection", menu=conn_menu)
conn_menu.add_command(label="Connect to QuickBooks Online...", command=self._show_qbo_connect)
if HAS_QBD:
conn_menu.add_command(label="Connect to QuickBooks Desktop...", command=self._connect_qbd)
conn_menu.add_separator()
conn_menu.add_command(label="Disconnect", command=self._disconnect)
conn_menu.add_command(label="Test Connection", command=self._test_connection)
# Tools menu
tools_menu = tk.Menu(menubar, tearoff=0)
menubar.add_cascade(label="Tools", menu=tools_menu)
tools_menu.add_command(label="Manage Templates...", command=self._show_templates)
tools_menu.add_command(label="Settings...", command=self._show_settings)
tools_menu.add_separator()
# QuickBooks Lookup submenu
lookup_menu = tk.Menu(tools_menu, tearoff=0)
tools_menu.add_cascade(label="QuickBooks Lookup", menu=lookup_menu)
lookup_menu.add_command(label="Accounts...", command=lambda: self._show_qb_lookup('account'))
lookup_menu.add_command(label="Customers...", command=lambda: self._show_qb_lookup('customer'))
lookup_menu.add_command(label="Vendors...", command=lambda: self._show_qb_lookup('vendor'))
lookup_menu.add_command(label="Items/Products...", command=lambda: self._show_qb_lookup('item'))
lookup_menu.add_separator()
lookup_menu.add_command(label="Classes...", command=lambda: self._show_qb_lookup('class'))
lookup_menu.add_command(label="Departments...", command=lambda: self._show_qb_lookup('department'))
lookup_menu.add_command(label="Payment Terms...", command=lambda: self._show_qb_lookup('term'))
lookup_menu.add_command(label="Payment Methods...", command=lambda: self._show_qb_lookup('paymentmethod'))
lookup_menu.add_command(label="Employees...", command=lambda: self._show_qb_lookup('employee'))
tools_menu.add_separator()
tools_menu.add_command(label="View Logs...", command=self._show_logs)
# Help menu
help_menu = tk.Menu(menubar, tearoff=0)
menubar.add_cascade(label="Help", menu=help_menu)
help_menu.add_command(label="Documentation", command=self._show_help)
help_menu.add_command(label="About", command=self._show_about)
# Keyboard shortcuts
self.root.bind('<Control-o>', lambda e: self._open_file())
def _create_main_layout(self):
"""Create the main application layout."""
# Main container
self.main_frame = ttk.Frame(self.root, padding=10)
self.main_frame.pack(fill=tk.BOTH, expand=True)
# Create notebook for tabs
self.notebook = ttk.Notebook(self.main_frame)
self.notebook.pack(fill=tk.BOTH, expand=True)
# Create tabs - Order: Dashboard | Connection | Import | Templates | Lookup
self._create_dashboard_tab()
self._create_connection_tab()
self._create_import_tab()
self._create_templates_tab()
self._create_lookup_tab()
def _create_dashboard_tab(self):
"""Create the dashboard tab."""
tab = ttk.Frame(self.notebook, padding=20)
self.notebook.add(tab, text=" Dashboard ")
# Welcome header
header_frame = ttk.Frame(tab)
header_frame.pack(fill=tk.X, pady=(0, 20))
ttk.Label(
header_frame,
text="Welcome to QBO Excel Sync",
style='Title.TLabel'
).pack(anchor=tk.W)
ttk.Label(
header_frame,
text="Import your Excel data to QuickBooks with ease.",
style='Subtitle.TLabel',
foreground='gray'
).pack(anchor=tk.W, pady=(5, 0))
# Stats cards frame
stats_frame = ttk.Frame(tab)
stats_frame.pack(fill=tk.X, pady=10)
# Connection status card
self.conn_card = self._create_stat_card(
stats_frame,
"Connection Status",
"Not Connected",
"Connect to QuickBooks to start"
)
self.conn_card.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(0, 10))
# Supported types card
types_card = self._create_stat_card(
stats_frame,
"Supported Types",
"6 Types",
"Checks, Invoices, Bills, & more"
)
types_card.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(10, 0))
# Quick actions
actions_frame = ttk.LabelFrame(tab, text="Quick Actions", padding=15)
actions_frame.pack(fill=tk.X, pady=20)
actions_inner = ttk.Frame(actions_frame)
actions_inner.pack(fill=tk.X)
ttk.Button(
actions_inner,
text="🔗 Manage Connection",
command=lambda: self.notebook.select(1), # Connection tab (index 1)
width=25
).pack(side=tk.LEFT, padx=5)
ttk.Button(
actions_inner,
text="📁 Import Excel File",
command=lambda: self.notebook.select(2), # Import tab (index 2)
width=25
).pack(side=tk.LEFT, padx=5)
ttk.Button(
actions_inner,
text="📋 Manage Templates",
command=lambda: self.notebook.select(3), # Templates tab (index 3)
width=25
).pack(side=tk.LEFT, padx=5)
ttk.Button(
actions_inner,
text="🔍 QuickBooks Lookup",
command=lambda: self.notebook.select(4), # Lookup tab (index 4)
width=25
).pack(side=tk.LEFT, padx=5)
# Supported data types
types_frame = ttk.LabelFrame(tab, text="Supported Data Types", padding=15)
types_frame.pack(fill=tk.BOTH, expand=True, pady=10)
types_grid = ttk.Frame(types_frame)
types_grid.pack(fill=tk.BOTH, expand=True)
type_info = [
("✓ Checks", "Import check payments with payee, amount, and account details."),
("✓ Invoices", "Create invoices with customer, items, and payment terms."),
("✓ Bills", "Import vendor bills with expense accounts and due dates."),
("✓ Customers", "Add new customers with contact and address information."),
("✓ Vendors", "Import vendor records with payment and tax details."),
("✓ Accounts", "Create accounts with types, numbers, and descriptions."),
]
for i, (title, desc) in enumerate(type_info):
row = i // 3
col = i % 3
type_card = ttk.Frame(types_grid, padding=10)
type_card.grid(row=row, column=col, sticky='nsew', padx=5, pady=5)
ttk.Label(type_card, text=title, font=('Segoe UI', 10, 'bold')).pack(anchor=tk.W)
ttk.Label(type_card, text=desc, wraplength=200, foreground='gray').pack(anchor=tk.W)
types_grid.columnconfigure(col, weight=1)
def _create_stat_card(self, parent, title: str, value: str, detail: str) -> ttk.Frame:
"""Create a statistics card widget."""
card = ttk.Frame(parent, padding=15, relief='solid', borderwidth=1)
ttk.Label(card, text=title, font=('Segoe UI', 9), foreground='gray').pack(anchor=tk.W)
value_label = ttk.Label(card, text=value, font=('Segoe UI', 14, 'bold'))
value_label.pack(anchor=tk.W, pady=(5, 0))
detail_label = ttk.Label(card, text=detail, font=('Segoe UI', 9), foreground='gray')
detail_label.pack(anchor=tk.W)
# Store references for updating
card.value_label = value_label
card.detail_label = detail_label
return card
def _create_import_tab(self):
"""Create the import tab."""
tab = ttk.Frame(self.notebook, padding=20)
self.notebook.add(tab, text=" Import ")
# Left panel - File selection and options
left_frame = ttk.Frame(tab)
left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(0, 10))
# File selection
file_frame = ttk.LabelFrame(left_frame, text="1. Select Excel File", padding=15)
file_frame.pack(fill=tk.X, pady=(0, 10))
file_inner = ttk.Frame(file_frame)
file_inner.pack(fill=tk.X)
self.file_entry = ttk.Entry(file_inner, width=50)
self.file_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 10))
ttk.Button(file_inner, text="Browse...", command=self._open_file).pack(side=tk.LEFT)
ttk.Button(file_inner, text="Refresh", command=self._refresh_file).pack(side=tk.LEFT, padx=(5, 0))
self.file_info_label = ttk.Label(file_frame, text="No file selected", foreground='gray')
self.file_info_label.pack(anchor=tk.W, pady=(10, 0))
# Data type and template selection
type_frame = ttk.LabelFrame(left_frame, text="2. Select Data Type & Template", padding=15)
type_frame.pack(fill=tk.X, pady=10)
# Data type radio buttons
self.data_type_var = tk.StringVar(value="check")
type_grid = ttk.Frame(type_frame)
type_grid.pack(fill=tk.X)
for i, (label, value) in enumerate(DATA_TYPES):
rb = ttk.Radiobutton(
type_grid,
text=label,
variable=self.data_type_var,
value=value,
command=self._on_data_type_change
)
rb.grid(row=i // 3, column=i % 3, sticky=tk.W, padx=10, pady=5)
# Template selection
template_row = ttk.Frame(type_frame)
template_row.pack(fill=tk.X, pady=(10, 0))
ttk.Label(template_row, text="Template:").pack(side=tk.LEFT)
self.template_var = tk.StringVar(value="")
self.template_combo = ttk.Combobox(
template_row,
textvariable=self.template_var,
values=["(Default Template)"],
width=30,
state='readonly'
)
self.template_combo.pack(side=tk.LEFT, padx=(10, 0))
self.template_combo.current(0)
self.template_combo.bind('<<ComboboxSelected>>', self._on_template_change)
ttk.Button(template_row, text="↻", command=self._refresh_template_list, width=3).pack(side=tk.LEFT, padx=(5, 0))
# Field mapping
mapping_frame = ttk.LabelFrame(left_frame, text="3. Field Mapping", padding=15)
mapping_frame.pack(fill=tk.BOTH, expand=True, pady=10)
# Mapping treeview
mapping_inner = ttk.Frame(mapping_frame)
mapping_inner.pack(fill=tk.BOTH, expand=True)
columns = ('excel_column', 'qbo_field')
self.mapping_tree = ttk.Treeview(mapping_inner, columns=columns, show='headings', height=10)
self.mapping_tree.heading('excel_column', text='Excel Column')
self.mapping_tree.heading('qbo_field', text='QuickBooks Field')
self.mapping_tree.column('excel_column', width=200)
self.mapping_tree.column('qbo_field', width=200)
mapping_scroll = ttk.Scrollbar(mapping_inner, orient=tk.VERTICAL, command=self.mapping_tree.yview)
self.mapping_tree.configure(yscrollcommand=mapping_scroll.set)
self.mapping_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
mapping_scroll.pack(side=tk.RIGHT, fill=tk.Y)
mapping_buttons = ttk.Frame(mapping_frame)
mapping_buttons.pack(fill=tk.X, pady=(10, 0))
ttk.Button(mapping_buttons, text="Auto-Map", command=self._auto_map_fields).pack(side=tk.LEFT)
ttk.Button(mapping_buttons, text="Edit Mapping...", command=self._edit_mapping).pack(side=tk.LEFT, padx=5)
ttk.Button(mapping_buttons, text="Save as Template", command=self._save_template).pack(side=tk.LEFT)
# Right panel - Preview and import
right_frame = ttk.Frame(tab)
right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=(10, 0))
# Data preview
preview_frame = ttk.LabelFrame(right_frame, text="Data Preview", padding=15)
preview_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10))
self.preview_tree = ttk.Treeview(preview_frame, show='headings', height=10)
preview_scroll_y = ttk.Scrollbar(preview_frame, orient=tk.VERTICAL, command=self.preview_tree.yview)
preview_scroll_x = ttk.Scrollbar(preview_frame, orient=tk.HORIZONTAL, command=self.preview_tree.xview)
self.preview_tree.configure(yscrollcommand=preview_scroll_y.set, xscrollcommand=preview_scroll_x.set)
self.preview_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
preview_scroll_y.pack(side=tk.RIGHT, fill=tk.Y)
# Progress and import
import_frame = ttk.LabelFrame(right_frame, text="Import", padding=15)
import_frame.pack(fill=tk.X, pady=10)
self.progress_frame = ProgressFrame(import_frame)
self.progress_frame.pack(fill=tk.X, pady=(0, 10))
import_buttons = ttk.Frame(import_frame)
import_buttons.pack(fill=tk.X)
self.import_btn = ttk.Button(
import_buttons,
text="▶ Start Import",
command=self._start_import,
width=20
)
self.import_btn.pack(side=tk.LEFT)
self.stop_btn = ttk.Button(
import_buttons,
text="■ Stop",
command=self._stop_import,
state=tk.DISABLED,
width=10
)
self.stop_btn.pack(side=tk.LEFT, padx=10)
# Log viewer
log_frame = ttk.LabelFrame(right_frame, text="Import Log", padding=10)
log_frame.pack(fill=tk.BOTH, expand=True)
self.log_viewer = LogViewer(log_frame)
self.log_viewer.pack(fill=tk.BOTH, expand=True)
def _create_connection_tab(self):
"""Create the connection tab."""
tab = ttk.Frame(self.notebook, padding=20)
self.notebook.add(tab, text=" Connection ")
# Connection type selection
type_frame = ttk.LabelFrame(tab, text="Connection Type", padding=15)
type_frame.pack(fill=tk.X, pady=(0, 20))
self.conn_type_var = tk.StringVar(value='qbo')
qbo_rb = ttk.Radiobutton(
type_frame,
text="QuickBooks Online",
variable=self.conn_type_var,
value='qbo',
command=self._on_conn_type_change
)
qbo_rb.pack(anchor=tk.W, pady=5)
qbd_rb = ttk.Radiobutton(
type_frame,
text="QuickBooks Desktop" + ("" if HAS_QBD else " (Not Available)"),
variable=self.conn_type_var,
value='qbd',
state=tk.NORMAL if HAS_QBD else tk.DISABLED,
command=self._on_conn_type_change
)
qbd_rb.pack(anchor=tk.W, pady=5)
# QBO Connection Frame
self.qbo_frame = ttk.LabelFrame(tab, text="QuickBooks Online Connection", padding=15)
self.qbo_frame.pack(fill=tk.X, pady=10)
# Profile selection row
profile_frame = ttk.Frame(self.qbo_frame)
profile_frame.pack(fill=tk.X, pady=(0, 15))
ttk.Label(profile_frame, text="Connection Profile:", font=('Segoe UI', 10, 'bold')).pack(side=tk.LEFT)
self.profile_var = tk.StringVar(value="")
self.profile_combo = ttk.Combobox(
profile_frame,
textvariable=self.profile_var,
values=[],
width=30,
state='readonly'
)
self.profile_combo.pack(side=tk.LEFT, padx=(10, 5))
self.profile_combo.bind('<<ComboboxSelected>>', self._on_profile_change)
ttk.Button(profile_frame, text="Load", command=self._load_selected_profile, width=8).pack(side=tk.LEFT, padx=2)
ttk.Button(profile_frame, text="Save", command=self._save_profile, width=8).pack(side=tk.LEFT, padx=2)
ttk.Button(profile_frame, text="Save As...", command=self._save_profile_as, width=10).pack(side=tk.LEFT, padx=2)
ttk.Button(profile_frame, text="Delete", command=self._delete_profile, width=8).pack(side=tk.LEFT, padx=2)
# Import/Export buttons row
profile_io_frame = ttk.Frame(self.qbo_frame)
profile_io_frame.pack(fill=tk.X, pady=(5, 0))
ttk.Label(profile_io_frame, text="Profile Import/Export:", foreground='gray').pack(side=tk.LEFT)
ttk.Button(profile_io_frame, text="Import Profile", command=self._import_profile, width=14).pack(side=tk.LEFT, padx=(10, 2))
ttk.Button(profile_io_frame, text="Export Profile", command=self._export_profile, width=14).pack(side=tk.LEFT, padx=2)
ttk.Button(profile_io_frame, text="Export All", command=self._export_all_profiles, width=12).pack(side=tk.LEFT, padx=2)
# Separator
ttk.Separator(self.qbo_frame, orient=tk.HORIZONTAL).pack(fill=tk.X, pady=10)
# API Credentials
creds_frame = ttk.Frame(self.qbo_frame)
creds_frame.pack(fill=tk.X, pady=(0, 10))
ttk.Label(creds_frame, text="Client ID:").grid(row=0, column=0, sticky=tk.W, pady=5)
self.client_id_entry = ttk.Entry(creds_frame, width=50)
self.client_id_entry.grid(row=0, column=1, sticky=tk.W, padx=10, pady=5)
ttk.Label(creds_frame, text="Client Secret:").grid(row=1, column=0, sticky=tk.W, pady=5)
self.client_secret_entry = ttk.Entry(creds_frame, width=50, show='*')
self.client_secret_entry.grid(row=1, column=1, sticky=tk.W, padx=10, pady=5)
ttk.Label(creds_frame, text="Redirect URI:").grid(row=2, column=0, sticky=tk.W, pady=5)
self.redirect_uri_entry = ttk.Entry(creds_frame, width=50)
self.redirect_uri_entry.insert(0, "http://localhost:9000/oauth/callback")
self.redirect_uri_entry.grid(row=2, column=1, sticky=tk.W, padx=10, pady=5)
ttk.Label(creds_frame, text="Environment:").grid(row=3, column=0, sticky=tk.W, pady=5)
self.env_var = tk.StringVar(value=self.settings.qbo_environment)
env_combo = ttk.Combobox(creds_frame, textvariable=self.env_var, values=['sandbox', 'production'], width=15)
env_combo.grid(row=3, column=1, sticky=tk.W, padx=10, pady=5)
# QBO buttons
qbo_buttons = ttk.Frame(self.qbo_frame)
qbo_buttons.pack(fill=tk.X, pady=10)
ttk.Button(qbo_buttons, text="Connect to QuickBooks", command=self._start_oauth, width=25).pack(side=tk.LEFT)
# Advanced options in a second row
qbo_buttons2 = ttk.Frame(self.qbo_frame)
qbo_buttons2.pack(fill=tk.X, pady=(5, 0))
ttk.Label(qbo_buttons2, text="Advanced:", foreground='gray').pack(side=tk.LEFT)
ttk.Button(qbo_buttons2, text="Enter Code Manually", command=self._show_auth_code_dialog, width=22).pack(side=tk.LEFT, padx=(10, 5))
ttk.Button(qbo_buttons2, text="Enter Token Manually", command=self._enter_token_manually, width=22).pack(side=tk.LEFT)
# QBD Connection Frame
self.qbd_frame = ttk.LabelFrame(tab, text="QuickBooks Desktop Connection", padding=15)
# Initially hidden - will be shown when QBD is selected
if HAS_QBD:
qbd_info = ttk.Label(
self.qbd_frame,
text="Connect to QuickBooks Desktop running on this computer.\n"
"Make sure QuickBooks Desktop is open with a company file.",
foreground='gray'
)
else:
qbd_info = ttk.Label(
self.qbd_frame,
text="QuickBooks Desktop integration is not available.\n"
"This feature requires Windows and pywin32.",
foreground='red'
)
qbd_info.pack(anchor=tk.W, pady=(0, 10))
qbd_buttons = ttk.Frame(self.qbd_frame)
qbd_buttons.pack(fill=tk.X)
self.qbd_connect_btn = ttk.Button(
qbd_buttons,
text="Connect to QuickBooks Desktop",
command=self._connect_qbd,
state=tk.NORMAL if HAS_QBD else tk.DISABLED
)
self.qbd_connect_btn.pack(side=tk.LEFT)
# Connection status
self.status_frame = ttk.LabelFrame(tab, text="Connection Status", padding=15)
self.status_frame.pack(fill=tk.X, pady=20)
self.conn_status_label = ttk.Label(
self.status_frame,
text="● Not Connected",
font=('Segoe UI', 11)
)
self.conn_status_label.pack(anchor=tk.W)
self.conn_company_label = ttk.Label(
self.status_frame,
text="",
font=('Segoe UI', 10),
foreground='gray'
)
self.conn_company_label.pack(anchor=tk.W, pady=(5, 0))
conn_actions = ttk.Frame(self.status_frame)
conn_actions.pack(fill=tk.X, pady=(15, 0))
test_btn = ttk.Button(conn_actions, text="Test Connection", command=self._test_connection, width=15)
test_btn.pack(side=tk.LEFT)
disconnect_btn = ttk.Button(conn_actions, text="Disconnect", command=self._disconnect, width=12)
disconnect_btn.pack(side=tk.LEFT, padx=10)
# Load profiles and saved credentials
self._refresh_profiles()
self._load_saved_credentials()
def _create_templates_tab(self):
"""Create the templates management tab."""
tab = ttk.Frame(self.notebook, padding=20)
self.notebook.add(tab, text=" Templates ")
# Header
header = ttk.Frame(tab)
header.pack(fill=tk.X, pady=(0, 20))
ttk.Label(header, text="Field Mapping Templates", style='Title.TLabel').pack(side=tk.LEFT)
ttk.Button(header, text="+ New Template", command=self._create_template).pack(side=tk.RIGHT)
# Templates list
list_frame = ttk.Frame(tab)
list_frame.pack(fill=tk.BOTH, expand=True)
columns = ('name', 'data_type', 'created', 'mappings')
self.templates_tree = ttk.Treeview(list_frame, columns=columns, show='headings', height=15)
self.templates_tree.heading('name', text='Template Name')
self.templates_tree.heading('data_type', text='Data Type')
self.templates_tree.heading('created', text='Created')
self.templates_tree.heading('mappings', text='Mappings')
templates_scroll = ttk.Scrollbar(list_frame, orient=tk.VERTICAL, command=self.templates_tree.yview)
self.templates_tree.configure(yscrollcommand=templates_scroll.set)
self.templates_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
templates_scroll.pack(side=tk.RIGHT, fill=tk.Y)
# Template actions
actions = ttk.Frame(tab)
actions.pack(fill=tk.X, pady=(15, 0))
ttk.Button(actions, text="Edit", command=self._edit_template).pack(side=tk.LEFT)
ttk.Button(actions, text="Duplicate", command=self._duplicate_template).pack(side=tk.LEFT, padx=5)
ttk.Button(actions, text="Delete", command=self._delete_template).pack(side=tk.LEFT)
ttk.Button(actions, text="Refresh", command=self._refresh_templates).pack(side=tk.RIGHT)
# Load templates
self._refresh_templates()
def _create_lookup_tab(self):
"""Create the QuickBooks Lookup tab."""
tab = ttk.Frame(self.notebook, padding=20)
self.notebook.add(tab, text=" Lookup ")
# Header
header_frame = ttk.Frame(tab)
header_frame.pack(fill=tk.X, pady=(0, 15))
ttk.Label(header_frame, text="QuickBooks Data Lookup", style='Title.TLabel').pack(side=tk.LEFT)
# Connection status
self.lookup_conn_label = ttk.Label(header_frame, text="Not connected", foreground='gray')
self.lookup_conn_label.pack(side=tk.RIGHT)
# Entity type selection
select_frame = ttk.LabelFrame(tab, text="Select Entity Type", padding=10)
select_frame.pack(fill=tk.X, pady=(0, 10))
self.lookup_entity_var = tk.StringVar(value="account")
entity_types = [
("Accounts", "account"),
("Customers", "customer"),
("Vendors", "vendor"),
("Items/Products", "item"),
("Classes", "class"),
("Departments", "department"),
("Payment Terms", "term"),
("Payment Methods", "paymentmethod"),
("Employees", "employee"),
]
entity_grid = ttk.Frame(select_frame)
entity_grid.pack(fill=tk.X)
for i, (label, value) in enumerate(entity_types):
rb = ttk.Radiobutton(
entity_grid,
text=label,
variable=self.lookup_entity_var,
value=value,
command=self._on_lookup_entity_change
)
rb.grid(row=i // 5, column=i % 5, sticky=tk.W, padx=15, pady=3)
# Search frame
search_frame = ttk.Frame(tab)
search_frame.pack(fill=tk.X, pady=10)
ttk.Label(search_frame, text="Search:").pack(side=tk.LEFT)
self.lookup_search_var = tk.StringVar()
self.lookup_search_entry = ttk.Entry(search_frame, textvariable=self.lookup_search_var, width=30)
self.lookup_search_entry.pack(side=tk.LEFT, padx=(5, 15))
self.lookup_search_entry.bind('<Return>', lambda e: self._perform_lookup())
# Type filter for accounts
ttk.Label(search_frame, text="Account Type:").pack(side=tk.LEFT)
self.lookup_type_var = tk.StringVar(value="All Types")
self.lookup_type_combo = ttk.Combobox(
search_frame,
textvariable=self.lookup_type_var,
values=["All Types", "Bank", "Accounts Receivable", "Other Current Asset", "Fixed Asset",
"Accounts Payable", "Credit Card", "Other Current Liability",
"Long Term Liability", "Equity", "Income", "Cost of Goods Sold",
"Expense", "Other Income", "Other Expense"],
width=20,
state='readonly'
)
self.lookup_type_combo.pack(side=tk.LEFT, padx=(5, 15))
ttk.Button(search_frame, text="Search", command=self._perform_lookup, width=12).pack(side=tk.LEFT)
# Results treeview
results_frame = ttk.LabelFrame(tab, text="Results", padding=10)
results_frame.pack(fill=tk.BOTH, expand=True, pady=10)
# Create treeview with scrollbars
tree_container = ttk.Frame(results_frame)
tree_container.pack(fill=tk.BOTH, expand=True)
self.lookup_columns = ['RawId', 'Id', 'Name', 'Type', 'SubType', 'Balance', 'Active']
self.lookup_tree = ttk.Treeview(tree_container, columns=self.lookup_columns, show='headings', height=15)
# Default column configuration
self.lookup_tree.heading('RawId', text='Raw ID')
self.lookup_tree.heading('Id', text='ID')
self.lookup_tree.heading('Name', text='Name')
self.lookup_tree.heading('Type', text='Type')
self.lookup_tree.heading('SubType', text='SubType')
self.lookup_tree.heading('Balance', text='Balance')
self.lookup_tree.heading('Active', text='Active')
self.lookup_tree.column('RawId', width=150, minwidth=100)
self.lookup_tree.column('Id', width=60, minwidth=50)
self.lookup_tree.column('Name', width=200, minwidth=100)
self.lookup_tree.column('Type', width=120, minwidth=80)
self.lookup_tree.column('SubType', width=120, minwidth=80)
self.lookup_tree.column('Balance', width=100, minwidth=60)
self.lookup_tree.column('Active', width=60, minwidth=50)
# Scrollbars
vsb = ttk.Scrollbar(tree_container, orient=tk.VERTICAL, command=self.lookup_tree.yview)
hsb = ttk.Scrollbar(tree_container, orient=tk.HORIZONTAL, command=self.lookup_tree.xview)
self.lookup_tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
self.lookup_tree.grid(row=0, column=0, sticky='nsew')
vsb.grid(row=0, column=1, sticky='ns')
hsb.grid(row=1, column=0, sticky='ew')
tree_container.grid_rowconfigure(0, weight=1)
tree_container.grid_columnconfigure(0, weight=1)
# Double-click to copy
self.lookup_tree.bind('<Double-1>', lambda e: self._copy_lookup_selection())
# Status and action frame
bottom_frame = ttk.Frame(tab)
bottom_frame.pack(fill=tk.X, pady=(10, 0))
self.lookup_status_var = tk.StringVar(value="Select an entity type and click Search")
ttk.Label(bottom_frame, textvariable=self.lookup_status_var, foreground='gray').pack(side=tk.LEFT)
# Action buttons
btn_frame = ttk.Frame(bottom_frame)
btn_frame.pack(side=tk.RIGHT)
ttk.Button(btn_frame, text="Copy Selected", command=self._copy_lookup_selection, width=14).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Export to CSV", command=self._export_lookup_csv, width=14).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Refresh", command=self._perform_lookup, width=10).pack(side=tk.LEFT)
def _on_lookup_entity_change(self):
"""Handle lookup entity type change."""
entity_type = self.lookup_entity_var.get()
# Show/hide account type filter
if entity_type == 'account':
self.lookup_type_combo.config(state='readonly')
else:
self.lookup_type_combo.config(state='disabled')
# Update columns based on entity type
self._update_lookup_columns(entity_type)
# Clear results
for item in self.lookup_tree.get_children():
self.lookup_tree.delete(item)
self.lookup_status_var.set(f"Click Search to load {entity_type}s")
def _update_lookup_columns(self, entity_type: str):
"""Update treeview columns based on entity type."""
# Entity column configurations
column_configs = {
'account': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('AccountType', 'Type', 120),
('AccountSubType', 'SubType', 120),
('CurrentBalance', 'Balance', 100),
('Active', 'Active', 60),
],
'customer': [
('Id', 'ID', 60),
('DisplayName', 'Name', 200),
('CompanyName', 'Company', 150),
('Email', 'Email', 180),
('Phone', 'Phone', 120),
('Balance', 'Balance', 100),
('Active', 'Active', 60),
],
'vendor': [
('Id', 'ID', 60),
('DisplayName', 'Name', 200),
('CompanyName', 'Company', 150),
('Email', 'Email', 180),
('Phone', 'Phone', 120),
('Balance', 'Balance', 100),
('Active', 'Active', 60),
],
'item': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('Type', 'Type', 100),
('Description', 'Description', 250),
('UnitPrice', 'Price', 80),
('Active', 'Active', 60),
],
'class': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('FullyQualifiedName', 'Full Name', 300),
('Active', 'Active', 60),
],
'department': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('FullyQualifiedName', 'Full Name', 300),
('Active', 'Active', 60),
],
'term': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('DueDays', 'Due Days', 100),
('Active', 'Active', 60),
],
'paymentmethod': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('Type', 'Type', 150),
('Active', 'Active', 60),
],
'employee': [
('Id', 'ID', 60),
('DisplayName', 'Name', 200),
('GivenName', 'First Name', 120),
('FamilyName', 'Last Name', 120),
('Active', 'Active', 60),
],
}
config = column_configs.get(entity_type, column_configs['account'])
# Update columns
columns = [col[0] for col in config]
self.lookup_tree['columns'] = columns
for col_key, col_name, col_width in config:
self.lookup_tree.heading(col_key, text=col_name)
self.lookup_tree.column(col_key, width=col_width, minwidth=50)
def _perform_lookup(self):
"""Perform the lookup search."""
if not self.is_connected:
messagebox.showwarning("Not Connected", "Please connect to QuickBooks first.")
return
# Update connection label
conn_type = "Online" if self.connection_type == 'qbo' else "Desktop"
self.lookup_conn_label.config(text=f"Connected: {self.company_name} ({conn_type})", foreground='green')
entity_type = self.lookup_entity_var.get()
search_term = self.lookup_search_var.get().strip()
type_filter = self.lookup_type_var.get() if self.lookup_type_var.get() != "All Types" else ""
# Clear existing results
for item in self.lookup_tree.get_children():
self.lookup_tree.delete(item)
self.lookup_status_var.set("Searching...")
self.root.update()
try:
entities = []
# Entity query configurations
query_config = {
'account': ('Account', 'Name'),
'customer': ('Customer', 'DisplayName'),
'vendor': ('Vendor', 'DisplayName'),
'item': ('Item', 'Name'),
'class': ('Class', 'Name'),
'department': ('Department', 'Name'),
'term': ('Term', 'Name'),
'paymentmethod': ('PaymentMethod', 'Name'),
'employee': ('Employee', 'DisplayName'),
}
query_name, search_field = query_config.get(entity_type, ('Account', 'Name'))
if self.connection_type == 'qbo' and self.qbo_client:
entities = self._fetch_qbo_entities(
query_name,
search_field,
search_term,
type_filter if entity_type == 'account' else None
)
elif self.connection_type == 'qbd' and self.is_connected:
entities = self._fetch_qbd_entities(
entity_type,
search_term,
type_filter if entity_type == 'account' else None
)
# Populate tree
for entity in entities:
values = self._extract_entity_values(entity, entity_type)
self.lookup_tree.insert('', tk.END, values=values)
self.lookup_status_var.set(f"Found {len(entities)} record(s)")
logger.info(f"[LOOKUP] {entity_type}: Found {len(entities)} records")
except Exception as e:
self.lookup_status_var.set(f"Error: {str(e)}")
logger.error(f"Lookup error: {e}")
messagebox.showerror("Lookup Error", f"Failed to fetch data: {str(e)}")
def _extract_entity_values(self, entity: Dict, entity_type: str) -> tuple:
"""Extract values from entity based on type."""
def get_nested(obj, path):
"""Get nested value using dot notation."""
keys = path.split('.')
value = obj
for key in keys:
if isinstance(value, dict):
value = value.get(key)
else:
return None
return value
def format_val(value, is_currency=False):
"""Format value for display."""
if value is None:
return '-'
elif isinstance(value, bool):
return 'Yes' if value else 'No'
elif is_currency:
try:
return f"${float(value):,.2f}"
except:
return str(value)
return str(value)
if entity_type == 'account':
return (
entity.get('RawId', entity.get('Id', '')),
entity.get('Id', ''),
entity.get('Name', ''),
entity.get('AccountType', ''),
entity.get('AccountSubType', ''),
format_val(entity.get('CurrentBalance'), True),
format_val(entity.get('Active')),
)
elif entity_type in ['customer', 'vendor']:
return (
entity.get('RawId', entity.get('Id', '')),
entity.get('Id', ''),
entity.get('DisplayName', ''),
entity.get('CompanyName', ''),
get_nested(entity, 'PrimaryEmailAddr.Address') or '',
get_nested(entity, 'PrimaryPhone.FreeFormNumber') or '',
format_val(entity.get('Balance'), True),
format_val(entity.get('Active')),
)
elif entity_type == 'item':
return (
entity.get('RawId', entity.get('Id', '')),
entity.get('Id', ''),
entity.get('Name', ''),
entity.get('Type', ''),
entity.get('Description', '')[:50] if entity.get('Description') else '',
format_val(entity.get('UnitPrice'), True),
format_val(entity.get('Active')),
)
elif entity_type in ['class', 'department']:
return (
entity.get('RawId', entity.get('Id', '')),
entity.get('Id', ''),
entity.get('Name', ''),
entity.get('FullyQualifiedName', ''),
format_val(entity.get('Active')),
)
elif entity_type == 'term':
return (
entity.get('RawId', entity.get('Id', '')),
entity.get('Id', ''),
entity.get('Name', ''),
entity.get('DueDays', ''),
format_val(entity.get('Active')),
)
elif entity_type == 'paymentmethod':
return (
entity.get('RawId', entity.get('Id', '')),
entity.get('Id', ''),
entity.get('Name', ''),
entity.get('Type', ''),
format_val(entity.get('Active')),
)
elif entity_type == 'employee':
return (
entity.get('RawId', entity.get('Id', '')),
entity.get('Id', ''),
entity.get('DisplayName', ''),
entity.get('GivenName', ''),
entity.get('FamilyName', ''),
format_val(entity.get('Active')),
)
else:
return (entity.get('RawId', entity.get('Id', '')), entity.get('Id', ''), entity.get('Name', ''))
def _copy_lookup_selection(self):
"""Copy selected lookup row to clipboard."""
selection = self.lookup_tree.selection()
if not selection:
messagebox.showinfo("No Selection", "Please select a row to copy.")
return
values = self.lookup_tree.item(selection[0])['values']
# Format as ID: Name (ID is now at index 1, Name at index 2)
text = f"{values[1]}: {values[2]}"
self.root.clipboard_clear()
self.root.clipboard_append(text)
self.lookup_status_var.set(f"Copied to clipboard: {text}")
def _export_lookup_csv(self):
"""Export lookup results to CSV."""
items = self.lookup_tree.get_children()
if not items:
messagebox.showwarning("No Data", "No data to export. Please perform a search first.")
return
from tkinter import filedialog
entity_type = self.lookup_entity_var.get()
filepath = filedialog.asksaveasfilename(
defaultextension=".csv",
filetypes=[("CSV files", "*.csv"), ("All files", "*.*")],
initialfile=f"{entity_type}_export.csv"
)
if not filepath:
return
import csv
columns = self.lookup_tree['columns']
with open(filepath, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
# Header - get display names from headings
headers = [self.lookup_tree.heading(col)['text'] for col in columns]
writer.writerow(headers)
# Data
for item in items:
writer.writerow(self.lookup_tree.item(item)['values'])
messagebox.showinfo("Export Complete", f"Exported {len(items)} records to:\n{filepath}")
logger.info(f"[EXPORT] Exported {len(items)} {entity_type} to {filepath}")
def _create_status_bar(self):
"""Create the status bar."""
self.status_bar = StatusBar(self.root)
self.status_bar.pack(fill=tk.X, side=tk.BOTTOM, padx=10, pady=5)
# =========================================================================
# Connection Methods
# =========================================================================
def _try_restore_connection(self):
"""Try to restore a previous connection."""
creds = self.settings.get_credentials()
if creds and creds.access_token:
try:
self.qbo_client = QBOClient(creds)
if self.qbo_client.is_authenticated:
company_info = self.qbo_client.get_company_info()
self.is_connected = True
self.connection_type = 'qbo'
self.company_name = company_info.get('CompanyName', 'Unknown')
self._update_connection_ui()
logger.info(f"Restored connection to {self.company_name}")
except Exception as e:
logger.warning(f"Could not restore connection: {e}")
def _show_qbo_connect(self):
"""Show QBO connection dialog."""
self.notebook.select(2) # Switch to Connection tab
self.conn_type_var.set('qbo')
self._on_conn_type_change()
def _save_qbo_credentials(self):
"""Save QBO API credentials."""
client_id = self.client_id_entry.get().strip()
client_secret = self.client_secret_entry.get().strip()
redirect_uri = self.redirect_uri_entry.get().strip()
environment = self.env_var.get()
if not client_id or not client_secret:
messagebox.showwarning("Missing Information", "Please enter Client ID and Client Secret.")
return
creds = QBOCredentials(
client_id=client_id,
client_secret=client_secret,
redirect_uri=redirect_uri,
environment=environment
)
self.settings.qbo_environment = environment
self.settings.save_credentials(creds)
messagebox.showinfo("Saved", "Credentials saved successfully.")
logger.info("QBO credentials saved")
def _start_oauth(self):
"""Start the OAuth flow."""
client_id = self.client_id_entry.get().strip()
client_secret = self.client_secret_entry.get().strip()
redirect_uri = self.redirect_uri_entry.get().strip()
if not client_id or not client_secret:
messagebox.showwarning("Missing Information", "Please enter Client ID and Client Secret first.")
return
# Check if using localhost or remote callback
if 'localhost' in redirect_uri or '127.0.0.1' in redirect_uri:
# Local callback - use automatic capture
self._start_oauth_with_callback(client_id, client_secret, redirect_uri)
else:
# Remote callback - open browser and prompt for manual code entry
self._start_oauth_remote_callback(client_id, client_secret, redirect_uri)
def _start_oauth_remote_callback(self, client_id: str, client_secret: str, redirect_uri: str):
"""Start OAuth flow with remote callback server (for production)."""
from urllib.parse import urlencode
# Generate OAuth URL
auth_base = "https://appcenter.intuit.com/connect/oauth2"
params = {
'client_id': client_id,
'response_type': 'code',
'scope': 'com.intuit.quickbooks.accounting',
'redirect_uri': redirect_uri,
'state': 'desktop_app',
'prompt': 'consent',
}
auth_url = f"{auth_base}?{urlencode(params)}"
# Open browser
webbrowser.open(auth_url)
# Show dialog for manual code entry
dialog = tk.Toplevel(self.root)
dialog.title("Complete Authorization")
dialog.geometry("550x350")
dialog.transient(self.root)
dialog.grab_set()
# Center the dialog
dialog.update_idletasks()
x = (dialog.winfo_screenwidth() - 550) // 2
y = (dialog.winfo_screenheight() - 350) // 2
dialog.geometry(f"550x350+{x}+{y}")
ttk.Label(
dialog,
text="Complete the authorization in your browser.\n\n"
"After authorizing, you'll be redirected to your callback server.\n"
"Copy the Authorization Code and Realm ID from that page:",
wraplength=500,
justify=tk.CENTER,
font=('Segoe UI', 10)
).pack(padx=20, pady=20)
# Code entry
code_frame = ttk.Frame(dialog)
code_frame.pack(fill=tk.X, padx=30, pady=5)
ttk.Label(code_frame, text="Authorization Code:", width=18, anchor=tk.W).pack(side=tk.LEFT)
code_entry = ttk.Entry(code_frame, width=40)
code_entry.pack(side=tk.LEFT, padx=10)
# Realm ID entry
realm_frame = ttk.Frame(dialog)
realm_frame.pack(fill=tk.X, padx=30, pady=5)
ttk.Label(realm_frame, text="Realm ID (Company ID):", width=18, anchor=tk.W).pack(side=tk.LEFT)
realm_entry = ttk.Entry(realm_frame, width=40)
realm_entry.pack(side=tk.LEFT, padx=10)
def submit():
code = code_entry.get().strip()
realm_id = realm_entry.get().strip()
if not code or not realm_id:
messagebox.showwarning("Missing Information", "Please enter both the Authorization Code and Realm ID.")
return
dialog.destroy()
self._exchange_code_for_token(code, realm_id)
def reopen_browser():
webbrowser.open(auth_url)
# Buttons
btn_frame = ttk.Frame(dialog)
btn_frame.pack(pady=25)
ttk.Button(btn_frame, text="Connect", command=submit, width=15).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Re-open Browser", command=reopen_browser, width=15).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Cancel", command=dialog.destroy, width=12).pack(side=tk.LEFT, padx=5)
# Help text
ttk.Label(
dialog,
text="Tip: The code expires in 5 minutes. Click 'Re-open Browser' if needed.",
foreground='gray',
font=('Segoe UI', 9)
).pack(pady=(0, 10))
logger.info(f"[OAUTH] Started remote callback OAuth flow, redirect_uri: {redirect_uri}")
def _start_oauth_with_callback(self, client_id: str, client_secret: str, redirect_uri: str):
"""Start OAuth flow with automatic callback capture using a local server."""
import http.server
import socketserver
from urllib.parse import urlencode, urlparse, parse_qs
import socket
# Parse redirect URI to get port
parsed = urlparse(redirect_uri)
port = parsed.port or 9000
# Variables to store callback data
callback_data = {'code': None, 'realm_id': None, 'error': None}
server_instance = [None] # Use list to allow modification in nested function
class OAuthCallbackHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
"""Handle the OAuth callback."""
try:
parsed_path = urlparse(self.path)
params = parse_qs(parsed_path.query)
if 'code' in params:
callback_data['code'] = params['code'][0]
callback_data['realm_id'] = params.get('realmId', [None])[0]
# Send success response
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
success_html = """
<html>
<head><title>Authorization Successful</title></head>
<body style="font-family: Arial, sans-serif; text-align: center; padding: 50px;">
<h1 style="color: #22c55e;">✓ Authorization Successful!</h1>
<p>You can close this window and return to the application.</p>
<script>window.close();</script>
</body>
</html>
"""
self.wfile.write(success_html.encode())
elif 'error' in params:
callback_data['error'] = params.get('error_description', params['error'])[0]
self.send_response(400)
self.send_header('Content-type', 'text/html')
self.end_headers()
error_html = f"""
<html>
<head><title>Authorization Failed</title></head>
<body style="font-family: Arial, sans-serif; text-align: center; padding: 50px;">
<h1 style="color: #ef4444;">✗ Authorization Failed</h1>
<p>{callback_data['error']}</p>
<p>Please close this window and try again.</p>
</body>
</html>
"""
self.wfile.write(error_html.encode())
else:
self.send_response(400)
self.end_headers()
except Exception as e:
callback_data['error'] = str(e)
self.send_response(500)
self.end_headers()
def log_message(self, format, *args):
"""Suppress server logs."""
pass
def run_server():
"""Run the callback server."""
try:
# Allow address reuse
socketserver.TCPServer.allow_reuse_address = True
with socketserver.TCPServer(("", port), OAuthCallbackHandler) as httpd:
server_instance[0] = httpd
httpd.handle_request() # Handle one request then stop
except Exception as e:
callback_data['error'] = f"Server error: {str(e)}"
# Show waiting dialog
dialog = tk.Toplevel(self.root)
dialog.title("Connecting to QuickBooks")
dialog.geometry("450x200")
dialog.transient(self.root)
dialog.grab_set()
dialog.resizable(False, False)
ttk.Label(
dialog,
text="Waiting for authorization...",
font=('Segoe UI', 12, 'bold')
).pack(pady=(30, 10))
ttk.Label(
dialog,
text="A browser window has opened for QuickBooks authorization.\n"
"Please log in and authorize the application.\n\n"
"This window will close automatically when complete.",
wraplength=400,
justify=tk.CENTER
).pack(pady=10)
progress = ttk.Progressbar(dialog, mode='indeterminate', length=300)
progress.pack(pady=20)
progress.start(10)
cancel_clicked = [False]
def cancel():
cancel_clicked[0] = True
if server_instance[0]:
try:
# Create a dummy connection to unblock the server
import socket as sock
s = sock.socket(sock.AF_INET, sock.SOCK_STREAM)
s.settimeout(1)
try:
s.connect(('localhost', port))
s.close()
except:
pass
except:
pass
dialog.destroy()
ttk.Button(dialog, text="Cancel", command=cancel, width=15).pack(pady=10)
# Start server in background thread
import threading
server_thread = threading.Thread(target=run_server, daemon=True)
server_thread.start()
# Generate OAuth URL
environment = self.env_var.get()
auth_base = "https://appcenter.intuit.com/connect/oauth2"
params = {
'client_id': client_id,
'response_type': 'code',
'scope': 'com.intuit.quickbooks.accounting',
'redirect_uri': redirect_uri,
'state': 'desktop_app',
# Force user to select company file every time
# This ensures switching profiles connects to the correct company
'prompt': 'consent',
}
auth_url = f"{auth_base}?{urlencode(params)}"
# Open browser
webbrowser.open(auth_url)
def check_callback():
"""Check if callback was received."""
if cancel_clicked[0]:
return
if callback_data['code']:
# Success - got the code
dialog.destroy()
self._exchange_code_for_token(callback_data['code'], callback_data['realm_id'])
elif callback_data['error']:
# Error
dialog.destroy()
messagebox.showerror("Authorization Failed", callback_data['error'])
elif server_thread.is_alive():
# Still waiting
self.root.after(100, check_callback)
else:
# Server stopped without getting code
if not cancel_clicked[0]:
dialog.destroy()
messagebox.showerror("Authorization Failed", "No authorization response received.")
# Start checking for callback
self.root.after(100, check_callback)
def _show_auth_code_dialog(self):
"""Show dialog to enter authorization code manually (fallback)."""
dialog = tk.Toplevel(self.root)
dialog.title("Enter Authorization Code Manually")
dialog.geometry("500x250")
dialog.transient(self.root)
dialog.grab_set()
ttk.Label(
dialog,
text="If automatic capture didn't work, you can enter the\n"
"authorization code and Realm ID manually:",
wraplength=450,
justify=tk.CENTER
).pack(padx=20, pady=20)
code_frame = ttk.Frame(dialog)
code_frame.pack(fill=tk.X, padx=20, pady=5)
ttk.Label(code_frame, text="Authorization Code:", width=18, anchor=tk.W).pack(side=tk.LEFT)
code_entry = ttk.Entry(code_frame, width=40)
code_entry.pack(side=tk.LEFT, padx=10)
realm_frame = ttk.Frame(dialog)
realm_frame.pack(fill=tk.X, padx=20, pady=5)
ttk.Label(realm_frame, text="Realm ID:", width=18, anchor=tk.W).pack(side=tk.LEFT)
realm_entry = ttk.Entry(realm_frame, width=40)
realm_entry.pack(side=tk.LEFT, padx=10)
def submit():
code = code_entry.get().strip()
realm_id = realm_entry.get().strip()
if not code or not realm_id:
messagebox.showwarning("Missing Information", "Please enter both the authorization code and Realm ID.")
return
dialog.destroy()
self._exchange_code_for_token(code, realm_id)
btn_frame = ttk.Frame(dialog)
btn_frame.pack(pady=20)
ttk.Button(btn_frame, text="Connect", command=submit, width=12).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Cancel", command=dialog.destroy, width=12).pack(side=tk.LEFT, padx=5)
def _exchange_code_for_token(self, code: str, realm_id: str):
"""Exchange authorization code for access token."""
import requests
from base64 import b64encode
client_id = self.client_id_entry.get().strip()
client_secret = self.client_secret_entry.get().strip()
redirect_uri = self.redirect_uri_entry.get().strip()
# Token endpoint
token_url = "https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer"
# Prepare request
auth_string = b64encode(f"{client_id}:{client_secret}".encode()).decode()
headers = {
'Authorization': f'Basic {auth_string}',
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/json'
}
data = {
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': redirect_uri
}
try:
response = requests.post(token_url, headers=headers, data=data, timeout=30)
if response.status_code == 200:
token_data = response.json()
# Save credentials with environment setting
environment = self.env_var.get()
creds = QBOCredentials(
client_id=client_id,
client_secret=client_secret,
redirect_uri=redirect_uri,
environment=environment,
access_token=token_data.get('access_token'),
refresh_token=token_data.get('refresh_token'),
realm_id=realm_id,
token_expiry=datetime.now().isoformat()
)
self.settings.save_credentials(creds)
# Connect
self.qbo_client = QBOClient(creds)
company_info = self.qbo_client.get_company_info()
self.is_connected = True
self.connection_type = 'qbo'
self.company_name = company_info.get('CompanyName', 'Unknown')
self._update_connection_ui()
# Update current profile with tokens if one is selected
profile_name = self.profile_var.get() if hasattr(self, 'profile_var') else None
if profile_name:
profile = ConnectionProfile(
name=profile_name,
client_id=client_id,
client_secret=client_secret,
redirect_uri=redirect_uri,
environment=self.env_var.get(),
access_token=token_data.get('access_token'),
refresh_token=token_data.get('refresh_token'),
realm_id=realm_id,
company_name=self.company_name,
token_expiry=datetime.now().isoformat(),
)
self.settings.save_profile(profile)
logger.info(f"[PROFILE] Updated profile '{profile_name}' with new tokens")
messagebox.showinfo("Connected", f"Successfully connected to {self.company_name}")
logger.info(f"Connected to QBO: {self.company_name}")
else:
error_msg = response.json().get('error_description', response.text)
messagebox.showerror("Connection Failed", f"Failed to get access token: {error_msg}")
logger.error(f"OAuth token exchange failed: {error_msg}")
except Exception as e:
messagebox.showerror("Error", f"Connection error: {str(e)}")
logger.error(f"OAuth error: {e}")
def _enter_token_manually(self):
"""Show dialog to enter token manually."""
dialog = tk.Toplevel(self.root)
dialog.title("Enter Token Manually")
dialog.geometry("500x300")
dialog.transient(self.root)
dialog.grab_set()
ttk.Label(dialog, text="Access Token:").pack(padx=20, pady=(20, 5), anchor=tk.W)
access_entry = ttk.Entry(dialog, width=60)
access_entry.pack(padx=20, pady=5)
ttk.Label(dialog, text="Refresh Token:").pack(padx=20, pady=(10, 5), anchor=tk.W)
refresh_entry = ttk.Entry(dialog, width=60)
refresh_entry.pack(padx=20, pady=5)
ttk.Label(dialog, text="Realm ID (Company ID):").pack(padx=20, pady=(10, 5), anchor=tk.W)
realm_entry = ttk.Entry(dialog, width=60)
realm_entry.pack(padx=20, pady=5)
def submit():
access_token = access_entry.get().strip()
refresh_token = refresh_entry.get().strip()
realm_id = realm_entry.get().strip()
if not access_token or not realm_id:
messagebox.showwarning("Missing Information", "Please enter Access Token and Realm ID.")
return
creds = QBOCredentials(
client_id=self.client_id_entry.get().strip(),
client_secret=self.client_secret_entry.get().strip(),
redirect_uri=self.redirect_uri_entry.get().strip(),
environment=self.env_var.get(),
access_token=access_token,
refresh_token=refresh_token,
realm_id=realm_id,
token_expiry=datetime.now().isoformat()
)
try:
self.settings.save_credentials(creds)
self.qbo_client = QBOClient(creds)
company_info = self.qbo_client.get_company_info()
self.is_connected = True
self.connection_type = 'qbo'
self.company_name = company_info.get('CompanyName', 'Unknown')
self._update_connection_ui()
dialog.destroy()
messagebox.showinfo("Connected", f"Successfully connected to {self.company_name}")
except Exception as e:
messagebox.showerror("Error", f"Connection failed: {str(e)}")
ttk.Button(dialog, text="Connect", command=submit).pack(pady=20)
def _connect_qbd(self):
"""Connect to QuickBooks Desktop."""
if not HAS_QBD:
messagebox.showwarning(
"Not Available",
"QuickBooks Desktop integration is not available.\n"
"This feature requires Windows and pywin32."
)
return
self.qbd_connect_btn.config(state=tk.DISABLED, text="Connecting...")
self.root.update()
def connect_thread():
try:
# Run connection in subprocess to avoid COM threading 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 Desktop", 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)
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)
'''
# Get Python executable (handles PyInstaller bundled case)
python_exe = self._get_python_executable()
result = subprocess.run(
[python_exe, '-c', connect_script],
capture_output=True,
text=True,
timeout=60
)
if result.returncode == 0:
output = json.loads(result.stdout.strip())
if output.get('success'):
self.root.after(0, lambda: self._qbd_connect_success(output.get('company_name', 'QuickBooks Desktop')))
else:
self.root.after(0, lambda: self._qbd_connect_error(output.get('error', 'Unknown error')))
else:
self.root.after(0, lambda: self._qbd_connect_error(result.stderr or result.stdout))
except subprocess.TimeoutExpired:
self.root.after(0, lambda: self._qbd_connect_error("Connection timed out. Check QuickBooks Desktop."))
except Exception as e:
self.root.after(0, lambda: self._qbd_connect_error(str(e)))
threading.Thread(target=connect_thread, daemon=True).start()
def _qbd_connect_success(self, company_name: str):
"""Handle successful QBD connection."""
self.is_connected = True
self.connection_type = 'qbd'
self.company_name = company_name
self._update_connection_ui()
self.qbd_connect_btn.config(state=tk.NORMAL, text="Connect to QuickBooks Desktop")
messagebox.showinfo("Connected", f"Successfully connected to {company_name}")
logger.info(f"Connected to QBD: {company_name}")
def _qbd_connect_error(self, error: str):
"""Handle QBD connection error."""
self.qbd_connect_btn.config(state=tk.NORMAL, text="Connect to QuickBooks Desktop")
messagebox.showerror("Connection Failed", f"Failed to connect: {error}")
logger.error(f"QBD connection failed: {error}")
def _disconnect(self):
"""Disconnect from QuickBooks."""
if not self.is_connected:
return
if messagebox.askyesno("Confirm", "Are you sure you want to disconnect?"):
self.is_connected = False
self.qbo_client = None
self.qbd_client = None
self.company_name = ""
self._update_connection_ui()
logger.info("Disconnected from QuickBooks")
def _test_connection(self):
"""Test the current connection."""
if not self.is_connected:
messagebox.showwarning("Not Connected", "Please connect to QuickBooks first.")
return
try:
if self.connection_type == 'qbo' and self.qbo_client:
company_info = self.qbo_client.get_company_info()
customers = self.qbo_client.get_customers()
messagebox.showinfo(
"Connection OK",
f"Connected to: {company_info.get('CompanyName', 'Unknown')}\n"
f"Customers: {len(customers)}"
)
elif self.connection_type == 'qbd':
messagebox.showinfo("Connection OK", f"Connected to: {self.company_name}")
except Exception as e:
messagebox.showerror("Connection Error", f"Connection test failed: {str(e)}")
def _update_connection_ui(self):
"""Update all UI elements to reflect connection status."""
if self.is_connected:
# Status bar
self.status_bar.set_connected(self.company_name, self.connection_type)
# Dashboard card
self.conn_card.value_label.config(text="Connected")
self.conn_card.detail_label.config(text=self.company_name)
# Connection tab
type_text = "Desktop" if self.connection_type == 'qbd' else "Online"
self.conn_status_label.config(text=f"● Connected ({type_text})", foreground='green')
self.conn_company_label.config(text=self.company_name)
else:
# Status bar
self.status_bar.set_disconnected()
# Dashboard card
self.conn_card.value_label.config(text="Not Connected")
self.conn_card.detail_label.config(text="Connect to QuickBooks to start")
# Connection tab
self.conn_status_label.config(text="● Not Connected", foreground='red')
self.conn_company_label.config(text="")
def _on_conn_type_change(self):
"""Handle connection type change."""
conn_type = self.conn_type_var.get()
if conn_type == 'qbo':
# Show QBO frame, hide QBD frame
self.qbd_frame.pack_forget()
self.qbo_frame.pack(fill=tk.X, pady=10, before=self.status_frame)
else:
# Show QBD frame, hide QBO frame
self.qbo_frame.pack_forget()
self.qbd_frame.pack(fill=tk.X, pady=10, before=self.status_frame)
def _load_saved_credentials(self):
"""Load saved credentials into the form."""
creds = self.settings.get_credentials()
if creds:
self.client_id_entry.delete(0, tk.END)
self.client_id_entry.insert(0, creds.client_id or '')
self.client_secret_entry.delete(0, tk.END)
self.client_secret_entry.insert(0, creds.client_secret or '')
if creds.redirect_uri:
self.redirect_uri_entry.delete(0, tk.END)
self.redirect_uri_entry.insert(0, creds.redirect_uri)
# =========================================================================
# Profile Management Methods
# =========================================================================
def _refresh_profiles(self):
"""Refresh the profile dropdown list."""
profiles = self.settings.get_profiles()
profile_names = [p.name for p in profiles]
self.profile_combo['values'] = profile_names
# Select active profile if exists
active_profile = self.settings.get_active_profile_name()
if active_profile and active_profile in profile_names:
self.profile_var.set(active_profile)
elif profile_names:
self.profile_var.set(profile_names[0])
else:
self.profile_var.set("")
def _on_profile_change(self, event=None):
"""Handle profile selection change."""
# Just update the selection, don't auto-load
pass
def _load_selected_profile(self):
"""Load the selected profile into the form."""
profile_name = self.profile_var.get()
if not profile_name:
messagebox.showwarning("No Profile", "Please select a profile to load.")
return
profile = self.settings.get_profile(profile_name)
if not profile:
messagebox.showerror("Error", f"Profile '{profile_name}' not found.")
return
# Load profile data into form
self.client_id_entry.delete(0, tk.END)
self.client_id_entry.insert(0, profile.client_id or '')
self.client_secret_entry.delete(0, tk.END)
self.client_secret_entry.insert(0, profile.client_secret or '')
self.redirect_uri_entry.delete(0, tk.END)
self.redirect_uri_entry.insert(0, profile.redirect_uri or 'http://localhost:9000/oauth/callback')
self.env_var.set(profile.environment or 'sandbox')
# Set as active profile
self.settings.set_active_profile(profile_name)
# If profile has tokens, try to connect
if profile.access_token and profile.realm_id:
try:
creds = QBOCredentials(
client_id=profile.client_id,
client_secret=profile.client_secret,
redirect_uri=profile.redirect_uri,
environment=profile.environment or 'sandbox',
access_token=profile.access_token,
refresh_token=profile.refresh_token,
realm_id=profile.realm_id,
token_expiry=profile.token_expiry
)
self.settings.save_credentials(creds)
self.qbo_client = QBOClient(creds)
if self.qbo_client.is_authenticated:
company_info = self.qbo_client.get_company_info()
self.is_connected = True
self.connection_type = 'qbo'
self.company_name = company_info.get('CompanyName', profile.company_name or 'Unknown')
self._update_connection_ui()
logger.info(f"[PROFILE] Connected using profile: {profile_name}")
messagebox.showinfo("Profile Loaded", f"Connected to {self.company_name}")
else:
messagebox.showinfo("Profile Loaded", f"Profile '{profile_name}' loaded.\nClick 'Connect to QuickBooks' to authenticate.")
except Exception as e:
logger.warning(f"Could not auto-connect with profile: {e}")
messagebox.showinfo("Profile Loaded", f"Profile '{profile_name}' loaded.\nTokens expired. Please reconnect.")
else:
messagebox.showinfo("Profile Loaded", f"Profile '{profile_name}' loaded.\nClick 'Connect to QuickBooks' to authenticate.")
logger.info(f"[PROFILE] Loaded profile: {profile_name}")
def _save_profile(self):
"""Save current settings to the selected profile."""
profile_name = self.profile_var.get()
if not profile_name:
# No profile selected, prompt for new name
self._save_profile_as()
return
# Confirm overwrite
if not messagebox.askyesno("Save Profile", f"Overwrite profile '{profile_name}'?"):
return
self._save_profile_with_name(profile_name)
def _save_profile_as(self):
"""Save current settings as a new profile."""
# Ask for profile name
name = tk.simpledialog.askstring(
"Save Profile As",
"Enter a name for this profile:",
parent=self.root
)
if not name:
return
name = name.strip()
if not name:
messagebox.showwarning("Invalid Name", "Please enter a valid profile name.")
return
# Check if profile already exists
existing = self.settings.get_profile(name)
if existing:
if not messagebox.askyesno("Profile Exists", f"Profile '{name}' already exists. Overwrite?"):
return
self._save_profile_with_name(name)
# Update dropdown
self._refresh_profiles()
self.profile_var.set(name)
def _save_profile_with_name(self, name: str):
"""Save profile with the given name."""
# Get current credentials if connected
creds = self.settings.get_credentials()
profile = ConnectionProfile(
name=name,
client_id=self.client_id_entry.get().strip(),
client_secret=self.client_secret_entry.get().strip(),
redirect_uri=self.redirect_uri_entry.get().strip(),
environment=self.env_var.get(),
access_token=creds.access_token if creds else None,
refresh_token=creds.refresh_token if creds else None,
realm_id=creds.realm_id if creds else None,
company_name=self.company_name if self.is_connected else None,
token_expiry=creds.token_expiry if creds else None,
)
self.settings.save_profile(profile)
self.settings.set_active_profile(name)
messagebox.showinfo("Profile Saved", f"Profile '{name}' saved successfully.")
logger.info(f"[PROFILE] Saved profile: {name}")
# Refresh the dropdown
self._refresh_profiles()
def _delete_profile(self):
"""Delete the selected profile."""
profile_name = self.profile_var.get()
if not profile_name:
messagebox.showwarning("No Profile", "Please select a profile to delete.")
return
if not messagebox.askyesno("Delete Profile", f"Are you sure you want to delete profile '{profile_name}'?"):
return
self.settings.delete_profile(profile_name)
# Clear active profile if it was the deleted one
if self.settings.get_active_profile_name() == profile_name:
self.settings.set_active_profile(None)
# Refresh dropdown
self._refresh_profiles()
# Clear form if needed
if not self.profile_var.get():
self.client_id_entry.delete(0, tk.END)
self.client_secret_entry.delete(0, tk.END)
messagebox.showinfo("Profile Deleted", f"Profile '{profile_name}' deleted.")
logger.info(f"[PROFILE] Deleted profile: {profile_name}")
def _import_profile(self):
"""Import profile(s) from a JSON file."""
file_path = filedialog.askopenfilename(
title="Import Profile",
filetypes=[
("JSON Files", "*.json"),
("All Files", "*.*")
]
)
if not file_path:
return
try:
with open(file_path, 'r') as f:
data = json.load(f)
# Handle both single profile and multiple profiles
if isinstance(data, list):
profiles_data = data
elif isinstance(data, dict):
profiles_data = [data]
else:
raise ValueError("Invalid profile format")
imported_count = 0
skipped_count = 0
for profile_data in profiles_data:
# Validate required fields
if 'name' not in profile_data:
skipped_count += 1
continue
profile_name = profile_data['name']
# Check if profile already exists
existing = self.settings.get_profile(profile_name)
if existing:
overwrite = messagebox.askyesno(
"Profile Exists",
f"Profile '{profile_name}' already exists. Overwrite?"
)
if not overwrite:
skipped_count += 1
continue
# Create ConnectionProfile from data
profile = ConnectionProfile(
name=profile_data.get('name', ''),
client_id=profile_data.get('client_id', ''),
client_secret=profile_data.get('client_secret', ''),
redirect_uri=profile_data.get('redirect_uri', 'http://localhost:9000/oauth/callback'),
environment=profile_data.get('environment', 'sandbox'),
access_token=profile_data.get('access_token'),
refresh_token=profile_data.get('refresh_token'),
realm_id=profile_data.get('realm_id'),
company_name=profile_data.get('company_name'),
token_expiry=profile_data.get('token_expiry'),
created_at=profile_data.get('created_at', datetime.now().isoformat()),
updated_at=datetime.now().isoformat(),
)
self.settings.save_profile(profile)
imported_count += 1
logger.info(f"[PROFILE] Imported profile: {profile_name}")
# Refresh the dropdown
self._refresh_profiles()
# Show result
msg = f"Imported {imported_count} profile(s)."
if skipped_count > 0:
msg += f"\nSkipped {skipped_count} profile(s)."
messagebox.showinfo("Import Complete", msg)
except json.JSONDecodeError as e:
messagebox.showerror("Import Error", f"Invalid JSON file: {str(e)}")
logger.error(f"[PROFILE] Import failed - invalid JSON: {e}")
except Exception as e:
messagebox.showerror("Import Error", f"Failed to import profile: {str(e)}")
logger.error(f"[PROFILE] Import failed: {e}")
def _export_profile(self):
"""Export the selected profile to a JSON file."""
profile_name = self.profile_var.get()
if not profile_name:
messagebox.showwarning("No Profile", "Please select a profile to export.")
return
profile = self.settings.get_profile(profile_name)
if not profile:
messagebox.showerror("Error", f"Profile '{profile_name}' not found.")
return
# Ask whether to include tokens (sensitive data)
include_tokens = messagebox.askyesno(
"Include Tokens?",
"Do you want to include access tokens in the export?\n\n"
"• Yes - Include tokens (allows immediate connection on import)\n"
"• No - Exclude tokens (more secure, requires re-authentication)"
)
# Prepare export data
from dataclasses import asdict
export_data = asdict(profile)
if not include_tokens:
export_data['access_token'] = None
export_data['refresh_token'] = None
export_data['token_expiry'] = None
# Get save location
safe_name = "".join(c for c in profile_name if c.isalnum() or c in " -_").strip()
file_path = filedialog.asksaveasfilename(
title="Export Profile",
defaultextension=".json",
filetypes=[("JSON Files", "*.json"), ("All Files", "*.*")],
initialfile=f"qbo_profile_{safe_name}.json"
)
if not file_path:
return
try:
with open(file_path, 'w') as f:
json.dump(export_data, f, indent=2)
messagebox.showinfo("Export Complete", f"Profile '{profile_name}' exported successfully.")
logger.info(f"[PROFILE] Exported profile: {profile_name} to {file_path}")
except Exception as e:
messagebox.showerror("Export Error", f"Failed to export profile: {str(e)}")
logger.error(f"[PROFILE] Export failed: {e}")
def _export_all_profiles(self):
"""Export all profiles to a single JSON file."""
profiles = self.settings.get_profiles()
if not profiles:
messagebox.showwarning("No Profiles", "No profiles to export.")
return
# Ask whether to include tokens (sensitive data)
include_tokens = messagebox.askyesno(
"Include Tokens?",
f"Exporting {len(profiles)} profile(s).\n\n"
"Do you want to include access tokens in the export?\n\n"
"• Yes - Include tokens (allows immediate connection on import)\n"
"• No - Exclude tokens (more secure, requires re-authentication)"
)
# Prepare export data
from dataclasses import asdict
export_data = []
for profile in profiles:
profile_dict = asdict(profile)
if not include_tokens:
profile_dict['access_token'] = None
profile_dict['refresh_token'] = None
profile_dict['token_expiry'] = None
export_data.append(profile_dict)
# Get save location
file_path = filedialog.asksaveasfilename(
title="Export All Profiles",
defaultextension=".json",
filetypes=[("JSON Files", "*.json"), ("All Files", "*.*")],
initialfile=f"qbo_profiles_backup_{datetime.now().strftime('%Y%m%d')}.json"
)
if not file_path:
return
try:
with open(file_path, 'w') as f:
json.dump(export_data, f, indent=2)
messagebox.showinfo(
"Export Complete",
f"Exported {len(profiles)} profile(s) successfully."
)
logger.info(f"[PROFILE] Exported {len(profiles)} profiles to {file_path}")
except Exception as e:
messagebox.showerror("Export Error", f"Failed to export profiles: {str(e)}")
logger.error(f"[PROFILE] Export all failed: {e}")
# =========================================================================
# Import Methods
# =========================================================================
def _open_file(self):
"""Open file dialog to select Excel file."""
file_path = filedialog.askopenfilename(
title="Select Excel File",
filetypes=[
("Excel Files", "*.xlsx *.xls"),
("All Files", "*.*")
]
)
if file_path:
self.current_file = Path(file_path)
self.file_entry.delete(0, tk.END)
self.file_entry.insert(0, str(self.current_file))
self._parse_file()
def _refresh_file(self):
"""Refresh the current file."""
if self.current_file and self.current_file.exists():
self._parse_file()
else:
messagebox.showwarning("No File", "Please select a file first.")
def _parse_file(self):
"""Parse the selected Excel file."""
if not self.current_file or not self.current_file.exists():
return
try:
parser = ExcelParser()
# Use preview to load data (returns columns and list of row dicts)
self.excel_columns, self.excel_data = parser.preview(
str(self.current_file),
max_rows=1000 # Load more rows for actual import
)
# Store parser for later use
self.parser = parser
# Update file info
row_count = len(self.excel_data)
self.file_info_label.config(
text=f"Loaded {row_count} rows from {self.current_file.name}",
foreground='green'
)
# Update preview
self._update_preview()
# Update mapping
self._update_mapping()
logger.info(f"Parsed file: {self.current_file.name} ({row_count} rows)")
except Exception as e:
self.file_info_label.config(text=f"Error: {str(e)}", foreground='red')
logger.error(f"Error parsing file: {e}")
messagebox.showerror("Parse Error", f"Failed to parse file: {str(e)}")
def _update_preview(self):
"""Update the data preview tree."""
# Clear existing
for item in self.preview_tree.get_children():
self.preview_tree.delete(item)
if not hasattr(self, 'excel_data') or not self.excel_data:
return
# Get columns
columns = self.excel_columns if hasattr(self, 'excel_columns') else []
if not columns and self.excel_data:
# Fallback: get columns from first row
columns = list(self.excel_data[0].keys())
# Configure columns
self.preview_tree['columns'] = columns
for col in columns:
self.preview_tree.heading(col, text=col)
self.preview_tree.column(col, width=100)
# Add rows (limit to 100 for performance)
for row in self.excel_data[:100]:
values = [row.get(col, '') for col in columns]
self.preview_tree.insert('', tk.END, values=values)
def _update_mapping(self):
"""Update the field mapping tree."""
# Clear existing
for item in self.mapping_tree.get_children():
self.mapping_tree.delete(item)
if not hasattr(self, 'excel_columns') or not self.excel_columns:
return
# Add to tree
for col in self.excel_columns:
qbo_field = self.field_mappings.get(col, '(unmapped)')
self.mapping_tree.insert('', tk.END, values=(col, qbo_field))
def _on_data_type_change(self):
"""Handle data type selection change."""
# Refresh template list for new data type
self._refresh_template_list()
# Reset to default template
if hasattr(self, 'template_combo'):
self.template_combo.current(0)
# Re-parse file with new data type
if self.current_file:
self._parse_file()
def _refresh_template_list(self):
"""Refresh the template dropdown with available templates."""
if not hasattr(self, 'template_combo'):
return
data_type = self.data_type_var.get()
# Get templates for current data type
all_templates = self.settings.get_templates(data_type)
# Build list of template names
template_names = ["(Default Template)"]
for t in all_templates:
if hasattr(t, 'name'):
template_names.append(t.name)
self.template_combo['values'] = template_names
self.template_combo.current(0)
def _on_template_change(self, event=None):
"""Handle template selection change."""
selected = self.template_var.get()
if selected == "(Default Template)" or not selected:
# Use default template - run auto-map
if hasattr(self, 'excel_columns') and self.excel_columns:
self._auto_map_fields()
else:
# Load selected template
self._load_template(selected)
def _load_template(self, template_name: str):
"""Load a saved template."""
templates = self.settings.get_templates()
template = None
for t in templates:
if hasattr(t, 'name') and t.name == template_name:
template = t
break
if not template:
messagebox.showwarning("Template Not Found", f"Template '{template_name}' not found.")
return
# Convert template mappings to field_mappings dict
self.field_mappings = {}
if not hasattr(self, 'excel_columns') or not self.excel_columns:
# No file loaded yet, just store template info
for mapping in template.mappings:
if hasattr(mapping, 'excel_column'):
self.field_mappings[mapping.excel_column] = mapping.qbo_field
else:
# Match template columns to actual Excel columns
excel_columns_lower = {col.lower().replace(' ', '').replace('_', ''): col for col in self.excel_columns}
for mapping in template.mappings:
if hasattr(mapping, 'excel_column'):
template_col = mapping.excel_column
template_col_lower = template_col.lower().replace(' ', '').replace('_', '')
# Try exact match
if template_col in self.excel_columns:
self.field_mappings[template_col] = mapping.qbo_field
# Try normalized match
elif template_col_lower in excel_columns_lower:
actual_col = excel_columns_lower[template_col_lower]
self.field_mappings[actual_col] = mapping.qbo_field
self._update_mapping()
logger.info(f"[EDIT] Loaded template: {template_name} ({len(self.field_mappings)} mappings)")
def _auto_map_fields(self):
"""Automatically map fields based on column names using default templates."""
if not hasattr(self, 'excel_columns') or not self.excel_columns:
messagebox.showwarning("No Data", "Please load a file first.")
return
data_type = self.data_type_var.get()
excel_columns = self.excel_columns
# Get default template mappings from settings
default_templates = self.settings.get_default_templates()
template = default_templates.get(data_type)
if template:
# Use template mappings - match Excel columns to template's excel_column names
self.field_mappings = {}
excel_columns_lower = {col.lower().replace(' ', '').replace('_', ''): col for col in excel_columns}
for mapping in template.mappings:
# Try to find matching Excel column
template_col_lower = mapping.excel_column.lower().replace(' ', '').replace('_', '')
# Direct match
if template_col_lower in excel_columns_lower:
self.field_mappings[excel_columns_lower[template_col_lower]] = mapping.qbo_field
else:
# Partial match
for excel_col_lower, excel_col in excel_columns_lower.items():
if template_col_lower in excel_col_lower or excel_col_lower in template_col_lower:
self.field_mappings[excel_col] = mapping.qbo_field
break
else:
# Fallback to simple common mappings
common_mappings = {
'check': {
'date': 'TxnDate',
'txndate': 'TxnDate',
'payee': 'EntityRef.value',
'vendor': 'EntityRef.value',
'amount': 'Line.Amount',
'total': 'Line.Amount',
'bankacct': 'AccountRef.value',
'bankaccount': 'AccountRef.value',
'account': 'AccountRef.value',
'postingacct': 'Line.AccountBasedExpenseLineDetail.AccountRef.value',
'expenseaccount': 'Line.AccountBasedExpenseLineDetail.AccountRef.value',
'memo': 'PrivateNote',
'periodmemo': 'PrivateNote',
'checknumber': 'DocNumber',
'checkno': 'DocNumber',
'number': 'DocNumber',
},
'invoice': {
'date': 'TxnDate',
'txndate': 'TxnDate',
'customer': 'CustomerRef.value',
'customerref': 'CustomerRef.value',
'amount': 'Line.Amount',
'item': 'Line.SalesItemLineDetail.ItemRef.value',
'description': 'Line.Description',
'quantity': 'Line.SalesItemLineDetail.Qty',
'qty': 'Line.SalesItemLineDetail.Qty',
'rate': 'Line.SalesItemLineDetail.UnitPrice',
'unitprice': 'Line.SalesItemLineDetail.UnitPrice',
'duedate': 'DueDate',
'docnumber': 'DocNumber',
'invoicenumber': 'DocNumber',
},
'bill': {
'date': 'TxnDate',
'txndate': 'TxnDate',
'vendor': 'VendorRef.value',
'vendorref': 'VendorRef.value',
'amount': 'Line.Amount',
'account': 'Line.AccountBasedExpenseLineDetail.AccountRef.value',
'expenseaccount': 'Line.AccountBasedExpenseLineDetail.AccountRef.value',
'duedate': 'DueDate',
'memo': 'PrivateNote',
'docnumber': 'DocNumber',
},
'customer': {
'name': 'DisplayName',
'displayname': 'DisplayName',
'companyname': 'CompanyName',
'firstname': 'GivenName',
'givenname': 'GivenName',
'lastname': 'FamilyName',
'familyname': 'FamilyName',
'email': 'PrimaryEmailAddr.Address',
'phone': 'PrimaryPhone.FreeFormNumber',
},
'vendor': {
'name': 'DisplayName',
'displayname': 'DisplayName',
'companyname': 'CompanyName',
'firstname': 'GivenName',
'givenname': 'GivenName',
'lastname': 'FamilyName',
'familyname': 'FamilyName',
'email': 'PrimaryEmailAddr.Address',
'phone': 'PrimaryPhone.FreeFormNumber',
},
'account': {
'name': 'Name',
'accountname': 'Name',
'type': 'AccountType',
'accounttype': 'AccountType',
'subtype': 'AccountSubType',
'number': 'AcctNum',
'accountnumber': 'AcctNum',
'description': 'Description',
},
}
mappings = common_mappings.get(data_type, {})
self.field_mappings = {}
for col in excel_columns:
col_lower = col.lower().replace(' ', '').replace('_', '')
for pattern, qbo_field in mappings.items():
if pattern in col_lower or col_lower in pattern:
self.field_mappings[col] = qbo_field
break
self._update_mapping()
mapped_count = len(self.field_mappings)
total_count = len(excel_columns)
messagebox.showinfo("Auto-Map Complete", f"Mapped {mapped_count} of {total_count} fields.")
def _edit_mapping(self):
"""Open mapping editor dialog."""
if not hasattr(self, 'excel_columns') or not self.excel_columns:
messagebox.showwarning("No Data", "Please load a file first.")
return
# Create mapping editor dialog
dialog = tk.Toplevel(self.root)
dialog.title("Edit Field Mapping")
dialog.geometry("700x500")
dialog.transient(self.root)
dialog.grab_set()
ttk.Label(dialog, text="Map Excel columns to QuickBooks fields:", font=('Segoe UI', 10)).pack(pady=10)
# Scrollable frame for mappings
canvas = tk.Canvas(dialog)
scrollbar = ttk.Scrollbar(dialog, orient=tk.VERTICAL, command=canvas.yview)
scrollable_frame = ttk.Frame(canvas)
scrollable_frame.bind(
"<Configure>",
lambda e: canvas.configure(scrollregion=canvas.bbox("all"))
)
canvas.create_window((0, 0), window=scrollable_frame, anchor=tk.NW)
canvas.configure(yscrollcommand=scrollbar.set)
# Get QBO fields for this data type
qbo_fields = self._get_qbo_fields(self.data_type_var.get())
# Create mapping entries
excel_columns = self.excel_columns
mapping_vars = {}
for i, col in enumerate(excel_columns):
frame = ttk.Frame(scrollable_frame)
frame.pack(fill=tk.X, padx=10, pady=2)
ttk.Label(frame, text=col, width=25, anchor=tk.W).pack(side=tk.LEFT)
ttk.Label(frame, text="→").pack(side=tk.LEFT, padx=10)
var = tk.StringVar(value=self.field_mappings.get(col, ''))
combo = ttk.Combobox(frame, textvariable=var, values=[''] + qbo_fields, width=40)
combo.pack(side=tk.LEFT, fill=tk.X, expand=True)
mapping_vars[col] = var
canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=10)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
def save_mappings():
self.field_mappings = {col: var.get() for col, var in mapping_vars.items() if var.get()}
self._update_mapping()
dialog.destroy()
logger.info(f"[EDIT] Field mappings updated: {len(self.field_mappings)} fields mapped")
btn_frame = ttk.Frame(dialog)
btn_frame.pack(fill=tk.X, pady=10)
ttk.Button(btn_frame, text="Save", command=save_mappings).pack(side=tk.RIGHT, padx=10)
ttk.Button(btn_frame, text="Cancel", command=dialog.destroy).pack(side=tk.RIGHT)
def _get_qbo_fields(self, data_type: str) -> List[str]:
"""Get available QBO fields for a data type from default templates."""
# Try to get fields from default templates
default_templates = self.settings.get_default_templates()
template = default_templates.get(data_type)
if template:
# Extract unique QBO fields from template
return list(set(m.qbo_field for m in template.mappings))
# Fallback to predefined fields with dot notation
fields = {
'check': [
'TxnDate',
'EntityRef.value',
'AccountRef.value',
'Line.Amount',
'Line.AccountBasedExpenseLineDetail.AccountRef.value',
'DocNumber',
'PrivateNote',
'Line.Description',
],
'invoice': [
'TxnDate',
'CustomerRef.value',
'DueDate',
'DocNumber',
'Line.SalesItemLineDetail.ItemRef.value',
'Line.Description',
'Line.SalesItemLineDetail.Qty',
'Line.SalesItemLineDetail.UnitPrice',
'Line.Amount',
'PrivateNote',
],
'bill': [
'TxnDate',
'VendorRef.value',
'DueDate',
'DocNumber',
'Line.AccountBasedExpenseLineDetail.AccountRef.value',
'Line.Description',
'Line.Amount',
'PrivateNote',
],
'customer': [
'DisplayName',
'CompanyName',
'GivenName',
'FamilyName',
'PrimaryEmailAddr.Address',
'PrimaryPhone.FreeFormNumber',
'BillAddr.Line1',
'BillAddr.City',
'BillAddr.CountrySubDivisionCode',
'BillAddr.PostalCode',
],
'vendor': [
'DisplayName',
'CompanyName',
'GivenName',
'FamilyName',
'PrimaryEmailAddr.Address',
'PrimaryPhone.FreeFormNumber',
'BillAddr.Line1',
'BillAddr.City',
'BillAddr.CountrySubDivisionCode',
'BillAddr.PostalCode',
'TaxIdentifier',
],
'account': [
'Name',
'AccountType',
'AccountSubType',
'AcctNum',
'Description',
'CurrentBalance',
],
}
return fields.get(data_type, [])
def _save_template(self):
"""Save current mapping as a template."""
if not self.field_mappings:
messagebox.showwarning("No Mapping", "Please map fields first.")
return
# Ask for template name
name = tk.simpledialog.askstring("Save Template", "Enter template name:")
if not name:
return
# Import required classes
from src.config.settings import MappingTemplate, FieldMapping
# Convert field_mappings dict to list of FieldMapping objects
mappings = []
for excel_col, qbo_field in self.field_mappings.items():
if qbo_field:
mappings.append(FieldMapping(
excel_column=excel_col,
qbo_field=qbo_field
))
template = MappingTemplate(
name=name,
data_type=self.data_type_var.get(),
mappings=mappings
)
self.settings.save_template(template)
messagebox.showinfo("Saved", f"Template '{name}' saved successfully.")
logger.info(f"[CREATE] Template saved: {name}")
self._refresh_templates()
def _start_import(self):
"""Start the import process."""
if not self.is_connected:
messagebox.showwarning("Not Connected", "Please connect to QuickBooks first.")
return
if not hasattr(self, 'excel_data') or not self.excel_data:
messagebox.showwarning("No Data", "Please load a file first.")
return
if not self.field_mappings:
messagebox.showwarning("No Mapping", "Please map fields first.")
return
# Confirm import
row_count = len(self.excel_data)
if not messagebox.askyesno("Confirm Import", f"Import {row_count} records to QuickBooks?"):
return
# Disable buttons
self.import_btn.config(state=tk.DISABLED)
self.stop_btn.config(state=tk.NORMAL)
# Clear log
self.log_viewer.clear()
self.log_viewer.log(f"Starting import of {row_count} records...")
# Start import in background thread
self._import_cancelled = False
threading.Thread(target=self._run_import, daemon=True).start()
def _run_import(self):
"""Run the import in a background thread."""
try:
data_type = self.data_type_var.get()
total = len(self.excel_data)
successful = 0
failed = 0
# Cache for lookups to avoid repeated API calls
vendor_cache = {}
customer_cache = {}
account_cache = {}
# Pre-load caches if needed
if self.connection_type == 'qbo' and self.qbo_client:
self.root.after(0, lambda: self.log_viewer.log("Loading reference data...", "INFO"))
try:
if data_type in ['check', 'bill']:
for v in self.qbo_client.get_vendors():
vendor_cache[v.get('DisplayName', '').lower()] = v['Id']
if data_type == 'invoice':
for c in self.qbo_client.get_customers():
customer_cache[c.get('DisplayName', '').lower()] = c['Id']
# Load accounts for check/bill
if data_type in ['check', 'bill']:
for a in self.qbo_client.get_accounts():
account_cache[a.get('Name', '').lower()] = a['Id']
except Exception as e:
self.root.after(0, lambda: self.log_viewer.log(f"Warning: Could not load all reference data: {e}", "WARNING"))
for i, row in enumerate(self.excel_data):
if self._import_cancelled:
self.root.after(0, lambda: self.log_viewer.log("Import cancelled by user.", "WARNING"))
break
try:
# Build QBO data from mapping
raw_data = {}
for excel_col, qbo_field in self.field_mappings.items():
if qbo_field and excel_col in row:
raw_data[qbo_field] = row[excel_col]
# Build proper QBO structure based on data type
qbo_data = self._build_qbo_structure(
data_type, raw_data,
vendor_cache, customer_cache, account_cache
)
# Create record in QuickBooks
if self.connection_type == 'qbo' and self.qbo_client:
if data_type == 'check':
self.qbo_client.create_check(qbo_data)
elif data_type == 'invoice':
self.qbo_client.create_invoice(qbo_data)
elif data_type == 'bill':
self.qbo_client.create_bill(qbo_data)
elif data_type == 'customer':
self.qbo_client.create_customer(qbo_data)
elif data_type == 'vendor':
self.qbo_client.create_vendor(qbo_data)
elif data_type == 'account':
self.qbo_client.create_account(qbo_data)
successful += 1
self.root.after(0, lambda i=i, s=successful: self._update_import_progress(i + 1, total, s))
except Exception as e:
failed += 1
error_msg = str(e)
self.root.after(0, lambda i=i, e=error_msg: self.log_viewer.log(f"Row {i+1} failed: {e}", "ERROR"))
# Complete
self.root.after(0, lambda: self._import_complete(successful, failed, total))
except Exception as e:
self.root.after(0, lambda: self._import_error(str(e)))
def _build_qbo_structure(
self,
data_type: str,
raw_data: Dict[str, Any],
vendor_cache: Dict[str, str],
customer_cache: Dict[str, str],
account_cache: Dict[str, str]
) -> Dict[str, Any]:
"""Build proper QBO API structure from raw mapped data with dot-notation fields."""
def get_value(key: str) -> Any:
"""Get value from raw_data handling dot notation keys."""
# Direct match first
if key in raw_data:
return raw_data[key]
# Try with .value suffix
if key + '.value' in raw_data:
return raw_data[key + '.value']
return None
def format_date(value: Any) -> str:
"""Format a date value to YYYY-MM-DD."""
if value is None:
return None
if hasattr(value, 'strftime'):
return value.strftime('%Y-%m-%d')
# Handle string dates
str_val = str(value)
if 'T' in str_val:
return str_val.split('T')[0]
return str_val[:10]
def resolve_vendor(name: str) -> str:
"""Resolve vendor name to ID."""
if not name:
return None
vendor_id = vendor_cache.get(str(name).strip().lower())
if not vendor_id:
raise ValueError(f"Vendor not found: {name}")
return vendor_id
def resolve_customer(name: str) -> str:
"""Resolve customer name to ID."""
if not name:
return None
customer_id = customer_cache.get(str(name).strip().lower())
if not customer_id:
raise ValueError(f"Customer not found: {name}")
return customer_id
def resolve_account(name: str) -> str:
"""Resolve account name to ID."""
if not name:
return None
account_id = account_cache.get(str(name).strip().lower())
if not account_id:
raise ValueError(f"Account not found: {name}")
return account_id
if data_type == 'check':
# Build check/purchase structure
check = {"PaymentType": "Check"}
# Transaction date
txn_date = get_value('TxnDate')
if txn_date:
check['TxnDate'] = format_date(txn_date)
# Entity reference (vendor) - handle EntityRef.value
vendor_name = get_value('EntityRef')
if vendor_name:
vendor_id = resolve_vendor(vendor_name)
check['EntityRef'] = {'value': vendor_id, 'name': str(vendor_name).strip()}
# Bank account reference - handle AccountRef.value
account_name = get_value('AccountRef')
if account_name:
account_id = resolve_account(account_name)
check['AccountRef'] = {'value': account_id, 'name': str(account_name).strip()}
# Line items
amount = get_value('Line.Amount')
if amount:
line = {
'Amount': float(amount),
'DetailType': 'AccountBasedExpenseLineDetail',
'AccountBasedExpenseLineDetail': {}
}
# Expense account
expense_account = get_value('Line.AccountBasedExpenseLineDetail.AccountRef')
if expense_account:
expense_id = resolve_account(expense_account)
line['AccountBasedExpenseLineDetail']['AccountRef'] = {
'value': expense_id,
'name': str(expense_account).strip()
}
elif check.get('AccountRef'):
# Use bank account as fallback
line['AccountBasedExpenseLineDetail']['AccountRef'] = check['AccountRef']
# Description
description = get_value('Line.Description')
if description:
line['Description'] = str(description)
check['Line'] = [line]
# Doc number
doc_number = get_value('DocNumber')
if doc_number:
check['DocNumber'] = str(doc_number)
# Private note
private_note = get_value('PrivateNote')
if private_note:
check['PrivateNote'] = str(private_note)
return check
elif data_type == 'invoice':
invoice = {}
# Transaction date
txn_date = get_value('TxnDate')
if txn_date:
invoice['TxnDate'] = format_date(txn_date)
# Customer reference
customer_name = get_value('CustomerRef')
if customer_name:
customer_id = resolve_customer(customer_name)
invoice['CustomerRef'] = {'value': customer_id, 'name': str(customer_name).strip()}
# Due date
due_date = get_value('DueDate')
if due_date:
invoice['DueDate'] = format_date(due_date)
# Line items
amount = get_value('Line.Amount')
if amount:
line = {
'Amount': float(amount),
'DetailType': 'SalesItemLineDetail',
'SalesItemLineDetail': {}
}
# Item reference
item_ref = get_value('Line.SalesItemLineDetail.ItemRef')
if item_ref:
line['SalesItemLineDetail']['ItemRef'] = {'value': str(item_ref)}
# Quantity
qty = get_value('Line.SalesItemLineDetail.Qty')
if qty:
line['SalesItemLineDetail']['Qty'] = float(qty)
# Unit price
unit_price = get_value('Line.SalesItemLineDetail.UnitPrice')
if unit_price:
line['SalesItemLineDetail']['UnitPrice'] = float(unit_price)
# Description
description = get_value('Line.Description')
if description:
line['Description'] = str(description)
invoice['Line'] = [line]
# Doc number
doc_number = get_value('DocNumber')
if doc_number:
invoice['DocNumber'] = str(doc_number)
# Private note
private_note = get_value('PrivateNote')
if private_note:
invoice['PrivateNote'] = str(private_note)
return invoice
elif data_type == 'bill':
bill = {}
# Transaction date
txn_date = get_value('TxnDate')
if txn_date:
bill['TxnDate'] = format_date(txn_date)
# Vendor reference
vendor_name = get_value('VendorRef')
if vendor_name:
vendor_id = resolve_vendor(vendor_name)
bill['VendorRef'] = {'value': vendor_id, 'name': str(vendor_name).strip()}
# Due date
due_date = get_value('DueDate')
if due_date:
bill['DueDate'] = format_date(due_date)
# Line items
amount = get_value('Line.Amount')
if amount:
line = {
'Amount': float(amount),
'DetailType': 'AccountBasedExpenseLineDetail',
'AccountBasedExpenseLineDetail': {}
}
# Account reference
account_name = get_value('Line.AccountBasedExpenseLineDetail.AccountRef')
if account_name:
account_id = resolve_account(account_name)
line['AccountBasedExpenseLineDetail']['AccountRef'] = {
'value': account_id,
'name': str(account_name).strip()
}
# Description
description = get_value('Line.Description')
if description:
line['Description'] = str(description)
bill['Line'] = [line]
# Doc number
doc_number = get_value('DocNumber')
if doc_number:
bill['DocNumber'] = str(doc_number)
# Private note
private_note = get_value('PrivateNote')
if private_note:
bill['PrivateNote'] = str(private_note)
return bill
elif data_type == 'customer':
customer = {}
display_name = get_value('DisplayName')
if display_name:
customer['DisplayName'] = str(display_name)
company_name = get_value('CompanyName')
if company_name:
customer['CompanyName'] = str(company_name)
given_name = get_value('GivenName')
if given_name:
customer['GivenName'] = str(given_name)
family_name = get_value('FamilyName')
if family_name:
customer['FamilyName'] = str(family_name)
email = get_value('PrimaryEmailAddr') or get_value('PrimaryEmailAddr.Address')
if email:
customer['PrimaryEmailAddr'] = {'Address': str(email)}
phone = get_value('PrimaryPhone') or get_value('PrimaryPhone.FreeFormNumber')
if phone:
customer['PrimaryPhone'] = {'FreeFormNumber': str(phone)}
# Address
addr_line1 = get_value('BillAddr.Line1')
if addr_line1:
customer['BillAddr'] = customer.get('BillAddr', {})
customer['BillAddr']['Line1'] = str(addr_line1)
addr_city = get_value('BillAddr.City')
if addr_city:
customer['BillAddr'] = customer.get('BillAddr', {})
customer['BillAddr']['City'] = str(addr_city)
addr_state = get_value('BillAddr.CountrySubDivisionCode')
if addr_state:
customer['BillAddr'] = customer.get('BillAddr', {})
customer['BillAddr']['CountrySubDivisionCode'] = str(addr_state)
addr_postal = get_value('BillAddr.PostalCode')
if addr_postal:
customer['BillAddr'] = customer.get('BillAddr', {})
customer['BillAddr']['PostalCode'] = str(addr_postal)
return customer
elif data_type == 'vendor':
vendor = {}
display_name = get_value('DisplayName')
if display_name:
vendor['DisplayName'] = str(display_name)
company_name = get_value('CompanyName')
if company_name:
vendor['CompanyName'] = str(company_name)
given_name = get_value('GivenName')
if given_name:
vendor['GivenName'] = str(given_name)
family_name = get_value('FamilyName')
if family_name:
vendor['FamilyName'] = str(family_name)
email = get_value('PrimaryEmailAddr') or get_value('PrimaryEmailAddr.Address')
if email:
vendor['PrimaryEmailAddr'] = {'Address': str(email)}
phone = get_value('PrimaryPhone') or get_value('PrimaryPhone.FreeFormNumber')
if phone:
vendor['PrimaryPhone'] = {'FreeFormNumber': str(phone)}
tax_id = get_value('TaxIdentifier')
if tax_id:
vendor['TaxIdentifier'] = str(tax_id)
# Address
addr_line1 = get_value('BillAddr.Line1')
if addr_line1:
vendor['BillAddr'] = vendor.get('BillAddr', {})
vendor['BillAddr']['Line1'] = str(addr_line1)
addr_city = get_value('BillAddr.City')
if addr_city:
vendor['BillAddr'] = vendor.get('BillAddr', {})
vendor['BillAddr']['City'] = str(addr_city)
addr_state = get_value('BillAddr.CountrySubDivisionCode')
if addr_state:
vendor['BillAddr'] = vendor.get('BillAddr', {})
vendor['BillAddr']['CountrySubDivisionCode'] = str(addr_state)
addr_postal = get_value('BillAddr.PostalCode')
if addr_postal:
vendor['BillAddr'] = vendor.get('BillAddr', {})
vendor['BillAddr']['PostalCode'] = str(addr_postal)
return vendor
elif data_type == 'account':
account = {}
name = get_value('Name')
if name:
account['Name'] = str(name)
account_type = get_value('AccountType')
if account_type:
account['AccountType'] = str(account_type)
account_sub_type = get_value('AccountSubType')
if account_sub_type:
account['AccountSubType'] = str(account_sub_type)
acct_num = get_value('AcctNum')
if acct_num:
account['AcctNum'] = str(acct_num)
description = get_value('Description')
if description:
account['Description'] = str(description)
return account
else:
# Return raw data as-is for unknown types
return raw_data
def _update_import_progress(self, current: int, total: int, successful: int):
"""Update import progress."""
self.progress_frame.set_progress(
current, total,
f"Importing... ({current}/{total})",
f"{successful} successful"
)
def _import_complete(self, successful: int, failed: int, total: int):
"""Handle import completion."""
self.import_btn.config(state=tk.NORMAL)
self.stop_btn.config(state=tk.DISABLED)
self.progress_frame.set_progress(
total, total,
"Import Complete",
f"{successful} successful, {failed} failed"
)
self.log_viewer.log(f"Import complete: {successful} successful, {failed} failed", "SUCCESS" if failed == 0 else "WARNING")
messagebox.showinfo("Import Complete", f"Imported {successful} of {total} records.\n{failed} failed.")
def _import_error(self, error: str):
"""Handle import error."""
self.import_btn.config(state=tk.NORMAL)
self.stop_btn.config(state=tk.DISABLED)
self.log_viewer.log(f"Import error: {error}", "ERROR")
messagebox.showerror("Import Error", f"Import failed: {error}")
def _stop_import(self):
"""Stop the current import."""
self._import_cancelled = True
self.stop_btn.config(state=tk.DISABLED)
# =========================================================================
# Template Methods
# =========================================================================
def _refresh_templates(self):
"""Refresh the templates list."""
# Clear existing
for item in self.templates_tree.get_children():
self.templates_tree.delete(item)
# Load templates
templates = self.settings.get_templates()
for template in templates:
# Handle both MappingTemplate objects and dict format
if hasattr(template, 'name'):
# MappingTemplate object
name = template.name
data_type = template.data_type
created = template.created_at[:10] if template.created_at else ''
mapping_count = len(template.mappings)
else:
# Dict format (from save_template with dict)
name = template.get('name', '')
data_type = template.get('data_type', '')
created = template.get('created', template.get('created_at', ''))[:10]
mapping_count = len(template.get('mappings', {}))
self.templates_tree.insert('', tk.END, values=(
name,
data_type,
created,
f"{mapping_count} fields"
))
def _create_template(self):
"""Create a new template."""
messagebox.showinfo("Create Template", "Load a file and map fields first, then use 'Save as Template'.")
self.notebook.select(1) # Switch to Import tab
def _edit_template(self):
"""Edit selected template."""
selection = self.templates_tree.selection()
if not selection:
messagebox.showwarning("No Selection", "Please select a template to edit.")
return
# Get template name
item = self.templates_tree.item(selection[0])
template_name = item['values'][0]
# Find template in list
templates = self.settings.get_templates()
template = None
for t in templates:
if hasattr(t, 'name') and t.name == template_name:
template = t
break
if template:
# Convert MappingTemplate.mappings to dict format for field_mappings
self.field_mappings = {}
for mapping in template.mappings:
if hasattr(mapping, 'excel_column'):
self.field_mappings[mapping.excel_column] = mapping.qbo_field
else:
# Dict format fallback
self.field_mappings[mapping.get('excel_column', '')] = mapping.get('qbo_field', '')
self.data_type_var.set(template.data_type if hasattr(template, 'data_type') else 'check')
self._update_mapping()
self.notebook.select(1) # Switch to Import tab
logger.info(f"[EDIT] Loaded template: {template_name}")
def _duplicate_template(self):
"""Duplicate selected template."""
selection = self.templates_tree.selection()
if not selection:
messagebox.showwarning("No Selection", "Please select a template to duplicate.")
return
item = self.templates_tree.item(selection[0])
template_name = item['values'][0]
new_name = tk.simpledialog.askstring("Duplicate Template", "Enter new template name:")
if not new_name:
return
# Find template in list
templates = self.settings.get_templates()
template = None
for t in templates:
if hasattr(t, 'name') and t.name == template_name:
template = t
break
if template:
from src.config.settings import MappingTemplate
# Create new template with copied mappings
new_template = MappingTemplate(
name=new_name,
data_type=template.data_type,
mappings=template.mappings.copy() if hasattr(template.mappings, 'copy') else list(template.mappings)
)
self.settings.save_template(new_template)
logger.info(f"[CREATE] Template duplicated: {template_name} -> {new_name}")
self._refresh_templates()
def _delete_template(self):
"""Delete selected template."""
selection = self.templates_tree.selection()
if not selection:
messagebox.showwarning("No Selection", "Please select a template to delete.")
return
item = self.templates_tree.item(selection[0])
template_name = item['values'][0]
if messagebox.askyesno("Confirm Delete", f"Delete template '{template_name}'?"):
self.settings.delete_template(template_name)
self._refresh_templates()
# =========================================================================
# Other UI Methods
# =========================================================================
def _show_templates(self):
"""Show templates tab."""
self.notebook.select(3)
def _show_qb_lookup(self, entity_type: str):
"""Show QuickBooks entity lookup dialog."""
if not self.is_connected:
messagebox.showwarning("Not Connected", "Please connect to QuickBooks first.")
return
# Entity configurations
entity_config = {
'account': {
'query_name': 'Account',
'display_name': 'Accounts',
'columns': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('AccountType', 'Type', 120),
('AccountSubType', 'SubType', 120),
('AcctNum', 'Number', 80),
('CurrentBalance', 'Balance', 100),
('Active', 'Active', 60),
],
'search_field': 'Name',
'type_filter': True,
'type_options': ['Bank', 'Accounts Receivable', 'Other Current Asset', 'Fixed Asset',
'Accounts Payable', 'Credit Card', 'Other Current Liability',
'Long Term Liability', 'Equity', 'Income', 'Cost of Goods Sold',
'Expense', 'Other Income', 'Other Expense'],
},
'customer': {
'query_name': 'Customer',
'display_name': 'Customers',
'columns': [
('Id', 'ID', 60),
('DisplayName', 'Name', 200),
('CompanyName', 'Company', 150),
('PrimaryEmailAddr.Address', 'Email', 180),
('PrimaryPhone.FreeFormNumber', 'Phone', 120),
('Balance', 'Balance', 100),
('Active', 'Active', 60),
],
'search_field': 'DisplayName',
},
'vendor': {
'query_name': 'Vendor',
'display_name': 'Vendors',
'columns': [
('Id', 'ID', 60),
('DisplayName', 'Name', 200),
('CompanyName', 'Company', 150),
('PrimaryEmailAddr.Address', 'Email', 180),
('PrimaryPhone.FreeFormNumber', 'Phone', 120),
('Balance', 'Balance', 100),
('Active', 'Active', 60),
],
'search_field': 'DisplayName',
},
'item': {
'query_name': 'Item',
'display_name': 'Items/Products',
'columns': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('Type', 'Type', 100),
('Description', 'Description', 250),
('UnitPrice', 'Price', 80),
('Active', 'Active', 60),
],
'search_field': 'Name',
},
'class': {
'query_name': 'Class',
'display_name': 'Classes',
'columns': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('FullyQualifiedName', 'Full Name', 250),
('Active', 'Active', 60),
],
'search_field': 'Name',
},
'department': {
'query_name': 'Department',
'display_name': 'Departments',
'columns': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('FullyQualifiedName', 'Full Name', 250),
('Active', 'Active', 60),
],
'search_field': 'Name',
},
'term': {
'query_name': 'Term',
'display_name': 'Payment Terms',
'columns': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('DueDays', 'Due Days', 80),
('Active', 'Active', 60),
],
'search_field': 'Name',
},
'paymentmethod': {
'query_name': 'PaymentMethod',
'display_name': 'Payment Methods',
'columns': [
('Id', 'ID', 60),
('Name', 'Name', 200),
('Type', 'Type', 100),
('Active', 'Active', 60),
],
'search_field': 'Name',
},
'employee': {
'query_name': 'Employee',
'display_name': 'Employees',
'columns': [
('Id', 'ID', 60),
('DisplayName', 'Name', 200),
('GivenName', 'First Name', 100),
('FamilyName', 'Last Name', 100),
('PrimaryEmailAddr.Address', 'Email', 180),
('Active', 'Active', 60),
],
'search_field': 'DisplayName',
},
}
config = entity_config.get(entity_type)
if not config:
messagebox.showerror("Error", f"Unknown entity type: {entity_type}")
return
# Create dialog
dialog = tk.Toplevel(self.root)
dialog.title(f"QuickBooks {config['display_name']} Lookup")
dialog.geometry("1000x600")
dialog.transient(self.root)
# Connection info
conn_type = "Online" if self.connection_type == 'qbo' else "Desktop"
info_frame = ttk.Frame(dialog)
info_frame.pack(fill=tk.X, padx=10, pady=5)
ttk.Label(
info_frame,
text=f"Connected to: {self.company_name} ({conn_type})",
font=('Segoe UI', 9, 'italic'),
foreground='gray'
).pack(side=tk.LEFT)
# Search frame
search_frame = ttk.Frame(dialog)
search_frame.pack(fill=tk.X, padx=10, pady=10)
ttk.Label(search_frame, text="Search:").pack(side=tk.LEFT)
search_var = tk.StringVar()
search_entry = ttk.Entry(search_frame, textvariable=search_var, width=30)
search_entry.pack(side=tk.LEFT, padx=(5, 10))
# Type filter for accounts
type_var = tk.StringVar(value="")
if config.get('type_filter'):
ttk.Label(search_frame, text="Type:").pack(side=tk.LEFT, padx=(10, 5))
type_combo = ttk.Combobox(
search_frame,
textvariable=type_var,
values=["All Types"] + config.get('type_options', []),
width=20,
state='readonly'
)
type_combo.pack(side=tk.LEFT)
type_combo.current(0)
# Results treeview
tree_frame = ttk.Frame(dialog)
tree_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
columns = [col[0] for col in config['columns']]
tree = ttk.Treeview(tree_frame, columns=columns, show='headings')
for col_key, col_name, col_width in config['columns']:
tree.heading(col_key, text=col_name)
tree.column(col_key, width=col_width, minwidth=50)
# Scrollbars
vsb = ttk.Scrollbar(tree_frame, orient=tk.VERTICAL, command=tree.yview)
hsb = ttk.Scrollbar(tree_frame, orient=tk.HORIZONTAL, command=tree.xview)
tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
tree.grid(row=0, column=0, sticky='nsew')
vsb.grid(row=0, column=1, sticky='ns')
hsb.grid(row=1, column=0, sticky='ew')
tree_frame.grid_rowconfigure(0, weight=1)
tree_frame.grid_columnconfigure(0, weight=1)
# Status bar
status_var = tk.StringVar(value="Click 'Search' to load data")
status_label = ttk.Label(dialog, textvariable=status_var, foreground='gray')
status_label.pack(fill=tk.X, padx=10, pady=5)
# Button frame
btn_frame = ttk.Frame(dialog)
btn_frame.pack(fill=tk.X, padx=10, pady=10)
def get_nested_value(obj, key_path):
"""Get nested value from dictionary using dot notation."""
keys = key_path.split('.')
value = obj
for key in keys:
if isinstance(value, dict):
value = value.get(key)
else:
return None
return value
def format_value(value, col_key):
"""Format value for display."""
if value is None:
return '-'
elif isinstance(value, bool):
return 'Yes' if value else 'No'
elif 'Balance' in col_key or 'Price' in col_key:
try:
return f"${float(value):,.2f}"
except:
return str(value)
else:
return str(value)
def search_entities():
"""Search for entities."""
# Clear existing
for item in tree.get_children():
tree.delete(item)
search_term = search_var.get().strip()
type_filter = type_var.get() if type_var.get() != "All Types" else ""
status_var.set("Searching...")
dialog.update()
try:
entities = []
if self.connection_type == 'qbo' and self.qbo_client:
# QuickBooks Online query
entities = self._fetch_qbo_entities(
config['query_name'],
config['search_field'],
search_term,
type_filter if entity_type == 'account' else None
)
elif self.connection_type == 'qbd' and self.is_connected:
# QuickBooks Desktop query
entities = self._fetch_qbd_entities(
entity_type,
search_term,
type_filter if entity_type == 'account' else None
)
# Populate tree
for entity in entities:
values = []
for col_key, _, _ in config['columns']:
value = get_nested_value(entity, col_key)
values.append(format_value(value, col_key))
tree.insert('', tk.END, values=values)
status_var.set(f"Found {len(entities)} record(s)")
logger.info(f"[LOOKUP] {config['display_name']}: Found {len(entities)} records")
except Exception as e:
status_var.set(f"Error: {str(e)}")
logger.error(f"Lookup error: {e}")
messagebox.showerror("Error", f"Failed to fetch data: {str(e)}")
def export_csv():
"""Export results to CSV."""
items = tree.get_children()
if not items:
messagebox.showwarning("No Data", "No data to export.")
return
from tkinter import filedialog
filepath = filedialog.asksaveasfilename(
defaultextension=".csv",
filetypes=[("CSV files", "*.csv"), ("All files", "*.*")],
initialfile=f"{config['query_name'].lower()}_export.csv"
)
if not filepath:
return
import csv
with open(filepath, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
# Header
writer.writerow([col[1] for col in config['columns']])
# Data
for item in items:
writer.writerow(tree.item(item)['values'])
messagebox.showinfo("Export Complete", f"Exported {len(items)} records to:\n{filepath}")
logger.info(f"[EXPORT] Exported {len(items)} {config['display_name']} to {filepath}")
def copy_selected():
"""Copy selected row to clipboard."""
selection = tree.selection()
if not selection:
messagebox.showinfo("No Selection", "Please select a row to copy.")
return
values = tree.item(selection[0])['values']
# Format as ID: Name
text = f"{values[0]}: {values[1]}"
dialog.clipboard_clear()
dialog.clipboard_append(text)
status_var.set(f"Copied to clipboard: {text}")
# Buttons
ttk.Button(btn_frame, text="Search", command=search_entities, width=12).pack(side=tk.LEFT)
ttk.Button(btn_frame, text="Export CSV", command=export_csv, width=12).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Copy Selected", command=copy_selected, width=14).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Close", command=dialog.destroy, width=12).pack(side=tk.RIGHT)
# Bind Enter key to search
search_entry.bind('<Return>', lambda e: search_entities())
# Double-click to copy
tree.bind('<Double-1>', lambda e: copy_selected())
# Load data immediately
dialog.after(100, search_entities)
def _fetch_qbo_entities(self, query_name: str, search_field: str, search_term: str = None, type_filter: str = None) -> List[Dict]:
"""Fetch entities from QuickBooks Online."""
query = f"SELECT * FROM {query_name}"
conditions = []
if search_term:
conditions.append(f"{search_field} LIKE '%{search_term}%'")
if type_filter:
conditions.append(f"AccountType = '{type_filter}'")
if conditions:
query += " WHERE " + " AND ".join(conditions)
query += f" ORDERBY {search_field} MAXRESULTS 1000"
response = self.qbo_client._make_request('GET', 'query', params={'query': query})
if response and 'QueryResponse' in response:
return response['QueryResponse'].get(query_name, [])
return []
def _get_python_executable(self) -> str:
"""Get the path to a Python executable for subprocess calls.
When running as a PyInstaller bundle, sys.executable points to the bundle,
so we need to find the system Python installation."""
import shutil
# Check if we're running as a PyInstaller bundle
if getattr(sys, 'frozen', False):
# Try to find system Python
# Check common Python installation paths on Windows
python_paths = [
shutil.which('python'),
shutil.which('python3'),
shutil.which('py'),
r'C:\Python312\python.exe',
r'C:\Python311\python.exe',
r'C:\Python310\python.exe',
r'C:\Python39\python.exe',
r'C:\Python38\python.exe',
os.path.expandvars(r'%LOCALAPPDATA%\Programs\Python\Python312\python.exe'),
os.path.expandvars(r'%LOCALAPPDATA%\Programs\Python\Python311\python.exe'),
os.path.expandvars(r'%LOCALAPPDATA%\Programs\Python\Python310\python.exe'),
os.path.expandvars(r'%LOCALAPPDATA%\Programs\Python\Python39\python.exe'),
]
for python_path in python_paths:
if python_path and os.path.isfile(python_path):
logger.info(f"Found system Python at: {python_path}")
return python_path
# If no system Python found, raise an error
raise Exception(
"QuickBooks Desktop requires Python to be installed on your system.\n"
"Please install Python from https://www.python.org/downloads/\n"
"and make sure to check 'Add Python to PATH' during installation."
)
else:
# Running as script, use current Python
return sys.executable
def _fetch_qbd_entities(self, entity_type: str, search_term: str = None, type_filter: str = None) -> List[Dict]:
"""Fetch entities from QuickBooks Desktop using subprocess for COM compatibility."""
if not HAS_QBD:
raise Exception("QuickBooks Desktop integration not available")
if self.connection_type != 'qbd' or not self.is_connected:
raise Exception("Not connected to QuickBooks Desktop")
# Map entity types to QBD query request names
query_map = {
'account': 'AccountQueryRq',
'customer': 'CustomerQueryRq',
'vendor': 'VendorQueryRq',
'item': 'ItemQueryRq',
'employee': 'EmployeeQueryRq',
'class': 'ClassQueryRq',
'paymentmethod': 'PaymentMethodQueryRq',
'term': 'TermsQueryRq',
}
if entity_type not in query_map:
raise Exception(f"Unsupported entity type: {entity_type}")
query_rq = query_map[entity_type]
# Build the subprocess script - use regular string to avoid f-string issues
query_script = '''
import sys
import json
import xml.etree.ElementTree as ET
QUERY_RQ = "''' + query_rq + '''"
try:
import pythoncom
from win32com.client import Dispatch
pythoncom.CoInitialize()
rp = Dispatch("QBXMLRP2.RequestProcessor")
rp.OpenConnection2("", "QBO Excel Sync Lookup", 1)
ticket = rp.BeginSession("", 0)
# Build query XML
xml_request = """<?xml version="1.0" encoding="utf-8"?>
<?qbxml version="13.0"?>
<QBXML>
<QBXMLMsgsRq onError="stopOnError">
<""" + QUERY_RQ + """>
<MaxReturned>1000</MaxReturned>
<ActiveStatus>All</ActiveStatus>
</""" + QUERY_RQ + """>
</QBXMLMsgsRq>
</QBXML>"""
response = rp.ProcessRequest(ticket, xml_request)
rp.EndSession(ticket)
rp.CloseConnection()
pythoncom.CoUninitialize()
# Parse XML response
root = ET.fromstring(response)
results = []
# Find all return elements
for elem in root.iter():
# Check if this is a return element with ListID
if "Ret" in elem.tag:
list_id = elem.find("ListID")
if list_id is not None:
entity = {"Id": list_id.text}
# Extract common fields
for field in ["Name", "FullName", "IsActive", "AccountType", "AccountSubType",
"CompanyName", "FirstName", "LastName", "Balance", "TotalBalance",
"Description", "SalesDesc", "PaymentMethodType", "DueDays", "StdDueDays"]:
el = elem.find(field)
if el is not None and el.text:
entity[field] = el.text
# Handle nested phone
for phone_path in [".//Phone", ".//PrimaryPhone", ".//MainPhone"]:
phone_elem = elem.find(phone_path)
if phone_elem is not None:
phone_text = phone_elem.text or phone_elem.find("FreeFormNumber")
if phone_text is not None:
if hasattr(phone_text, "text"):
entity["Phone"] = phone_text.text
else:
entity["Phone"] = str(phone_text)
break
# Handle email
email = elem.find(".//Email")
if email is not None and email.text:
entity["Email"] = email.text
# Price fields for items
for price_path in [".//SalesOrPurchase/Price", ".//SalesAndPurchase/SalesPrice",
".//SalesPrice", ".//Price"]:
price = elem.find(price_path)
if price is not None and price.text:
entity["UnitPrice"] = price.text
break
results.append(entity)
print(json.dumps({"success": True, "data": results}))
except Exception as e:
import traceback
print(json.dumps({"success": False, "error": str(e), "traceback": traceback.format_exc()}))
sys.exit(1)
'''
try:
# Get Python executable (handles PyInstaller bundled case)
python_exe = self._get_python_executable()
result = subprocess.run(
[python_exe, '-c', query_script],
capture_output=True,
text=True,
timeout=120
)
# Check for output
stdout = result.stdout.strip() if result.stdout else ''
stderr = result.stderr.strip() if result.stderr else ''
if not stdout:
error_detail = stderr if stderr else 'No output from QuickBooks query'
raise Exception(f"Query failed: {error_detail}")
try:
output = json.loads(stdout)
except json.JSONDecodeError:
raise Exception(f"Invalid response from QuickBooks: {stdout[:500]}")
if output.get('success'):
raw_entities = output.get('data', [])
else:
error_msg = output.get('error', 'Unknown error')
tb = output.get('traceback', '')
logger.error(f"QBD query error: {error_msg}\n{tb}")
raise Exception(error_msg)
def clean_qbd_id(list_id: str) -> str:
"""Clean up QBD ListID for display.
QBD ListIDs are like '80000026-1335494949'.
Extract just the hex part and convert to a cleaner number."""
if not list_id:
return ''
# If it contains a dash, take just the first part (hex ID)
if '-' in list_id:
hex_part = list_id.split('-')[0]
# Convert hex to decimal for cleaner display
try:
# Remove leading '8' which is a prefix, then convert
if hex_part.startswith('8'):
return str(int(hex_part[1:], 16))
return str(int(hex_part, 16))
except:
return hex_part
return list_id
# Transform to standard format
entities = []
for e in raw_entities:
try:
raw_id = e.get('Id', '')
clean_id = clean_qbd_id(raw_id)
if entity_type == 'account':
balance = e.get('Balance') or e.get('TotalBalance') or '0'
entity = {
'RawId': raw_id,
'Id': clean_id,
'Name': e.get('Name') or e.get('FullName', ''),
'AccountType': e.get('AccountType', ''),
'AccountSubType': e.get('AccountSubType', ''),
'CurrentBalance': float(balance) if balance else 0,
'Active': str(e.get('IsActive', 'true')).lower() == 'true',
}
elif entity_type in ['customer', 'vendor']:
balance = e.get('Balance') or e.get('TotalBalance') or '0'
entity = {
'RawId': raw_id,
'Id': clean_id,
'DisplayName': e.get('Name') or e.get('FullName', ''),
'CompanyName': e.get('CompanyName', ''),
'PrimaryEmailAddr': {'Address': e.get('Email', '')},
'PrimaryPhone': {'FreeFormNumber': e.get('Phone', '')},
'Balance': float(balance) if balance else 0,
'Active': str(e.get('IsActive', 'true')).lower() == 'true',
}
elif entity_type == 'item':
price = e.get('UnitPrice') or e.get('SalesPrice') or '0'
entity = {
'RawId': raw_id,
'Id': clean_id,
'Name': e.get('Name') or e.get('FullName', ''),
'Type': 'Service',
'Description': e.get('Description') or e.get('SalesDesc', ''),
'UnitPrice': float(price) if price else 0,
'Active': str(e.get('IsActive', 'true')).lower() == 'true',
}
elif entity_type == 'employee':
entity = {
'RawId': raw_id,
'Id': clean_id,
'DisplayName': e.get('Name') or e.get('FullName', ''),
'GivenName': e.get('FirstName', ''),
'FamilyName': e.get('LastName', ''),
'Active': str(e.get('IsActive', 'true')).lower() == 'true',
}
elif entity_type == 'class':
entity = {
'RawId': raw_id,
'Id': clean_id,
'Name': e.get('Name', ''),
'FullyQualifiedName': e.get('FullName', ''),
'Active': str(e.get('IsActive', 'true')).lower() == 'true',
}
elif entity_type == 'paymentmethod':
entity = {
'RawId': raw_id,
'Id': clean_id,
'Name': e.get('Name', ''),
'Type': e.get('PaymentMethodType', ''),
'Active': str(e.get('IsActive', 'true')).lower() == 'true',
}
elif entity_type == 'term':
entity = {
'RawId': raw_id,
'Id': clean_id,
'Name': e.get('Name', ''),
'DueDays': e.get('DueDays') or e.get('StdDueDays', ''),
'Active': str(e.get('IsActive', 'true')).lower() == 'true',
}
else:
entity = {
'RawId': raw_id,
'Id': clean_id,
'Name': e.get('Name') or e.get('FullName', ''),
'Active': str(e.get('IsActive', 'true')).lower() == 'true',
}
entities.append(entity)
except Exception as transform_err:
logger.warning(f"Error transforming entity: {transform_err}")
continue
# Apply search filter
if search_term:
search_lower = search_term.lower()
entities = [ent for ent in entities
if search_lower in str(ent.get('Name', '')).lower()
or search_lower in str(ent.get('DisplayName', '')).lower()
or search_lower in str(ent.get('CompanyName', '')).lower()]
# Apply type filter for accounts
if type_filter and entity_type == 'account':
entities = [ent for ent in entities if ent.get('AccountType') == type_filter]
logger.info(f"[QBD LOOKUP] Fetched {len(entities)} {entity_type}(s)")
return entities
except subprocess.TimeoutExpired:
raise Exception("Query timed out. QuickBooks Desktop may not be responding.")
except Exception as e:
logger.error(f"QBD lookup error: {e}")
raise
def _show_settings(self):
"""Show settings dialog."""
dialog = tk.Toplevel(self.root)
dialog.title("Settings")
dialog.geometry("500x400")
dialog.transient(self.root)
dialog.grab_set()
notebook = ttk.Notebook(dialog)
notebook.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
# Import settings tab
import_tab = ttk.Frame(notebook, padding=15)
notebook.add(import_tab, text="Import")
import_settings = self.settings.get_import_settings()
ttk.Label(import_tab, text="Date Format:").grid(row=0, column=0, sticky=tk.W, pady=5)
date_format_var = tk.StringVar(value=import_settings.date_format)
ttk.Entry(import_tab, textvariable=date_format_var, width=30).grid(row=0, column=1, pady=5)
skip_empty_var = tk.BooleanVar(value=import_settings.skip_empty_rows)
ttk.Checkbutton(import_tab, text="Skip empty rows", variable=skip_empty_var).grid(row=1, column=0, columnspan=2, sticky=tk.W, pady=5)
validate_var = tk.BooleanVar(value=import_settings.validate_before_import)
ttk.Checkbutton(import_tab, text="Validate before import", variable=validate_var).grid(row=2, column=0, columnspan=2, sticky=tk.W, pady=5)
stop_on_error_var = tk.BooleanVar(value=import_settings.stop_on_error)
ttk.Checkbutton(import_tab, text="Stop on first error", variable=stop_on_error_var).grid(row=3, column=0, columnspan=2, sticky=tk.W, pady=5)
def save_settings():
import_settings.date_format = date_format_var.get()
import_settings.skip_empty_rows = skip_empty_var.get()
import_settings.validate_before_import = validate_var.get()
import_settings.stop_on_error = stop_on_error_var.get()
self.settings.save_import_settings(import_settings)
dialog.destroy()
messagebox.showinfo("Saved", "Settings saved successfully.")
ttk.Button(dialog, text="Save", command=save_settings).pack(pady=10)
def _show_logs(self):
"""Show log viewer dialog."""
dialog = tk.Toplevel(self.root)
dialog.title("Application Logs")
dialog.geometry("800x500")
# Read log file - check multiple possible locations
possible_log_paths = [
Path('desktop_app.log'), # Current working directory
PROJECT_ROOT / 'desktop_app.log', # Script directory
Path(sys.executable).parent / 'desktop_app.log', # Executable directory
Path.home() / 'desktop_app.log', # Home directory
]
log_file = None
for path in possible_log_paths:
if path.exists():
log_file = path
break
text = scrolledtext.ScrolledText(dialog, wrap=tk.WORD, font=('Consolas', 9))
text.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
if log_file and log_file.exists():
try:
with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
text.insert(tk.END, f.read())
text.insert(tk.END, f"\n\n--- Log file: {log_file} ---")
except Exception as e:
text.insert(tk.END, f"Error reading log file: {e}")
else:
text.insert(tk.END, f"No log file found.\n\nSearched locations:\n")
for path in possible_log_paths:
text.insert(tk.END, f" - {path}\n")
text.config(state=tk.DISABLED)
# Scroll to end
text.see(tk.END)
btn_frame = ttk.Frame(dialog)
btn_frame.pack(fill=tk.X, padx=10, pady=5)
def clear_logs():
if messagebox.askyesno("Confirm", "Clear all logs?"):
if log_file and log_file.exists():
log_file.unlink()
text.config(state=tk.NORMAL)
text.delete(1.0, tk.END)
text.insert(tk.END, "Logs cleared.")
text.config(state=tk.DISABLED)
def open_log_folder():
if log_file:
folder = log_file.parent
else:
folder = Path.cwd()
if sys.platform == 'win32':
os.startfile(folder)
elif sys.platform == 'darwin':
subprocess.run(['open', folder])
else:
subprocess.run(['xdg-open', folder])
ttk.Button(btn_frame, text="Clear Logs", command=clear_logs).pack(side=tk.LEFT)
ttk.Button(btn_frame, text="Open Folder", command=open_log_folder).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Refresh", command=lambda: (dialog.destroy(), self._show_logs())).pack(side=tk.LEFT, padx=5)
ttk.Button(btn_frame, text="Close", command=dialog.destroy).pack(side=tk.RIGHT)
def _show_help(self):
"""Show help documentation."""
help_text = """
╔═══════════════════════════════════════════════════════════════════════════╗
║ QBO EXCEL SYNC - USER GUIDE ║
╚═══════════════════════════════════════════════════════════════════════════╝
TABLE OF CONTENTS
─────────────────
1. Installation & Setup
2. Connecting to QuickBooks
3. Connection Profiles
4. Importing Data
5. Field Mapping & Templates
6. QuickBooks Lookup
7. Troubleshooting
═══════════════════════════════════════════════════════════════════════════
1. INSTALLATION & SETUP
═══════════════════════════════════════════════════════════════════════════
FOR EXECUTABLE VERSION (.exe):
• Simply double-click QBO_Excel_Sync.exe to run
• No installation required
• For QuickBooks Desktop: Python must be installed on your system
FOR SOURCE VERSION:
1. Install Python 3.8 or higher from https://www.python.org
2. Install dependencies: pip install -r requirements.txt
3. Run: python desktop_app.py
BUILDING THE EXECUTABLE:
1. Run build_windows.bat (Windows)
2. Find executable in dist/QBO_Excel_Sync.exe
═══════════════════════════════════════════════════════════════════════════
2. CONNECTING TO QUICKBOOKS
═══════════════════════════════════════════════════════════════════════════
QUICKBOOKS ONLINE:
1. Go to the Connection tab
2. Select "QuickBooks Online"
3. Enter your API credentials:
• Client ID: From Intuit Developer Portal
• Client Secret: From Intuit Developer Portal
• Redirect URI: Your callback URL
• Environment: sandbox (testing) or production (live)
4. Click "Connect to QuickBooks"
5. Authorize in the browser
6. For remote callback servers: Copy the code back to the app
QUICKBOOKS DESKTOP:
1. Open QuickBooks Desktop with your company file
2. Go to the Connection tab
3. Select "QuickBooks Desktop"
4. Click "Connect to QuickBooks Desktop"
5. Authorize the connection in QuickBooks Desktop
═══════════════════════════════════════════════════════════════════════════
3. CONNECTION PROFILES
═══════════════════════════════════════════════════════════════════════════
Profiles save your connection settings for easy switching between companies.
MANAGING PROFILES:
• Load: Load the selected profile into the form
• Save: Save current settings to the selected profile
• Save As: Create a new profile with current settings
• Delete: Remove the selected profile
IMPORT/EXPORT PROFILES:
• Import Profile: Load profile from a JSON file
• Export Profile: Save selected profile to a JSON file
• Export All: Backup all profiles to a single JSON file
When exporting, you can choose to include or exclude access tokens:
• Include tokens: Can connect immediately after import
• Exclude tokens: More secure, requires re-authentication
═══════════════════════════════════════════════════════════════════════════
4. IMPORTING DATA
═══════════════════════════════════════════════════════════════════════════
SUPPORTED DATA TYPES:
• Checks: Payment checks with payee, bank account, and expense details
• Invoices: Customer invoices with line items
• Bills: Vendor bills with expense accounts
• Customers: Customer records
• Vendors: Vendor records
• Accounts: Chart of accounts entries
IMPORT PROCESS:
1. Click "Browse" to select an Excel file (.xlsx, .xls)
2. Select the data type from the dropdown
3. Map Excel columns to QuickBooks fields (or use Auto-Map)
4. Click "Preview" to verify data
5. Click "Start Import" to begin
IMPORT OPTIONS (Tools → Settings):
• Skip Empty Rows: Ignore rows with no data
• Validate Before Import: Check data before sending
• Stop on Error: Halt import if an error occurs
• Date Format: Expected date format in your Excel file
═══════════════════════════════════════════════════════════════════════════
5. FIELD MAPPING & TEMPLATES
═══════════════════════════════════════════════════════════════════════════
FIELD MAPPING:
• Select an Excel column on the left
• Select a QuickBooks field on the right
• Click "Add Mapping" to create the link
• Use "Auto-Map" to automatically match common field names
TEMPLATES:
• Save frequently used mappings as templates
• Load templates to quickly apply mappings
• Templates are stored locally and persist between sessions
REQUIRED FIELDS (vary by data type):
• Checks: Payee, Bank Account, Date, Amount, Expense Account
• Invoices: Customer, Date, Line Items
• Bills: Vendor, Date, Expense Account, Amount
═══════════════════════════════════════════════════════════════════════════
6. QUICKBOOKS LOOKUP
═══════════════════════════════════════════════════════════════════════════
The Lookup tab lets you search and view QuickBooks data.
SEARCHABLE ENTITIES:
• Accounts (filter by type: Bank, Expense, Income, etc.)
• Customers
• Vendors
• Items
• Employees
• Classes
• Payment Methods
• Terms
FEATURES:
• Search by name
• Filter accounts by type
• Export results to CSV
• Double-click to copy ID and name
• View both Raw ID and Clean ID (for QuickBooks Desktop)
═══════════════════════════════════════════════════════════════════════════
7. TROUBLESHOOTING
═══════════════════════════════════════════════════════════════════════════
CONNECTION ISSUES:
• "Invalid account type" error: Verify the Bank Account is type "Bank"
• OAuth errors: Check Client ID, Secret, and Redirect URI match exactly
• Production sandbox mismatch: Ensure Environment setting is correct
QUICKBOOKS DESKTOP:
• "No module named encodings": Install Python on your system
• Connection timeout: Ensure QuickBooks Desktop is open with company file
• Authorization denied: Authorize the app in QuickBooks Desktop
IMPORT ERRORS:
• "Entity not found": Check that referenced accounts/customers exist
• "Invalid date": Verify date format matches your settings
• Account type errors: Use correct account types (Bank for checks, etc.)
LOG FILES:
• View logs: Tools → View Logs
• Log location: Same directory as the application
GETTING HELP:
• Check the log file for detailed error messages
• Verify your data in the Preview before importing
• Use the Lookup tab to find correct account names and IDs
═══════════════════════════════════════════════════════════════════════════
© 2024-2026 QBO Excel Sync
═══════════════════════════════════════════════════════════════════════════
"""
dialog = tk.Toplevel(self.root)
dialog.title("Help - User Guide")
dialog.geometry("750x600")
text = scrolledtext.ScrolledText(dialog, wrap=tk.WORD, font=('Consolas', 10))
text.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
text.insert(tk.END, help_text)
text.config(state=tk.DISABLED)
ttk.Button(dialog, text="Close", command=dialog.destroy).pack(pady=10)
def _show_about(self):
"""Show about dialog."""
messagebox.showinfo(
"About",
f"{APP_NAME}\nVersion {APP_VERSION}\n\n"
"Import Excel data to QuickBooks Online or Desktop.\n\n"
"© 2024 QBO Excel Sync"
)
def _on_close(self):
"""Handle window close event."""
if messagebox.askyesno("Exit", "Are you sure you want to exit?"):
logger.info("Application closing")
self.root.destroy()
def run(self):
"""Run the application."""
self.root.mainloop()
# =============================================================================
# Entry Point
# =============================================================================
def main():
"""Main entry point."""
# Add simpledialog for askstring
import tkinter.simpledialog
tk.simpledialog = tkinter.simpledialog
app = QBOExcelSyncApp()
app.run()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink