external_demand/app/festival_demand/service.py

"""节日需求每日任务入口。"""

from __future__ import annotations

from datetime import date, datetime
from typing import Any

from app.festival_demand.config import load_festival_demand_config
from app.festival_demand.demand_generation import generate_demands_from_matches
from app.festival_demand.detection import detect_active_festivals
from app.festival_demand.matching import (
    collect_matched_demand_names,
    match_demand_names_to_festivals,
)
from app.festival_demand.odps_demand_pool import fetch_demand_names
from app.festival_demand.odps_writer import build_odps_row, sync_festival_demands_to_odps
from app.festival_demand.repository import FestivalDemandRepository
from app.festival_demand.types import FestivalDemandConfig
from app.hot_content.config import load_flow_config
from app.hot_content.timezone import SHANGHAI_TZ


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


def print_festival_demand_summary(summary: dict[str, Any]) -> None:
    """将节日需求流程结果打印到控制台。"""
    print("=" * 60)
    print("节日需求任务结果")
    print("=" * 60)
    print(f"查询日期: {summary.get('query_date')}")
    print(f"ODPS 分区: {summary.get('partition_dt')}")
    print(f"需求池 strategy: {summary.get('demand_pool_strategy')}")

    if summary.get("skip_reason") in {
        "no_active_festivals",
        "skip_odps",
        "no_demand_names",
        "partition_data_exists",
    }:
        print(f"跳过原因: {summary.get('skip_reason')}")
        print("=" * 60)
        return

    festival_names = summary.get("festival_names") or []
    print(f"\n活跃节日 ({len(festival_names)}):")
    for name in festival_names:
        print(f"  - {name}")

    print(f"\nODPS 需求词总数: {summary.get('demand_name_total', 0)}")
    print(f"匹配特征词数: {summary.get('matched_count', 0)}")

    matched_names = summary.get("matched_demand_names") or []
    if matched_names:
        print("\n匹配到的特征词:")
        for name in matched_names:
            print(f"  * {name}")

    demands_by_type = summary.get("demands_by_type") or {}
    type_labels = {
        "year_festival": "方式1: 年份 + 节日",
        "festival": "方式2: 节日",
        "year_festival_feature": "方式3: 年份 + 节日 + 特征点",
    }
    generated_names = summary.get("generated_demand_names") or []
    print(f"\n生成需求总数: {summary.get('generated_demand_count', len(generated_names))}")
    for generation_type, label in type_labels.items():
        names = demands_by_type.get(generation_type) or []
        if not names:
            continue
        print(f"\n{label} ({len(names)}):")
        for name in names:
            print(f"  + {name}")

    if not generated_names:
        print("\n（暂无生成需求）")

    mysql_save = summary.get("mysql_save") or {}
    if mysql_save.get("skipped"):
        print(
            f"\nMySQL/ODPS 跳过: reason={mysql_save.get('skip_reason')} "
            f"partition_dt={mysql_save.get('partition_dt')}"
        )
    elif mysql_save:
        print(
            f"\nMySQL 写入: table={mysql_save.get('table')} "
            f"saved={mysql_save.get('saved_count', 0)} "
            f"strategy={mysql_save.get('strategy')}"
        )
        odps_sync = summary.get("odps_sync") or {}
        if odps_sync:
            print(
                f"ODPS 写入: table={odps_sync.get('target_table')} "
                f"partition_dt={odps_sync.get('partition_dt')} "
                f"written={odps_sync.get('written_count', 0)}"
            )

    print("=" * 60)


def _persist_generated_demands(
    *,
    demand_names: list[str],
    strategy: str,
    partition_dt: str,
    target_table: str,
) -> tuple[dict[str, Any], dict[str, Any]]:
    flow_config = load_flow_config()
    repository = FestivalDemandRepository(flow_config.mysql)
    try:
        mysql_save = repository.persist_generated_demands(
            demand_names=demand_names,
            strategy=strategy,
            partition_dt=partition_dt,
        )
        if mysql_save.get("skipped") or not mysql_save.get("rows"):
            return mysql_save, {}

        odps_rows = [
            build_odps_row(
                strategy=row["strategy"],
                demand_id=row["demand_id"],
                demand_name=row["demand_name"],
            )
            for row in mysql_save["rows"]
        ]
        odps_sync = sync_festival_demands_to_odps(
            rows=odps_rows,
            partition_dt=partition_dt,
            target_table=target_table,
        )
        return mysql_save, odps_sync
    finally:
        repository.close()


