external_demand/app/scheduler.py

"""统一定时任务调度入口。"""

from __future__ import annotations

import argparse
import json
import sys
import time
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Callable

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from app.hot_content.client import JsonApiClient
from app.hot_content.decode_result_service import run_once as run_decode_result_once
from app.hot_content.config import load_flow_config
from app.hot_content.postprocess_service import run_once as run_postprocess_once
from app.hot_content.repository import HotContentRepository
from app.hot_content.service import run_once
from app.hot_content.timezone import SHANGHAI_TZ
from app.hot_content.types import FlowConfig
from app.hot_content.wxindex_words import run_wxindex_words_daily_job
from app.hot_content.wxindex_heat_pattern import run_wxindex_heat_pattern_daily_job
from app.festival_demand.config import load_festival_demand_config
from app.festival_demand.repository import FestivalDemandRepository
from app.festival_demand.service import run_festival_demand_daily_job, print_festival_demand_summary
from app.festival_demand.types import FestivalDemandConfig
from app.gap_script_demand.config import load_gap_script_demand_config
from app.gap_script_demand.repository import GapScriptDemandRepository
from app.gap_script_demand.service import (
    print_gap_script_demand_summary,
    run_gap_script_demand_daily_job,
)
from app.gap_script_demand.types import GapScriptDemandConfig

# 当天失败后自动重试间隔（秒）
SAME_DAY_RETRY_INTERVAL_SECONDS = 10 * 60


def _import_blocking_scheduler() -> Any:
    try:
        from apscheduler.schedulers.blocking import BlockingScheduler
    except ImportError as exc:
        raise RuntimeError("缺少依赖：请先执行 pip install -r requirements.txt") from exc
    return BlockingScheduler


def run_hot_content_job(config: FlowConfig) -> None:
    try:
        summary = run_once(config)
        print(json.dumps(summary, ensure_ascii=False, indent=2))
    except Exception as exc:
        print(f"hot content flow failed: {exc}", file=sys.stderr)


def run_decode_result_job(config: FlowConfig) -> None:
    """解构结果拉取 -> 后处理 -> 写入 ODPS 需求表。"""
    summary: dict[str, Any] = {}
    try:
        summary["decode_result"] = run_decode_result_once(config)
    except Exception as exc:
        summary["decode_result_error"] = str(exc)
        print(f"decode result flow failed: {exc}", file=sys.stderr)
    try:
        summary["postprocess"] = run_postprocess_once(config)
    except Exception as exc:
        summary["postprocess_error"] = str(exc)
        print(f"postprocess flow failed: {exc}", file=sys.stderr)
    print(json.dumps(summary, ensure_ascii=False, indent=2))


def run_postprocess_job(config: FlowConfig) -> None:
    try:
        summary = run_postprocess_once(config)
        print(json.dumps({"job": "postprocess", "summary": summary}, ensure_ascii=False, indent=2))
    except Exception as exc:
        print(f"postprocess flow failed: {exc}", file=sys.stderr)


def run_wxindex_words_refresh_job(config: FlowConfig) -> None:
    repository = HotContentRepository(config.mysql)
    api_client = JsonApiClient(
        timeout_seconds=config.request_timeout_seconds,
        verify_ssl=config.https_verify_ssl,
    )
    try:
        summary = run_wxindex_words_daily_job(
            repository,
            api_client,
            config.wxindex_api_url,
        )
        print(
            json.dumps(
                {"job": "wxindex_words_refresh", "summary": summary},
                ensure_ascii=False,
                indent=2,
            )
        )
    except Exception as exc:
        print(f"wxindex words refresh failed: {exc}", file=sys.stderr)
    finally:
        repository.close()


def run_wxindex_heat_pattern_job(config: FlowConfig) -> None:
    repository = HotContentRepository(config.mysql)
    api_client = JsonApiClient(
        timeout_seconds=config.request_timeout_seconds,
        verify_ssl=config.https_verify_ssl,
    )
    try:
        summary = run_wxindex_heat_pattern_daily_job(
            repository,
            config=config,
            api_client=api_client,
        )
        print(
            json.dumps(
                {"job": "wxindex_heat_pattern", "summary": summary},
                ensure_ascii=False,
                indent=2,
            )
        )
    except Exception as exc:
        print(f"wxindex heat pattern failed: {exc}", file=sys.stderr)
    finally:
        repository.close()


def _today_shanghai() -> date:
    return datetime.now(SHANGHAI_TZ).date()


