#!/usr/bin/env python3 """ run_scheduler.py — momo-scheduler 容器入口點 排程任務清單(對齊 app.py init_scheduler + scheduler.py 全任務): 每 30 分鐘:auto_import、whitepage_check 每 1 小時:momo、edm、festival 每 4 小時:competitor_price_feeder、icaim_analysis 每 6 小時:openclaw_meta_analysis、quality_rescore 每 12 小時:dedup_batch 每 1 天 :db_backup(03:00)、cleanup_agent_context(03:30)、backup_monitor(04:00)、daily_report(09:00)、ai_smoke_summary(09:10)、pchome_match_backfill(10:30) 每 1 週 :weekly_strategy(週一 06:00) 每 1 月 :monthly_report(每月1日 07:00) """ import asyncio import logging import threading import time import schedule # 匯入全部排程任務函式 from scheduler import ( run_momo_task, run_edm_task, run_festival_task, run_promo_event_task, run_auto_import_task, run_whitepage_check, run_competitor_price_feeder_task, run_pchome_match_backfill_task, run_icaim_analysis_task, run_weekly_strategy_task, run_db_backup_task, run_backup_monitor_task, run_openclaw_meta_analysis_task, run_dedup_batch_task, run_quality_rescore_task, run_daily_report_task, run_ai_smoke_daily_summary_task, run_monthly_report_task, ) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', ) logger = logging.getLogger(__name__) def _register_schedules(): schedule.every(30).minutes.do(run_auto_import_task) logger.info("📅 每 30 分鐘:auto_import") schedule.every(30).minutes.do(run_whitepage_check) logger.info("📅 每 30 分鐘:whitepage_check") schedule.every(1).hours.do(run_momo_task) logger.info("📅 每 1 小時:momo_task") schedule.every(1).hours.do(run_edm_task) logger.info("📅 每 1 小時:edm_task") schedule.every(1).hours.do(run_festival_task) logger.info("📅 每 1 小時:festival_task") # 動態註冊促銷活動爬蟲(根據配置) from services.crawler_config_loader import get_enabled_crawlers enabled_crawlers = get_enabled_crawlers() promo_event_configs = { 'mothers_day_2026': {'lpn': '', 'page_type': 'mothers_day', 'name': '母親節超值限時購'}, 'valentine_520_2026': {'lpn': '', 'page_type': 'valentine_520', 'name': '520情人節限定購物'}, 'labor_day_2026': {'lpn': '', 'page_type': 'labor_day', 'name': '勞動節購物優惠'} } for crawler_key, config in enabled_crawlers.items(): if crawler_key in promo_event_configs: event_config = promo_event_configs[crawler_key] lpn_code = config.get('lpn_code', '') if lpn_code: schedule_hours = config.get('schedule_hours', 4) schedule.every(schedule_hours).hours.do( lambda lpn=lpn_code, pt=event_config['page_type'], an=event_config['name']: run_promo_event_task(lpn, pt, an) ) logger.info(f"📅 每 {schedule_hours} 小時:{event_config['name']} ({event_config['page_type']})") else: logger.warning(f"⚠️ {event_config['name']} 未配置 LPN 代碼,跳過排程") schedule.every(4).hours.do(run_competitor_price_feeder_task) logger.info("📅 每 4 小時:competitor_price_feeder") schedule.every(4).hours.do(run_icaim_analysis_task) logger.info("📅 每 4 小時:icaim_analysis") schedule.every(6).hours.do(run_openclaw_meta_analysis_task) logger.info("📅 每 6 小時:openclaw_meta_analysis") schedule.every(6).hours.do(run_quality_rescore_task) logger.info("📅 每 6 小時:quality_rescore") schedule.every(12).hours.do(run_dedup_batch_task) logger.info("📅 每 12 小時:dedup_batch") schedule.every().day.at("03:00").do(run_db_backup_task) logger.info("📅 每日 03:00:db_backup") schedule.every().day.at("03:30").do(run_cleanup_agent_context) logger.info("📅 每日 03:30:cleanup_agent_context") schedule.every().day.at("04:00").do(run_backup_monitor_task) logger.info("📅 每日 04:00:backup_monitor") schedule.every().monday.at("06:00").do(run_weekly_strategy_task) logger.info("📅 每週一 06:00:weekly_strategy") schedule.every().day.at("09:00").do(run_daily_report_task) logger.info("📅 每日 09:00:daily_report") schedule.every().day.at("09:10").do(run_ai_smoke_daily_summary_task) logger.info("📅 每日 09:10:ai_smoke_daily_summary") schedule.every().day.at("10:30").do(run_pchome_match_backfill_task) logger.info("📅 每日 10:30:pchome_match_backfill") # 每月1日 07:00 月報(schedule 不支援 every().month,用每日 07:00 + 日期判斷) def _monthly_report_gate(): from datetime import datetime as _dt if _dt.now().day == 1: run_monthly_report_task() schedule.every().day.at("07:00").do(_monthly_report_gate) logger.info("📅 每月1日 07:00:monthly_report") def run_cleanup_agent_context(): """每日 03:30 — 清理 agent_context 表中已過期的 TTL 記錄(migration 018 定義)""" from database.manager import get_session from sqlalchemy import text session = get_session() try: session.execute(text("SELECT cleanup_expired_agent_context()")) session.commit() logger.info("[Cleanup] agent_context TTL 清理完成") except Exception as e: logger.error(f"[Cleanup] agent_context 清理失敗: {e}") try: from services.event_router import notify_failure notify_failure( task_name="run_cleanup_agent_context", error=e, source="Scheduler.Cleanup", event_type="agent_context_cleanup_failure", priority="P2", title="agent_context TTL 清理失敗", dedup_ttl_sec=3600, ) except Exception as _router_e: logger.error(f"[Cleanup] event_router 失敗: {_router_e}") finally: session.close() def _run_elephant_alpha_engine(): """Daemon thread: ElephantAlpha 自主監控引擎(獨立 asyncio loop)""" loop = None try: from services.elephant_alpha_autonomous_engine import autonomous_engine loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) logger.info("🐘 [ElephantAlpha] Autonomous engine thread started") loop.run_until_complete(autonomous_engine.start_autonomous_monitoring()) except Exception as e: logger.error(f"🐘 [ElephantAlpha] Engine crashed: {e}") finally: if loop is not None: loop.close() if __name__ == "__main__": logger.info("🚀 momo-scheduler 啟動中...") _register_schedules() logger.info("✅ 全部排程任務已註冊") _ea_thread = threading.Thread( target=_run_elephant_alpha_engine, daemon=True, name="elephant-alpha-engine", ) _ea_thread.start() logger.info("🐘 [ElephantAlpha] Autonomous engine thread launched") logger.info("⏰ 排程主迴圈啟動,等待任務觸發...") _ea_watchdog_counter = 0 # 每 60 秒(60 次 sleep(1))做一次存活檢查 while True: try: schedule.run_pending() time.sleep(1) # 每 60 秒檢查 ElephantAlpha 執行緒是否還活著 _ea_watchdog_counter += 1 if _ea_watchdog_counter >= 60: _ea_watchdog_counter = 0 if not _ea_thread.is_alive(): logger.error("[ElephantAlpha] 監控執行緒已死亡,嘗試重啟") try: from services.event_router import dispatch_sync as _dispatch_sync _dispatch_sync({ "source": "Scheduler.ElephantAlpha", "event_type": "thread_crashed", "severity": "alert", "title": "ElephantAlpha 執行緒死亡", "status": "自動重啟中", "impact": "P2 - 自主監控引擎暫停", "summary": "ElephantAlphaEngine daemon thread 意外終止,排程主迴圈已偵測並觸發重啟", }) except Exception as _alert_err: logger.error(f"[ElephantAlpha] 無法發送告警: {_alert_err}") _ea_thread = threading.Thread( target=_run_elephant_alpha_engine, daemon=True, name="ElephantAlphaEngine", ) _ea_thread.start() logger.info("[ElephantAlpha] 執行緒已重啟") except KeyboardInterrupt: logger.info("⛔ Scheduler stopped.") break except Exception as e: logger.error(f"Scheduler error: {e}") time.sleep(5)