490 lines
20 KiB
Python
490 lines
20 KiB
Python
"""
|
|
Application settings and configuration management.
|
|
Handles OAuth tokens, mapping templates, and user preferences.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Optional, Dict, List, Any
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class QBOCredentials:
|
|
"""QuickBooks Online OAuth credentials."""
|
|
client_id: str = ""
|
|
client_secret: str = ""
|
|
redirect_uri: str = "http://localhost:5000/oauth/callback"
|
|
environment: str = "sandbox" # 'sandbox' or 'production'
|
|
|
|
# Token data (stored securely)
|
|
access_token: Optional[str] = None
|
|
refresh_token: Optional[str] = None
|
|
realm_id: Optional[str] = None
|
|
token_expiry: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class ConnectionProfile:
|
|
"""Connection profile for QuickBooks Online."""
|
|
name: str
|
|
client_id: str = ""
|
|
client_secret: str = ""
|
|
redirect_uri: str = "http://localhost:9000/oauth/callback"
|
|
environment: str = "sandbox" # 'sandbox' or 'production'
|
|
access_token: Optional[str] = None
|
|
refresh_token: Optional[str] = None
|
|
realm_id: Optional[str] = None
|
|
company_name: Optional[str] = None
|
|
token_expiry: Optional[str] = None
|
|
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
|
|
updated_at: str = field(default_factory=lambda: datetime.now().isoformat())
|
|
|
|
|
|
@dataclass
|
|
class FieldMapping:
|
|
"""Single field mapping configuration."""
|
|
excel_column: str
|
|
qbo_field: str
|
|
transform: Optional[str] = None # 'date', 'currency', 'lookup', etc.
|
|
default_value: Optional[str] = None
|
|
required: bool = False
|
|
|
|
|
|
@dataclass
|
|
class MappingTemplate:
|
|
"""Complete mapping template for a data type."""
|
|
name: str
|
|
data_type: str # 'check', 'invoice', 'bill', 'customer', 'vendor', 'account'
|
|
mappings: List[FieldMapping] = field(default_factory=list)
|
|
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
|
|
updated_at: str = field(default_factory=lambda: datetime.now().isoformat())
|
|
|
|
|
|
@dataclass
|
|
class ImportSettings:
|
|
"""Import behavior settings."""
|
|
skip_duplicates: bool = True
|
|
duplicate_check_fields: List[str] = field(default_factory=lambda: ["DocNumber"])
|
|
batch_size: int = 50
|
|
validate_before_import: bool = True
|
|
create_missing_references: bool = False # Auto-create missing customers/vendors
|
|
date_format: str = "%Y-%m-%d"
|
|
decimal_separator: str = "."
|
|
thousand_separator: str = ","
|
|
skip_empty_rows: bool = True
|
|
stop_on_error: bool = False
|
|
|
|
|
|
class Settings:
|
|
"""Application settings manager."""
|
|
|
|
APP_NAME = "QBOExcelSyncWeb"
|
|
|
|
def __init__(self):
|
|
self.config_dir = self._get_config_dir()
|
|
self.config_file = self.config_dir / "config.json"
|
|
self.templates_dir = self.config_dir / "templates"
|
|
self.credentials_file = self.config_dir / ".credentials"
|
|
|
|
# Ensure directories exist
|
|
self.config_dir.mkdir(parents=True, exist_ok=True)
|
|
self.templates_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Load or create default config
|
|
self._config = self._load_config()
|
|
|
|
# Credentials (stored separately for security)
|
|
self._credentials: Optional[QBOCredentials] = None
|
|
|
|
logger.info(f"Settings initialized. Config dir: {self.config_dir}")
|
|
|
|
def _get_config_dir(self) -> Path:
|
|
"""Get platform-specific config directory."""
|
|
import sys
|
|
|
|
# Use user's AppData folder for persistent storage
|
|
# This works correctly for both normal Python and PyInstaller executables
|
|
if sys.platform == 'win32':
|
|
# Windows: Use AppData/Local
|
|
app_data = os.environ.get('LOCALAPPDATA')
|
|
if app_data:
|
|
return Path(app_data) / self.APP_NAME
|
|
# Fallback to APPDATA if LOCALAPPDATA not available
|
|
app_data = os.environ.get('APPDATA')
|
|
if app_data:
|
|
return Path(app_data) / self.APP_NAME
|
|
elif sys.platform == 'darwin':
|
|
# macOS: Use ~/Library/Application Support
|
|
home = Path.home()
|
|
return home / "Library" / "Application Support" / self.APP_NAME
|
|
else:
|
|
# Linux/Unix: Use ~/.config
|
|
home = Path.home()
|
|
return home / ".config" / self.APP_NAME.lower()
|
|
|
|
# Final fallback: Use a 'data' folder relative to executable/script
|
|
if getattr(sys, 'frozen', False):
|
|
# Running as PyInstaller executable
|
|
base = Path(sys.executable).parent
|
|
else:
|
|
# Running as script
|
|
base = Path(__file__).parent.parent.parent
|
|
|
|
return base / "data"
|
|
|
|
def _load_config(self) -> Dict[str, Any]:
|
|
"""Load configuration from file."""
|
|
default_config = {
|
|
"import_settings": asdict(ImportSettings()),
|
|
"recent_files": [],
|
|
"last_template": None,
|
|
"qbo_environment": "sandbox",
|
|
}
|
|
|
|
if self.config_file.exists():
|
|
try:
|
|
with open(self.config_file, 'r') as f:
|
|
loaded = json.load(f)
|
|
# Merge with defaults
|
|
for key, value in default_config.items():
|
|
if key not in loaded:
|
|
loaded[key] = value
|
|
return loaded
|
|
except (json.JSONDecodeError, IOError) as e:
|
|
logger.warning(f"Failed to load config: {e}")
|
|
return default_config
|
|
|
|
return default_config
|
|
|
|
def save(self):
|
|
"""Save configuration to file."""
|
|
try:
|
|
with open(self.config_file, 'w') as f:
|
|
json.dump(self._config, f, indent=2)
|
|
logger.info("Configuration saved successfully")
|
|
except IOError as e:
|
|
logger.error(f"Failed to save config: {e}")
|
|
|
|
@property
|
|
def import_settings(self) -> ImportSettings:
|
|
"""Get import settings."""
|
|
return ImportSettings(**self._config.get("import_settings", {}))
|
|
|
|
@import_settings.setter
|
|
def import_settings(self, settings: ImportSettings):
|
|
"""Set import settings."""
|
|
self._config["import_settings"] = asdict(settings)
|
|
self.save()
|
|
logger.info("Import settings updated")
|
|
|
|
def get_import_settings(self) -> ImportSettings:
|
|
"""Get import settings (method version for compatibility)."""
|
|
try:
|
|
settings_data = self._config.get("import_settings", {})
|
|
return ImportSettings(**settings_data)
|
|
except Exception as e:
|
|
logger.warning(f"Error loading import settings: {e}")
|
|
return ImportSettings()
|
|
|
|
def save_import_settings(self, settings: ImportSettings):
|
|
"""Save import settings (method version for compatibility)."""
|
|
self._config["import_settings"] = asdict(settings)
|
|
self.save()
|
|
logger.info("Import settings saved")
|
|
|
|
@property
|
|
def recent_files(self) -> List[str]:
|
|
"""Get list of recently opened files."""
|
|
return self._config.get("recent_files", [])
|
|
|
|
def add_recent_file(self, filepath: str):
|
|
"""Add file to recent files list."""
|
|
files = self.recent_files
|
|
if filepath in files:
|
|
files.remove(filepath)
|
|
files.insert(0, filepath)
|
|
self._config["recent_files"] = files[:10] # Keep only 10 recent
|
|
self.save()
|
|
|
|
@property
|
|
def qbo_environment(self) -> str:
|
|
"""Get QBO environment (sandbox/production)."""
|
|
return self._config.get("qbo_environment", "sandbox")
|
|
|
|
@qbo_environment.setter
|
|
def qbo_environment(self, env: str):
|
|
"""Set QBO environment."""
|
|
if env not in ("sandbox", "production"):
|
|
raise ValueError("Environment must be 'sandbox' or 'production'")
|
|
self._config["qbo_environment"] = env
|
|
self.save()
|
|
logger.info(f"QBO environment set to: {env}")
|
|
|
|
# Credential storage (file-based for web app)
|
|
def get_credentials(self) -> QBOCredentials:
|
|
"""Get QBO credentials from file storage."""
|
|
if self._credentials is None:
|
|
self._credentials = QBOCredentials()
|
|
|
|
try:
|
|
if self.credentials_file.exists():
|
|
with open(self.credentials_file, 'r') as f:
|
|
creds_data = json.load(f)
|
|
self._credentials = QBOCredentials(**creds_data)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load credentials: {e}")
|
|
|
|
return self._credentials
|
|
|
|
def save_credentials(self, credentials: QBOCredentials):
|
|
"""Save QBO credentials to file storage."""
|
|
self._credentials = credentials
|
|
try:
|
|
with open(self.credentials_file, 'w') as f:
|
|
json.dump(asdict(credentials), f)
|
|
# Restrict file permissions (Unix only, skip on Windows)
|
|
try:
|
|
os.chmod(self.credentials_file, 0o600)
|
|
except (OSError, AttributeError):
|
|
pass # chmod not supported on Windows
|
|
logger.info("Credentials saved successfully")
|
|
except Exception as e:
|
|
logger.error(f"Failed to save credentials: {e}")
|
|
|
|
def clear_credentials(self):
|
|
"""Clear stored credentials."""
|
|
self._credentials = None
|
|
if self.credentials_file.exists():
|
|
self.credentials_file.unlink()
|
|
logger.info("Credentials cleared")
|
|
|
|
# Connection Profile management
|
|
def get_profiles(self) -> List[ConnectionProfile]:
|
|
"""Get all connection profiles."""
|
|
profiles = []
|
|
profiles_file = self.config_dir / "profiles.json"
|
|
|
|
if profiles_file.exists():
|
|
try:
|
|
with open(profiles_file, 'r') as f:
|
|
profiles_data = json.load(f)
|
|
for p in profiles_data:
|
|
profiles.append(ConnectionProfile(**p))
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load profiles: {e}")
|
|
|
|
return sorted(profiles, key=lambda p: p.name)
|
|
|
|
def get_profile(self, name: str) -> Optional[ConnectionProfile]:
|
|
"""Get a specific profile by name."""
|
|
profiles = self.get_profiles()
|
|
for p in profiles:
|
|
if p.name == name:
|
|
return p
|
|
return None
|
|
|
|
def save_profile(self, profile: ConnectionProfile):
|
|
"""Save a connection profile."""
|
|
profile.updated_at = datetime.now().isoformat()
|
|
profiles = self.get_profiles()
|
|
|
|
# Update existing or add new
|
|
found = False
|
|
for i, p in enumerate(profiles):
|
|
if p.name == profile.name:
|
|
profiles[i] = profile
|
|
found = True
|
|
break
|
|
|
|
if not found:
|
|
profiles.append(profile)
|
|
|
|
# Save to file
|
|
profiles_file = self.config_dir / "profiles.json"
|
|
try:
|
|
with open(profiles_file, 'w') as f:
|
|
json.dump([asdict(p) for p in profiles], f, indent=2)
|
|
logger.info(f"Profile saved: {profile.name}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to save profile: {e}")
|
|
|
|
def delete_profile(self, name: str):
|
|
"""Delete a connection profile."""
|
|
profiles = self.get_profiles()
|
|
profiles = [p for p in profiles if p.name != name]
|
|
|
|
profiles_file = self.config_dir / "profiles.json"
|
|
try:
|
|
with open(profiles_file, 'w') as f:
|
|
json.dump([asdict(p) for p in profiles], f, indent=2)
|
|
logger.info(f"Profile deleted: {name}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete profile: {e}")
|
|
|
|
def get_active_profile_name(self) -> Optional[str]:
|
|
"""Get the name of the currently active profile."""
|
|
return self._config.get("active_profile", None)
|
|
|
|
def set_active_profile(self, name: Optional[str]):
|
|
"""Set the active profile name."""
|
|
self._config["active_profile"] = name
|
|
self.save()
|
|
logger.info(f"Active profile set to: {name}")
|
|
|
|
# Template management
|
|
def get_templates(self, data_type: Optional[str] = None) -> List[MappingTemplate]:
|
|
"""Get all mapping templates, optionally filtered by data type."""
|
|
templates = []
|
|
|
|
for template_file in self.templates_dir.glob("*.json"):
|
|
try:
|
|
with open(template_file, 'r') as f:
|
|
data = json.load(f)
|
|
# Reconstruct FieldMapping objects
|
|
mappings = [FieldMapping(**m) for m in data.get("mappings", [])]
|
|
template = MappingTemplate(
|
|
name=data["name"],
|
|
data_type=data["data_type"],
|
|
mappings=mappings,
|
|
created_at=data.get("created_at", ""),
|
|
updated_at=data.get("updated_at", "")
|
|
)
|
|
if data_type is None or template.data_type == data_type:
|
|
templates.append(template)
|
|
except (json.JSONDecodeError, IOError, KeyError) as e:
|
|
logger.warning(f"Failed to load template {template_file}: {e}")
|
|
continue
|
|
|
|
return sorted(templates, key=lambda t: t.name)
|
|
|
|
def save_template(self, template: MappingTemplate):
|
|
"""Save a mapping template."""
|
|
template.updated_at = datetime.now().isoformat()
|
|
|
|
# Convert to serializable format
|
|
data = {
|
|
"name": template.name,
|
|
"data_type": template.data_type,
|
|
"mappings": [asdict(m) for m in template.mappings],
|
|
"created_at": template.created_at,
|
|
"updated_at": template.updated_at
|
|
}
|
|
|
|
# Safe filename
|
|
safe_name = "".join(c for c in template.name if c.isalnum() or c in " -_").strip()
|
|
filepath = self.templates_dir / f"{safe_name}.json"
|
|
|
|
with open(filepath, 'w') as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
logger.info(f"Template saved: {template.name}")
|
|
|
|
def delete_template(self, template_name: str):
|
|
"""Delete a mapping template."""
|
|
safe_name = "".join(c for c in template_name if c.isalnum() or c in " -_").strip()
|
|
filepath = self.templates_dir / f"{safe_name}.json"
|
|
|
|
if filepath.exists():
|
|
filepath.unlink()
|
|
logger.info(f"Template deleted: {template_name}")
|
|
|
|
def get_default_templates(self) -> Dict[str, MappingTemplate]:
|
|
"""Get built-in default templates for each data type."""
|
|
return {
|
|
"check": MappingTemplate(
|
|
name="Default Check Import",
|
|
data_type="check",
|
|
mappings=[
|
|
FieldMapping("Payee", "EntityRef.value", required=True),
|
|
FieldMapping("Bank Acct", "AccountRef.value", required=True),
|
|
FieldMapping("Date", "TxnDate", transform="date", required=True),
|
|
FieldMapping("Amount", "Line.Amount", transform="currency", required=True),
|
|
FieldMapping("Number", "DocNumber"),
|
|
FieldMapping("Posting Acct", "Line.AccountBasedExpenseLineDetail.AccountRef.value"),
|
|
FieldMapping("Period Memo", "PrivateNote"),
|
|
FieldMapping("Period Memo", "Line.Description"),
|
|
]
|
|
),
|
|
"invoice": MappingTemplate(
|
|
name="Default Invoice Import",
|
|
data_type="invoice",
|
|
mappings=[
|
|
FieldMapping("CustomerRef", "CustomerRef.value", required=True),
|
|
FieldMapping("TxnDate", "TxnDate", transform="date", required=True),
|
|
FieldMapping("DueDate", "DueDate", transform="date"),
|
|
FieldMapping("DocNumber", "DocNumber"),
|
|
FieldMapping("ItemRef", "Line.SalesItemLineDetail.ItemRef.value"),
|
|
FieldMapping("Description", "Line.Description"),
|
|
FieldMapping("Qty", "Line.SalesItemLineDetail.Qty", transform="number"),
|
|
FieldMapping("UnitPrice", "Line.SalesItemLineDetail.UnitPrice", transform="currency"),
|
|
FieldMapping("Amount", "Line.Amount", transform="currency"),
|
|
FieldMapping("PrivateNote", "PrivateNote"),
|
|
]
|
|
),
|
|
"bill": MappingTemplate(
|
|
name="Default Bill Import",
|
|
data_type="bill",
|
|
mappings=[
|
|
FieldMapping("VendorRef", "VendorRef.value", required=True),
|
|
FieldMapping("TxnDate", "TxnDate", transform="date", required=True),
|
|
FieldMapping("DueDate", "DueDate", transform="date"),
|
|
FieldMapping("DocNumber", "DocNumber"),
|
|
FieldMapping("AccountRef", "Line.AccountBasedExpenseLineDetail.AccountRef.value"),
|
|
FieldMapping("Description", "Line.Description"),
|
|
FieldMapping("Amount", "Line.Amount", transform="currency"),
|
|
FieldMapping("PrivateNote", "PrivateNote"),
|
|
]
|
|
),
|
|
"customer": MappingTemplate(
|
|
name="Default Customer Import",
|
|
data_type="customer",
|
|
mappings=[
|
|
FieldMapping("DisplayName", "DisplayName", required=True),
|
|
FieldMapping("CompanyName", "CompanyName"),
|
|
FieldMapping("GivenName", "GivenName"),
|
|
FieldMapping("FamilyName", "FamilyName"),
|
|
FieldMapping("Email", "PrimaryEmailAddr.Address"),
|
|
FieldMapping("Phone", "PrimaryPhone.FreeFormNumber"),
|
|
FieldMapping("Street", "BillAddr.Line1"),
|
|
FieldMapping("City", "BillAddr.City"),
|
|
FieldMapping("State", "BillAddr.CountrySubDivisionCode"),
|
|
FieldMapping("PostalCode", "BillAddr.PostalCode"),
|
|
FieldMapping("Country", "BillAddr.Country"),
|
|
]
|
|
),
|
|
"vendor": MappingTemplate(
|
|
name="Default Vendor Import",
|
|
data_type="vendor",
|
|
mappings=[
|
|
FieldMapping("DisplayName", "DisplayName", required=True),
|
|
FieldMapping("CompanyName", "CompanyName"),
|
|
FieldMapping("GivenName", "GivenName"),
|
|
FieldMapping("FamilyName", "FamilyName"),
|
|
FieldMapping("Email", "PrimaryEmailAddr.Address"),
|
|
FieldMapping("Phone", "PrimaryPhone.FreeFormNumber"),
|
|
FieldMapping("Street", "BillAddr.Line1"),
|
|
FieldMapping("City", "BillAddr.City"),
|
|
FieldMapping("State", "BillAddr.CountrySubDivisionCode"),
|
|
FieldMapping("PostalCode", "BillAddr.PostalCode"),
|
|
FieldMapping("TaxIdentifier", "TaxIdentifier"),
|
|
]
|
|
),
|
|
"account": MappingTemplate(
|
|
name="Default Chart of Accounts Import",
|
|
data_type="account",
|
|
mappings=[
|
|
FieldMapping("Name", "Name", required=True),
|
|
FieldMapping("AccountType", "AccountType", required=True),
|
|
FieldMapping("AccountSubType", "AccountSubType"),
|
|
FieldMapping("AcctNum", "AcctNum"),
|
|
FieldMapping("Description", "Description"),
|
|
FieldMapping("CurrentBalance", "CurrentBalance", transform="currency"),
|
|
]
|
|
),
|
|
} |