"""定时任务模块""" import logging from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger from app.tasks.delivery_warning import check_delivery_warnings, cleanup_expired_warnings from app.tasks.order_payment_timeout import run_close_timeout_online_orders from app.tasks.order_auto_confirm_receive import run_auto_confirm_receive_job logger = logging.getLogger(__name__) scheduler = BackgroundScheduler() def init_scheduled_tasks(get_db_session): """初始化定时任务 Args: get_db_session: 数据库会话获取函数(从 app.core.database 导入) """ # 配送时效预警检查 - 每5分钟执行一次 scheduler.add_job( func=_run_delivery_warning_check, trigger=IntervalTrigger(minutes=5), args=[get_db_session], id="delivery_warning_check", name="配送时效预警检查", replace_existing=True ) # 清理过期预警记录 - 每天凌晨2点执行 scheduler.add_job( func=_run_cleanup_expired_warnings, trigger=IntervalTrigger(hours=24), args=[get_db_session], id="cleanup_expired_warnings", name="清理过期预警记录", replace_existing=True ) # 线上订单超时未支付关闭 - 每1分钟执行 scheduler.add_job( func=_run_close_timeout_online_orders, trigger=IntervalTrigger(minutes=1), args=[get_db_session], id="close_timeout_online_orders", name="线上超时未支付订单关闭", replace_existing=True ) # 送达后超时未确认收货 - 每小时执行 scheduler.add_job( func=_run_auto_confirm_receive_overdue, trigger=IntervalTrigger(hours=1), args=[get_db_session], id="auto_confirm_receive_overdue", name="送达超时自动确认收货", replace_existing=True ) scheduler.start() logger.info("定时任务调度器已启动") def _run_delivery_warning_check(get_db_session): """执行配送时效预警检查(定时任务包装)""" from app.core.database import SessionLocal db = SessionLocal() try: result = check_delivery_warnings(db) logger.info(f"配送时效预警检查完成: 检查了 {result.get('checked', 0)} 个任务," f"创建了 {result.get('warnings_created', 0)} 个预警") except Exception as e: logger.error(f"配送时效预警检查失败: {str(e)}") finally: db.close() def _run_cleanup_expired_warnings(get_db_session): """清理过期预警记录(定时任务包装)""" from app.core.database import SessionLocal db = SessionLocal() try: count = cleanup_expired_warnings(db, days=7) logger.info(f"清理过期预警记录完成: 删除了 {count} 条记录") except Exception as e: logger.error(f"清理过期预警记录失败: {str(e)}") finally: db.close() def _run_close_timeout_online_orders(get_db_session): """关闭线上超时未支付订单(定时任务包装)""" from app.core.database import SessionLocal db = SessionLocal() try: run_close_timeout_online_orders(db) except Exception as e: logger.error(f"线上超时未支付订单关闭失败: {str(e)}") finally: db.close() def _run_auto_confirm_receive_overdue(get_db_session): from app.core.database import SessionLocal db = SessionLocal() try: run_auto_confirm_receive_job(db) except Exception as e: logger.error(f"自动确认收货任务失败: {str(e)}") finally: db.close() def shutdown_scheduler(): """关闭定时任务调度器""" if scheduler.running: scheduler.shutdown(wait=False) logger.info("定时任务调度器已关闭") __all__ = [ "check_delivery_warnings", "cleanup_expired_warnings", "scheduler", "init_scheduled_tasks", "shutdown_scheduler" ]