def run_festival_demand_daily_job(
    config: FestivalDemandConfig,
    *,
    query_date: date | None = None,
    skip_odps: bool = False,
) -> dict[str, Any]:
    """执行节日需求流程：检测活跃节日 → 拉取 ODPS 需求词 → LLM 批量匹配。"""
    target_date = query_date or _today_shanghai()
    partition_dt = target_date.strftime("%Y%m%d")

    detection = detect_active_festivals(target_date, config)
    festivals = detection.get("festivals") or []

    summary: dict[str, Any] = {
        "query_date": target_date.isoformat(),
        "partition_dt": partition_dt,
        "demand_pool_strategy": config.demand_pool_strategy,
        "output_strategy": config.output_strategy,
        "demand_pool_source_table": config.demand_pool_source_table,
        "festival_count": len(festivals),
        "festival_names": detection.get("festival_names") or [],
        "festivals": festivals,
        "demand_name_total": 0,
        "match_batch_size": config.match_batch_size,
        "matched_count": 0,
        "matched_demand_names": [],
        "matches": [],
        "generated_demand_count": 0,
        "generated_demand_names": [],
        "generated_demands": [],
        "demands_by_type": {
            "year_festival": [],
            "festival": [],
            "year_festival_feature": [],
        },
        "mysql_save": {},
        "odps_sync": {},
    }

    if not festivals:
        summary["skip_reason"] = "no_active_festivals"
        return summary

    if skip_odps:
        summary["skip_reason"] = "skip_odps"
        return summary

    flow_config = load_flow_config()
    repository = FestivalDemandRepository(flow_config.mysql)
    try:
        if repository.has_partition_data(
            strategy=config.output_strategy,
            partition_dt=partition_dt,
        ):
            summary["skip_reason"] = "partition_data_exists"
            summary["mysql_save"] = {
                "table": "festival_demand_pool_di",
                "strategy": config.output_strategy,
                "partition_dt": partition_dt,
                "skipped": True,
                "skip_reason": "partition_data_exists",
            }
            return summary
    finally:
        repository.close()

    demand_names = fetch_demand_names(
        partition_dt=partition_dt,
        strategy=config.demand_pool_strategy,
        source_table=config.demand_pool_source_table,
    )
    summary["demand_name_total"] = len(demand_names)

    if not demand_names:
        summary["skip_reason"] = "no_demand_names"
        return summary

    print(
        f"festival demand: start matching "
        f"festivals={len(festivals)} demand_names={len(demand_names)} "
        f"batch_size={config.match_batch_size}",
        flush=True,
    )
    matches = match_demand_names_to_festivals(
        demand_names=demand_names,
        festivals=festivals,
        config=config,
    )
    matched_demand_names = collect_matched_demand_names(matches)
    generated = generate_demands_from_matches(matches, query_date=target_date)

    summary["matched_count"] = len(matched_demand_names)
    summary["matched_demand_names"] = matched_demand_names
    summary["matches"] = matches
    summary["generated_demand_count"] = generated["generated_demand_count"]
    summary["generated_demand_names"] = generated["generated_demand_names"]
    summary["generated_demands"] = generated["generated_demands"]
    summary["demands_by_type"] = generated["demands_by_type"]

    if generated["generated_demand_names"]:
        print("festival demand: persisting generated demands", flush=True)
        mysql_save, odps_sync = _persist_generated_demands(
            demand_names=generated["generated_demand_names"],
            strategy=config.output_strategy,
            partition_dt=partition_dt,
            target_table=config.demand_pool_source_table,
        )
        summary["mysql_save"] = mysql_save
        summary["odps_sync"] = odps_sync
        if mysql_save.get("skipped"):
            summary["skip_reason"] = mysql_save.get("skip_reason")
    return summary


def test_festivals_for_date(
    query_date: date,
    *,
    skip_odps: bool = False,
) -> dict[str, Any]:
    """测试指定日期的节日需求流程。"""
    config = load_festival_demand_config()
    return run_festival_demand_daily_job(
        config,
        query_date=query_date,
        skip_odps=skip_odps,
    )