from __future__ import annotations import asyncio import json import logging from datetime import datetime, timedelta, timezone from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.database import AsyncSessionLocal from app.plugins.xero_import.client import XeroClient from app.plugins.xero_import.models import XeroConnection, XeroSyncState from app.plugins.xero_import.services import XeroImportService logger = logging.getLogger("uvicorn.error") _sync_lock = asyncio.Lock() _last_cleanup: datetime | None = None async def _cleanup_old_logs(): """清理超过 7 天的同步日志，每天最多执行一次。""" global _last_cleanup now = datetime.now(timezone.utc) if _last_cleanup and (now - _last_cleanup) < timedelta(hours=24): return try: cutoff = now - timedelta(days=7) async with AsyncSessionLocal() as db: result = await db.execute( text("DELETE FROM xero_sync_logs WHERE created_at < :cutoff"), {"cutoff": cutoff}, ) await db.commit() deleted = result.rowcount if deleted: logger.info("[xero_scheduler] Cleaned up %d old sync logs (>7 days)", deleted) _last_cleanup = now except Exception as e: logger.error("[xero_scheduler] Log cleanup error: %s", e) async def _scheduler_loop(): logger.info("[xero_scheduler] Loop started, waiting 30s before first run...") await asyncio.sleep(30) logger.info("[xero_scheduler] Initial wait done, entering main loop") while True: try: # 每天清理一次过期日志 await _cleanup_old_logs() if _sync_lock.locked(): logger.debug("[xero_scheduler] Lock busy, skipping this tick") await asyncio.sleep(60) continue async with _sync_lock: await _run_all_tenants() except Exception as e: logger.error("[xero_scheduler] Loop error: %s", e, exc_info=True) await asyncio.sleep(60) def _cfg_bool(cfg: dict, key: str, default: bool = True) -> bool: """插件配置的布尔值以字符串 'true'/'false' 存储，需要正确解析。""" val = cfg.get(key) if val is None: return default if isinstance(val, bool): return val return str(val).lower() == "true" async def _run_all_tenants(): logger.info("[xero_scheduler] _run_all_tenants: checking configs...") async with AsyncSessionLocal() as db: rows = await db.execute( text( "SELECT pc.tenant_id, pc.config " "FROM plugin_configs pc " "WHERE pc.plugin_name = 'xero_import' AND pc.is_active = 1" ) ) configs = rows.fetchall() logger.info("[xero_scheduler] Found %d active xero_import configs", len(configs)) for row in configs: tenant_id = row[0] raw_cfg = row[1] cfg = raw_cfg if isinstance(raw_cfg, dict) else (json.loads(raw_cfg) if raw_cfg else {}) client_id = cfg.get("client_id") client_secret = cfg.get("client_secret") redirect_uri = cfg.get("redirect_uri") if not client_id or not client_secret: logger.warning("[xero_scheduler] Tenant %d: missing client_id/secret, skip", tenant_id) continue client = XeroClient(client_id, client_secret, redirect_uri or "") async with AsyncSessionLocal() as db: try: # 先保活：确保 token 不过期（Xero refresh_token 60 天未用会失效） await _keepalive_token(db, tenant_id, client) tax_mapping_raw = cfg.get("tax_type_mapping") if isinstance(tax_mapping_raw, str): tax_mapping = json.loads(tax_mapping_raw) if tax_mapping_raw else {} elif isinstance(tax_mapping_raw, dict): tax_mapping = tax_mapping_raw else: tax_mapping = {} sync_options = { "sync_price": _cfg_bool(cfg, "sync_price", False), "auto_delist_zero_stock": _cfg_bool(cfg, "auto_delist_zero_stock", True), "tax_type_mapping": tax_mapping, } svc = XeroImportService(db, tenant_id, client, sync_options=sync_options) if _cfg_bool(cfg, "sync_contacts", True): interval = int(cfg.get("contacts_interval_minutes") or 30) should = await _should_sync(db, tenant_id, "contacts", interval) logger.info("[xero_scheduler] Tenant %d contacts: interval=%dm, should_sync=%s", tenant_id, interval, should) if should: await svc.sync_contacts(mode="auto") if _cfg_bool(cfg, "sync_items", True): interval = int(cfg.get("items_interval_minutes") or 10) should = await _should_sync(db, tenant_id, "items", interval) logger.info("[xero_scheduler] Tenant %d items: interval=%dm, should_sync=%s", tenant_id, interval, should) if should: await svc.sync_items(mode="auto") if _cfg_bool(cfg, "push_order_enabled", False): interval = int(cfg.get("push_order_interval_minutes") or 5) should = await _should_sync(db, tenant_id, "invoices", interval) trigger_status = cfg.get("push_order_status", "paid") logger.info("[xero_scheduler] Tenant %d invoices: interval=%dm, status=%s, should_sync=%s", tenant_id, interval, trigger_status, should) if should: await svc.push_pending_orders(trigger_status, mode="auto") except Exception as e: logger.error("[xero_scheduler] Auto sync error for tenant %d: %s", tenant_id, e, exc_info=True) async def _keepalive_token(db: AsyncSession, tenant_id: int, client: XeroClient): """主动刷新 access_token，确保 refresh_token 不会 60 天过期失效。 刷新失败（如 401）时标记连接为非活跃并抛出异常，阻止后续同步。""" result = await db.execute( select(XeroConnection).where( XeroConnection.tenant_id == tenant_id, XeroConnection.is_active == 1, ).limit(1) ) conn = result.scalar_one_or_none() if not conn or not conn.refresh_token: return expires = conn.expires_at.replace(tzinfo=timezone.utc) if conn.expires_at.tzinfo is None else conn.expires_at if expires < datetime.now(timezone.utc): try: token_data = await client.refresh(conn.refresh_token) conn.access_token = token_data["access_token"] conn.refresh_token = token_data["refresh_token"] conn.expires_at = client.expires_at_from_token(token_data) await db.commit() logger.info("[xero_scheduler] Token refreshed for tenant %d", tenant_id) except Exception as e: logger.error("[xero_scheduler] Token refresh failed for tenant %d: %s — deactivating connection", tenant_id, e) conn.is_active = 0 await db.commit() raise async def _should_sync(db: AsyncSession, tenant_id: int, resource: str, interval_minutes: int) -> bool: result = await db.execute( select(XeroSyncState.last_success_at).where( XeroSyncState.tenant_id == tenant_id, XeroSyncState.resource == resource, ).limit(1) ) last = result.scalar_one_or_none() if not last: return True cutoff = datetime.now(timezone.utc) - timedelta(minutes=interval_minutes) last_utc = last.replace(tzinfo=timezone.utc) if last.tzinfo is None else last return last_utc < cutoff def register_scheduler(app): """在 lifespan 中被 load_plugins 调用，直接创建 asyncio task。 注意：不能用 @app.on_event("startup")，因为 FastAPI 使用 lifespan 时 on_event 不会触发。""" logger.info("[xero_scheduler] ====== Registering scheduler task ======") asyncio.ensure_future(_scheduler_loop())