def _is_write_success(summary: dict[str, Any]) -> bool:
    """MySQL 已写入，或分区已有数据（视为当天写入目标已达成）。"""
    skip_reason = summary.get("skip_reason")
    if skip_reason == "partition_data_exists":
        return True
    mysql_save = summary.get("mysql_save") or {}
    if mysql_save.get("skipped") and mysql_save.get("skip_reason") == "partition_data_exists":
        return True
    if mysql_save and not mysql_save.get("skipped"):
        return True
    return False


def _festival_demand_needs_retry(summary: dict[str, Any]) -> bool:
    """节日需求：异常外的可恢复失败（如源数据未就绪）需要重试。"""
    if _is_write_success(summary):
        return False
    skip_reason = summary.get("skip_reason")
    # 当天无活跃节日 / 测试跳过 / 流程跑完但无需落库 → 视为当天完成，不再重试
    if skip_reason in {"no_active_festivals", "skip_odps"}:
        return False
    if skip_reason == "no_demand_names":
        return True
    # 匹配完成但无生成需求（无需写入）
    if not skip_reason and not (summary.get("mysql_save") or {}):
        return False
    return True


def _gap_script_demand_needs_retry(summary: dict[str, Any]) -> bool:
    """脚本主驱：源数据未就绪需重试；无匹配项则当天完成。"""
    if _is_write_success(summary):
        return False
    skip_reason = summary.get("skip_reason")
    if skip_reason == "no_matched_demands":
        return False
    if skip_reason == "no_demand_names":
        return True
    if not skip_reason and not (summary.get("mysql_save") or {}):
        return False
    return True


def _run_with_same_day_retry(
    *,
    job_name: str,
    query_date: date,
    run_once_fn: Callable[[], dict[str, Any]],
    needs_retry: Callable[[dict[str, Any]], bool],
    print_summary: Callable[[dict[str, Any]], None],
    interval_seconds: int = SAME_DAY_RETRY_INTERVAL_SECONDS,
) -> None:
    """当天失败后每隔 interval_seconds 重试，直到写入成功或跨日停止。"""
    attempt = 0
    while True:
        attempt += 1
        try:
            summary = run_once_fn()
            print_summary(summary)
            print(
                json.dumps(
                    {
                        "job": job_name,
                        "attempt": attempt,
                        "query_date": query_date.isoformat(),
                        "summary": summary,
                    },
                    ensure_ascii=False,
                    indent=2,
                )
            )
            if not needs_retry(summary):
                if _is_write_success(summary):
                    print(
                        f"{job_name}: write success on attempt={attempt} "
                        f"query_date={query_date.isoformat()}",
                        flush=True,
                    )
                else:
                    print(
                        f"{job_name}: finished without retry "
                        f"attempt={attempt} skip_reason={summary.get('skip_reason')!r} "
                        f"query_date={query_date.isoformat()}",
                        flush=True,
                    )
                return

            retry_reason = summary.get("skip_reason") or "incomplete"
            print(
                f"{job_name}: not ready ({retry_reason}), "
                f"retry in {interval_seconds}s "
                f"(attempt={attempt}, query_date={query_date.isoformat()})",
                flush=True,
            )
        except Exception as exc:
            print(f"{job_name} failed: {exc}", file=sys.stderr)
            print(
                f"{job_name}: retry in {interval_seconds}s "
                f"(attempt={attempt}, query_date={query_date.isoformat()})",
                flush=True,
            )

        if _today_shanghai() != query_date:
            print(
                f"{job_name}: stop retry, day changed "
                f"(query_date={query_date.isoformat()})",
                flush=True,
            )
            return

        time.sleep(max(interval_seconds, 1))

        if _today_shanghai() != query_date:
            print(
                f"{job_name}: stop retry after sleep, day changed "
                f"(query_date={query_date.isoformat()})",
                flush=True,
            )
            return


def run_festival_demand_job(config: FestivalDemandConfig) -> None:
    query_date = _today_shanghai()
    _run_with_same_day_retry(
        job_name="festival_demand",
        query_date=query_date,
        run_once_fn=lambda: run_festival_demand_daily_job(config, query_date=query_date),
        needs_retry=_festival_demand_needs_retry,
        print_summary=print_festival_demand_summary,
    )


def run_gap_script_demand_job(config: GapScriptDemandConfig) -> None:
    query_date = _today_shanghai()
    _run_with_same_day_retry(
        job_name="gap_script_demand",
        query_date=query_date,
        run_once_fn=lambda: run_gap_script_demand_daily_job(config, query_date=query_date),
        needs_retry=_gap_script_demand_needs_retry,
        print_summary=print_gap_script_demand_summary,
    )


