""" 定时任务调度器 """ import logging import asyncio from datetime import datetime from typing import Optional from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger from apscheduler.jobstores.memory import MemoryJobStore from apscheduler.executors.asyncio import AsyncIOExecutor from app.core.config import settings from app.core.database import AsyncSessionLocal from app.services.import_service import ImportService logger = logging.getLogger(__name__) class OrderScheduler: """订单文件监控调度器""" def __init__(self): self.scheduler: Optional[AsyncIOScheduler] = None self.import_service = ImportService() self._is_running = False async def start(self): """启动调度器""" if self._is_running: logger.warning("Scheduler is already running") return # 配置调度器 jobstores = { 'default': MemoryJobStore() } executors = { 'default': AsyncIOExecutor() } job_defaults = { 'coalesce': True, # 合并堆积的任务 'max_instances': 1, # 同时只运行一个实例 'misfire_grace_time': 60 # 错过执行时间的容忍度（秒） } self.scheduler = AsyncIOScheduler( jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone='Asia/Shanghai' ) # 添加文件监控任务（根据配置决定是否启用自动调度） if not settings.DRY_RUN_ANALYZE_ONLY: if settings.AUTO_SCHEDULE_ENABLED: # 启用自动调度 self.scheduler.add_job( self._scan_and_import, trigger=IntervalTrigger(minutes=settings.IMPORT_INTERVAL_MINUTES), id='file_import_job', name='File Import Job', replace_existing=True ) logger.info(f"Auto-schedule enabled: File import job scheduled every {settings.IMPORT_INTERVAL_MINUTES} minutes") else: # 禁用自动调度，仅支持手动触发 logger.info("Auto-schedule disabled: Scheduler started in manual trigger mode only") else: logger.info("Scheduler started in DRY_RUN mode - no import jobs scheduled") # 启动调度器 self.scheduler.start() self._is_running = True logger.info("Order scheduler started successfully") async def stop(self): """停止调度器""" if not self._is_running: return if self.scheduler: self.scheduler.shutdown(wait=True) logger.info("Order scheduler stopped") self._is_running = False async def trigger_import_now(self) -> dict: """立即触发一次导入""" logger.info("Manual import triggered") return await self._scan_and_import() async def _scan_and_import(self) -> dict: """扫描并导入文件的主要逻辑 - 统一流程：导入→产品主表→采购订单""" start_time = datetime.now() result = { 'start_time': start_time, 'status': 'started', 'files_processed': 0, 'total_imported': 0, 'total_failed': 0, 'products_generated': 0, 'procurement_orders_created': 0, 'errors': [] } try: # 创建数据库会话 async with AsyncSessionLocal() as db: # ============ 步骤1: 扫描并导入文件 ============ logger.info("[Step 1/3] Starting file import...") import_results = await self.import_service.import_all_pending_files(db) # 统计导入结果 files_processed = len(import_results) total_imported = sum(r.get('rows_imported', 0) for r in import_results) total_failed = sum(r.get('rows_failed', 0) for r in import_results) # 收集错误 all_errors = [] for r in import_results: if r.get('errors'): all_errors.extend(r['errors']) if r.get('error_message'): all_errors.append(f"{r['file_name']}: {r['error_message']}") result.update({ 'files_processed': files_processed, 'total_imported': total_imported, 'total_failed': total_failed, 'errors': all_errors, 'file_results': import_results }) # 只有当有新数据导入时才继续后续步骤 if total_imported > 0: logger.info(f"[Step 1/4] Imported {total_imported} new records") # 添加延迟以确保数据库事务完成 delay_seconds = settings.PRODUCTS_GENERATION_DELAY logger.info(f"Waiting {delay_seconds} seconds to ensure database consistency...") await asyncio.sleep(delay_seconds) # ============ 步骤2: 生成产品主表 ============ logger.info("[Step 2/4] Generating products master...") try: from app.services.products_master_service import ProductsMasterService products_service = ProductsMasterService() # 增量生成产品主表（不强制重建） products_result = await products_service.generate_products_master( db, force_rebuild=False, batch_size=1000 ) result['products_generated'] = products_result.get('created', 0) logger.info(f"[Step 2/4] Generated {products_result.get('created', 0)} new products, " f"updated {products_result.get('updated', 0)} products") # ============ 步骤2.5: 自动批准产品（如果启用）============ if settings.PRODUCTS_AUTO_APPROVE and products_result.get('created', 0) > 0: logger.info(f"[Step 2.5/4] Auto-approving products after {settings.PRODUCTS_APPROVE_TIMEOUT} seconds...") await asyncio.sleep(settings.PRODUCTS_APPROVE_TIMEOUT) # 这里可以添加产品批准逻辑 # await products_service.auto_approve_products(db) logger.info("[Step 2.5/4] Products auto-approved") except Exception as e: logger.error(f"[Step 2/4] Products master generation failed: {e}") result['errors'].append(f"产品主表生成失败: {str(e)}") # 产品批准后的延迟 if settings.PROCUREMENT_GENERATION_DELAY > 0: logger.info(f"Waiting {settings.PROCUREMENT_GENERATION_DELAY} seconds before generating procurement orders...") await asyncio.sleep(settings.PROCUREMENT_GENERATION_DELAY) # ============ 步骤3: 同步采购订单 ============ logger.info("[Step 3/5] Syncing procurement orders...") try: from app.services.procurement_order_service_v2 import ProcurementOrderServiceV2 procurement_service = ProcurementOrderServiceV2() # 增量同步采购订单 procurement_result = await procurement_service.generate_procurement_orders( db=db, force_rebuild=False, batch_size=1000 ) result['procurement_orders_created'] = procurement_result.get('created_orders', 0) logger.info(f"[Step 3/5] Created {procurement_result.get('created_orders', 0)} procurement orders") except Exception as e: logger.error(f"[Step 3/5] Procurement orders sync failed: {e}") result['errors'].append(f"采购订单同步失败: {str(e)}") # ============ 步骤4: 同步采购订单状态 ============ logger.info("[Step 4/5] Syncing procurement order status...") try: from app.services.procurement_order_sync_service import ProcurementOrderSyncService sync_service = ProcurementOrderSyncService() # 同步所有采购订单的状态 sync_result = await sync_service.sync_order_status(db) result['orders_status_synced'] = sync_result.get('synced_orders', 0) logger.info(f"[Step 4/5] Synced status for {sync_result.get('synced_orders', 0)} procurement orders") except Exception as e: logger.error(f"[Step 4/5] Status sync failed: {e}") result['errors'].append(f"状态同步失败: {str(e)}") # ============ 步骤5: 图片下载已禁用 ============ # 改为使用在线URL，不再自动下载图片到本地 # try: # from app.services.image_download_service import ImageDownloadService # image_service = ImageDownloadService() # # # 创建独立的异步任务进行图片下载 # asyncio.create_task( # self._download_images_async(image_service) # ) # logger.info("[Step 4/4] Image download task scheduled in background") # # except Exception as e: # logger.error(f"[Step 5/5] Failed to schedule image download: {e}") # # 图片下载失败不影响主流程 logger.info("[Step 5/5] Skipped - Using online image URLs directly") else: logger.info("[Step 1/5] No new data imported, skipping products and procurement generation") result['status'] = 'completed' end_time = datetime.now() duration = (end_time - start_time).total_seconds() logger.info(f"Complete cycle finished in {duration:.2f}s: " f"{files_processed} files, {total_imported} imported, " f"{result['products_generated']} products, " f"{result['procurement_orders_created']} procurement orders") except Exception as e: logger.error(f"Complete cycle failed: {e}") result.update({ 'status': 'error', 'error_message': str(e), 'errors': [str(e)] }) return result async def _download_images_async(self, image_service): """异步批量下载图片 - 后台执行（两阶段处理）""" try: # 创建独立的数据库会话 from app.core.database import AsyncSessionLocal async with AsyncSessionLocal() as db: # 延迟5秒后开始下载，确保产品主表已完全生成 # 两阶段处理：1. 立即使用在线URL 2. 后台下载到本地 result = await image_service.delayed_batch_download( db, delay_seconds=5, use_online_first=True # 启用两阶段处理 ) logger.info(f"Background image download completed (two-phase): {result}") except Exception as e: logger.error(f"Background image download failed: {e}") def get_job_info(self) -> dict: """获取任务信息""" if not self.scheduler: return {'status': 'not_started'} jobs = [] for job in self.scheduler.get_jobs(): jobs.append({ 'id': job.id, 'name': job.name, 'next_run': job.next_run_time.isoformat() if job.next_run_time else None, 'trigger': str(job.trigger) }) return { 'status': 'running' if self._is_running else 'stopped', 'jobs': jobs } # 全局调度器实例 scheduler_instance: Optional[OrderScheduler] = None async def get_scheduler() -> OrderScheduler: """获取全局调度器实例""" global scheduler_instance if scheduler_instance is None: scheduler_instance = OrderScheduler() return scheduler_instance async def start_scheduler(): """启动全局调度器""" scheduler = await get_scheduler() await scheduler.start() async def stop_scheduler(): """停止全局调度器""" global scheduler_instance if scheduler_instance: await scheduler_instance.stop() scheduler_instance = None