""" Background sync service for fetching new invoices from simPRO API. Runs periodically to keep the database up-to-date. DISABLED: This service is currently disabled as we are using CSV-based imports. The code is kept for reference and potential future reactivation. """ from apscheduler.schedulers.asyncio import AsyncIOScheduler from datetime import datetime import logging import json from sqlalchemy.orm import Session from app.database import SessionLocal from app.models import Invoice, SyncMetadata from app.services.simpro_client import SimProClient from app.config import settings logger = logging.getLogger(__name__) class SyncService: """Service for synchronizing invoice data from simPRO.""" def __init__(self): self.scheduler = AsyncIOScheduler() self.is_syncing = False async def sync_branch(self, branch_id: str, db: Session): """ Sync invoices for a specific branch. DISABLED: SimPro API sync is disabled - using CSV imports instead. Args: branch_id: Branch identifier db: Database session """ logger.info(f"⏭ī¸ Skipping sync for {branch_id} (CSV mode enabled)") return # DISABLED: API sync disabled # DISABLED CODE BELOW - kept for reference """ logger.info(f"🔄 Starting sync for {branch_id}") try: client = SimProClient(branch_id) branch_config = settings.BRANCHES[branch_id] for company_id, company_info in branch_config["companies"].items(): company_name = company_info["name"] # Get last sync metadata sync_meta = db.query(SyncMetadata).filter( SyncMetadata.branch_id == branch_id, SyncMetadata.company_id == company_id ).first() # Determine if this is initial sync or incremental since_date = sync_meta.last_sync_date if sync_meta else None if since_date: logger.info(f"đŸ“Ĩ Incremental sync for {branch_config['name']} - {company_name} since {since_date}") else: logger.info(f"đŸ“Ĩ Initial sync for {branch_config['name']} - {company_name}") # Fetch invoices invoices = await client.fetch_company_invoices( company_id=company_id, year=2025, since_date=since_date ) if not invoices: logger.info(f"✅ No new invoices for {company_name}") continue # Process and store invoices new_count = 0 updated_count = 0 skipped_count = 0 latest_invoice_date = None seen_invoice_ids = set() # Track duplicates within this batch for inv_data in invoices: # Skip duplicates within the same batch if inv_data["ID"] in seen_invoice_ids: skipped_count += 1 continue seen_invoice_ids.add(inv_data["ID"]) # Parse date - store as naive datetime (no timezone) for SQLite compatibility # This ensures consistent date handling across all queries date_str = inv_data["DateIssued"].replace('Z', '+00:00') date_issued = datetime.fromisoformat(date_str).replace(tzinfo=None) # Track latest date if not latest_invoice_date or date_issued > latest_invoice_date: latest_invoice_date = date_issued # Extract Total.ExTax total_ex_tax = 0.0 if isinstance(inv_data.get("Total"), dict): total_ex_tax = float(inv_data["Total"].get("ExTax", 0)) elif isinstance(inv_data.get("Total"), (int, float)): total_ex_tax = float(inv_data["Total"]) # Check if invoice exists (unique by branch_id + company_id + invoice_id) existing = db.query(Invoice).filter( Invoice.branch_id == branch_id, Invoice.company_id == company_id, Invoice.invoice_id == inv_data["ID"] ).first() # Convert Stage and Status to strings (they come as dicts from API) stage_str = None if inv_data.get("Stage"): if isinstance(inv_data["Stage"], dict): stage_str = inv_data["Stage"].get("Name", str(inv_data["Stage"])) else: stage_str = str(inv_data["Stage"]) status_str = None if inv_data.get("Status"): if isinstance(inv_data["Status"], dict): status_str = inv_data["Status"].get("Name", str(inv_data["Status"])) else: status_str = str(inv_data["Status"]) if existing: # Update existing invoice existing.total_ex_tax = total_ex_tax existing.is_paid = inv_data.get("IsPaid", False) existing.stage = stage_str existing.status = status_str existing.customer = json.dumps(inv_data.get("Customer")) if inv_data.get("Customer") else None existing.jobs = json.dumps(inv_data.get("Jobs")) if inv_data.get("Jobs") else None existing.updated_at = datetime.utcnow() updated_count += 1 else: # Create new invoice invoice = Invoice( invoice_id=inv_data["ID"], branch_id=branch_id, company_id=company_id, company_name=company_name, date_issued=date_issued, total_ex_tax=total_ex_tax, is_paid=inv_data.get("IsPaid", False), customer=json.dumps(inv_data.get("Customer")) if inv_data.get("Customer") else None, jobs=json.dumps(inv_data.get("Jobs")) if inv_data.get("Jobs") else None, stage=stage_str, status=status_str ) db.add(invoice) new_count += 1 # Commit changes db.commit() # Update sync metadata total_invoices = db.query(Invoice).filter( Invoice.branch_id == branch_id, Invoice.company_id == company_id ).count() if sync_meta: sync_meta.last_sync_date = latest_invoice_date or sync_meta.last_sync_date sync_meta.last_sync_timestamp = datetime.utcnow() sync_meta.invoice_count = total_invoices else: sync_meta = SyncMetadata( branch_id=branch_id, company_id=company_id, last_sync_date=latest_invoice_date, last_sync_timestamp=datetime.utcnow(), invoice_count=total_invoices ) db.add(sync_meta) db.commit() if skipped_count > 0: logger.info(f"✅ {company_name}: {new_count} new, {updated_count} updated, {skipped_count} skipped duplicates, {total_invoices} total") else: logger.info(f"✅ {company_name}: {new_count} new, {updated_count} updated, {total_invoices} total") except Exception as e: logger.error(f"❌ Error syncing {branch_id}: {e}", exc_info=True) db.rollback() """ async def sync_all_branches(self): """Sync all branches.""" if self.is_syncing: logger.info("⏭ī¸ Sync already in progress, skipping") return self.is_syncing = True logger.info("🔄 Starting sync for all branches") try: db = SessionLocal() try: for branch_id in settings.BRANCHES.keys(): await self.sync_branch(branch_id, db) logger.info("✅ Sync complete for all branches") finally: db.close() finally: self.is_syncing = False def start(self): """Start the background sync scheduler.""" # Run sync immediately on startup self.scheduler.add_job( self.sync_all_branches, 'date', run_date=datetime.now(), id='initial_sync' ) # Schedule periodic sync interval_minutes = settings.SYNC_INTERVAL_SECONDS / 60 self.scheduler.add_job( self.sync_all_branches, 'interval', seconds=settings.SYNC_INTERVAL_SECONDS, id='periodic_sync', replace_existing=True ) self.scheduler.start() logger.info(f"✅ Sync scheduler started (interval: {interval_minutes:.1f} minutes)") def stop(self): """Stop the background sync scheduler.""" self.scheduler.shutdown() logger.info("⏚ī¸ Sync scheduler stopped") # Global sync service instance sync_service = SyncService()