def register_hot_content_job(scheduler: Any, config: FlowConfig) -> None:
    scheduler.add_job(
        run_hot_content_job,
        trigger="cron",
        hour=config.hot_flow_cron_hours,
        minute=config.hot_flow_cron_minute,
        timezone=SHANGHAI_TZ,
        args=[config],
        id="hot_content_flow",
        name="热点内容抓取搜索解构流程",
        replace_existing=True,
        coalesce=True,
        max_instances=1,
    )


def register_decode_result_job(scheduler: Any, config: FlowConfig) -> None:
    interval_seconds = max(config.decode_result_interval_seconds, 60)
    scheduler.add_job(
        run_decode_result_job,
        trigger="interval",
        seconds=interval_seconds,
        args=[config],
        id="decode_result_flow",
        name="解构结果拉取、后处理与 ODPS 需求表写入",
        replace_existing=True,
        coalesce=True,
        max_instances=1,
        next_run_time=datetime.now(SHANGHAI_TZ) + timedelta(seconds=interval_seconds),
    )


def register_wxindex_words_refresh_job(scheduler: Any, config: FlowConfig) -> None:
    scheduler.add_job(
        run_wxindex_words_refresh_job,
        trigger="cron",
        hour=config.wxindex_words_cron_hours,
        minute=config.wxindex_words_cron_minute,
        timezone=SHANGHAI_TZ,
        args=[config],
        id="wxindex_words_refresh",
        name="微信指数词汇总表补全缺失日期并清理低均值词",
        replace_existing=True,
        coalesce=True,
        max_instances=1,
    )


def register_wxindex_heat_pattern_job(scheduler: Any, config: FlowConfig) -> None:
    scheduler.add_job(
        run_wxindex_heat_pattern_job,
        trigger="cron",
        hour=config.wxindex_heat_pattern_cron_hours,
        minute=config.wxindex_heat_pattern_cron_minute,
        timezone=SHANGHAI_TZ,
        args=[config],
        id="wxindex_heat_pattern",
        name="微信指数热度模式分析（持续高热/上涨/暴涨）",
        replace_existing=True,
        coalesce=True,
        max_instances=1,
    )


def _has_today_output_data(*, strategy: str, repository_cls: type[Any]) -> bool:
    """判断当天输出 strategy 分区是否已有成功写入数据。"""
    partition_dt = _today_shanghai().strftime("%Y%m%d")
    flow_config = load_flow_config()
    repository = repository_cls(flow_config.mysql)
    try:
        return bool(
            repository.has_partition_data(
                strategy=strategy,
                partition_dt=partition_dt,
            )
        )
    finally:
        repository.close()


def _should_run_immediately_on_startup(
    *,
    strategy: str,
    repository_cls: type[Any],
    job_name: str,
) -> bool:
    """启动时判断：当天尚未成功写入则立刻执行，不只依赖 cron。"""
    try:
        already_done = _has_today_output_data(
            strategy=strategy,
            repository_cls=repository_cls,
        )
    except Exception as exc:
        print(
            f"{job_name}: startup check failed ({exc}), "
            "will run immediately",
            flush=True,
        )
        return True

    if already_done:
        print(
            f"{job_name}: today already succeeded "
            f"(strategy={strategy}, date={_today_shanghai().isoformat()}), "
            "skip startup run",
            flush=True,
        )
        return False

    print(
        f"{job_name}: today not succeeded yet "
        f"(strategy={strategy}, date={_today_shanghai().isoformat()}), "
        "schedule startup run now",
        flush=True,
    )
    return True


def register_festival_demand_job(scheduler: Any, config: FestivalDemandConfig) -> None:
    job_kwargs: dict[str, Any] = {
        "trigger": "cron",
        "hour": config.cron_hours,
        "minute": config.cron_minute,
        "timezone": SHANGHAI_TZ,
        "args": [config],
        "id": "festival_demand",
        "name": "节日需求：活跃节日检测 + ODPS 需求词匹配",
        "replace_existing": True,
        "coalesce": True,
        "max_instances": 1,
    }
    if _should_run_immediately_on_startup(
        strategy=config.output_strategy,
        repository_cls=FestivalDemandRepository,
        job_name="festival_demand",
    ):
        job_kwargs["next_run_time"] = datetime.now(SHANGHAI_TZ)
    scheduler.add_job(run_festival_demand_job, **job_kwargs)


