ewoooc/run_scheduler.py

#!/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）
  每  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_auto_import_task,
    run_whitepage_check,
    run_competitor_price_feeder_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_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")

    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")

    # 每月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}")
    finally:
        session.close()


def _run_elephant_alpha_engine():
    """Daemon thread: ElephantAlpha 自主監控引擎（獨立 asyncio loop）"""
    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:
        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 as _dispatch
                        _dispatch({
                            "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)