# inventory-backend/run.py import sys from app import create_app # ★ stdout 行缓冲。 # gunicorn 的 stdout 是管道,Python 默认对它做**块缓冲** —— print() 的内容 # 会攒在缓冲区里,docker logs 里看不到。实测 17:30 的日报任务:8 个 worker # 里只刷出 2 条日志,连抢到锁那个 worker 的成败都没出现,完全无从排查。 # 调度任务是后台线程,出问题本来就不容易发现,再叠加缓冲就等于瞎跑。 if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(line_buffering=True) app = create_app() # ========================================================= # 定时任务注册 # # ★★ 每个任务都必须套 advisory_lock,否则会被执行 **8 次**。 # # gunicorn.conf.py 配了 8 个 worker,且未开 preload_app —— 每个 worker # 都会独立 import 一次本文件,于是在模块级启动的这份调度器会**每个 # worker 各起一份**,同一个 cron 任务在相同时刻被并发执行 8 次。 # (库存预警此前正是如此:5 条预警配置,每天实际发 8 封重复邮件。 # APScheduler 的 max_instances 只管单个调度器实例内,管不了跨进程。) # # advisory_lock 用 PostgreSQL 会话级咨询锁做真正的跨进程互斥, # 抢到锁的 worker 才执行,其余静默跳过。详见 app/utils/job_lock.py。 # ========================================================= from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger import pytz beijing_tz = pytz.timezone('Asia/Shanghai') # 调度日志一律再加一层 flush=True(理由见文件头部的行缓冲说明)—— # 定时任务跑在后台线程,出问题时唯一的线索就是这几行输出,不能丢。 def _run_warning_job(): """库存预警扫描与邮件发送(每天 9:30 北京时间)""" with app.app_context(): from app.utils.job_lock import advisory_lock, LOCK_INVENTORY_WARNING with advisory_lock(LOCK_INVENTORY_WARNING) as acquired: if not acquired: # 正常路径:另一个 worker 正在执行同一任务 print("[Scheduler] 库存预警:另一 worker 正在执行,本轮跳过", flush=True) return try: from app.services.inventory_task import InventoryWarningService result = InventoryWarningService.check_and_send_warning_emails() print(f"[Scheduler] 库存预警扫描完成: red={result['red_count']}, yellow={result['yellow_count']}", flush=True) except Exception as e: print(f"[Scheduler] 库存预警任务失败: {e}", flush=True) def _run_daily_report_job(): """MOM 系统日报(每天 17:30 北京时间)""" with app.app_context(): from app.utils.job_lock import advisory_lock, LOCK_DAILY_REPORT with advisory_lock(LOCK_DAILY_REPORT) as acquired: if not acquired: print("[Scheduler] 系统日报:另一 worker 正在执行,本轮跳过", flush=True) return try: from app.services.daily_report_service import DailyReportService r = DailyReportService.send_daily_report() print(f"[Scheduler] 系统日报已发送: {r['subject']} -> {r['recipients']}", flush=True) except Exception as e: print(f"[Scheduler] 系统日报任务失败: {e}", flush=True) scheduler = BackgroundScheduler(timezone=beijing_tz) scheduler.add_job( func=_run_warning_job, trigger=CronTrigger(hour=9, minute=30, timezone=beijing_tz), id='inventory_warning_daily', name='库存预警每日邮件发送', replace_existing=True ) scheduler.add_job( func=_run_daily_report_job, trigger=CronTrigger(hour=17, minute=30, timezone=beijing_tz), id='daily_report', name='MOM系统日报', replace_existing=True ) scheduler.start() print("✅ 定时任务已启动:库存预警 9:30 / MOM系统日报 17:30(北京时间)", flush=True) if __name__ == '__main__': # ================================================= # 路由打印调试 (启动时会在控制台列出所有 URL) # 这一步能帮你确认 /api/inbound/base/list 是否存在 # ================================================= print("\n====== 当前生效的路由映射 ======") try: # 按 URL 排序打印,方便查找 sorted_rules = sorted(app.url_map.iter_rules(), key=lambda x: str(x)) for rule in sorted_rules: # 过滤掉一些系统自带的 static 路由,只显示 API if 'api' in str(rule): methods = ','.join(rule.methods - {'OPTIONS', 'HEAD'}) print(f"{str(rule):<50} | {methods:<10} | {rule.endpoint}") except Exception: pass print("==============================\n") # 启动开发服务器 # 端口设置为 5000 (Flask 默认) 或 8000,请确保与前端 Vite 代理一致 app.run(host='0.0.0.0', port=8000, debug=True)