def register_gap_script_demand_job(scheduler: Any, config: GapScriptDemandConfig) -> None:
    job_kwargs: dict[str, Any] = {
        "trigger": "cron",
        "hour": config.cron_hours,
        "minute": config.cron_minute,
        "timezone": SHANGHAI_TZ,
        "args": [config],
        "id": "gap_script_demand",
        "name": "当下供需gap：脚本主驱画面改造筛选",
        "replace_existing": True,
        "coalesce": True,
        "max_instances": 1,
    }
    if _should_run_immediately_on_startup(
        strategy=config.output_strategy,
        repository_cls=GapScriptDemandRepository,
        job_name="gap_script_demand",
    ):
        job_kwargs["next_run_time"] = datetime.now(SHANGHAI_TZ)
    scheduler.add_job(run_gap_script_demand_job, **job_kwargs)


def start_scheduler() -> None:
    BlockingScheduler = _import_blocking_scheduler()
    scheduler = BlockingScheduler(timezone=SHANGHAI_TZ)
    config = load_flow_config()
    festival_config = load_festival_demand_config()
    gap_script_config = load_gap_script_demand_config()
    register_hot_content_job(scheduler, config)
    register_decode_result_job(scheduler, config)
    register_wxindex_words_refresh_job(scheduler, config)
    register_wxindex_heat_pattern_job(scheduler, config)
    register_festival_demand_job(scheduler, festival_config)
    register_gap_script_demand_job(scheduler, gap_script_config)
    print(
        "scheduler started, timezone=Asia/Shanghai, "
        "jobs=['hot_content_flow', 'decode_result_flow', 'wxindex_words_refresh', "
        "'wxindex_heat_pattern', 'festival_demand', 'gap_script_demand'], "
        f"hot_cron={config.hot_flow_cron_hours}:{config.hot_flow_cron_minute:02d}, "
        f"decode_result_interval={config.decode_result_interval_seconds}s, "
        f"wxindex_words_cron={config.wxindex_words_cron_hours}:{config.wxindex_words_cron_minute:02d}, "
        f"wxindex_heat_pattern_cron="
        f"{config.wxindex_heat_pattern_cron_hours}:{config.wxindex_heat_pattern_cron_minute:02d}, "
        f"festival_demand_cron={festival_config.cron_hours}:{festival_config.cron_minute:02d}, "
        f"gap_script_demand_cron={gap_script_config.cron_hours}:{gap_script_config.cron_minute:02d}, "
        f"same_day_retry_interval={SAME_DAY_RETRY_INTERVAL_SECONDS}s, "
        "startup_check=festival_demand+gap_script_demand"
    )
    scheduler.start()


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="统一定时任务调度入口")
    parser.add_argument("--once", action="store_true", help="执行一次，不启动调度器")
    parser.add_argument(
        "--job",
        choices=(
            "all",
            "hot-content",
            "decode-result",
            "postprocess",
            "wxindex-refresh",
            "wxindex-heat-pattern",
            "festival-demand",
            "gap-script-demand",
        ),
        default="all",
        help="--once 时选择执行哪个任务",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    if args.once:
        config = load_flow_config()
        if args.job in {"all", "hot-content"}:
            summary = run_once(config)
            print(
                json.dumps(
                    {"job": "hot_content_flow", "summary": summary},
                    ensure_ascii=False,
                    indent=2,
                )
            )
        if args.job in {"all", "decode-result"}:
            summary: dict[str, Any] = {}
            try:
                summary["decode_result"] = run_decode_result_once(config)
            except Exception as exc:
                summary["decode_result_error"] = str(exc)
            try:
                summary["postprocess"] = run_postprocess_once(config)
            except Exception as exc:
                summary["postprocess_error"] = str(exc)
            print(
                json.dumps(
                    {"job": "decode_result_flow", "summary": summary},
                    ensure_ascii=False,
                    indent=2,
                )
            )
        if args.job in {"postprocess"}:
            summary = run_postprocess_once(config)
            print(
                json.dumps(
                    {"job": "postprocess", "summary": summary},
                    ensure_ascii=False,
                    indent=2,
                )
            )
        if args.job in {"wxindex-refresh"}:
            run_wxindex_words_refresh_job(config)
        if args.job in {"wxindex-heat-pattern"}:
            run_wxindex_heat_pattern_job(config)
        if args.job in {"festival-demand"}:
            festival_config = load_festival_demand_config()
            run_festival_demand_job(festival_config)
        if args.job in {"gap-script-demand"}:
            gap_script_config = load_gap_script_demand_config()
            run_gap_script_demand_job(gap_script_config)
        return
    start_scheduler()


if __name__ == "__main__":
    main()