external_demand/app/gap_script_demand/service.py

"""当下供需 gap 脚本主驱每日任务入口。"""

from __future__ import annotations

from datetime import date, datetime
from typing import Any

from app.gap_script_demand.config import load_gap_script_demand_config
from app.gap_script_demand.judging import judge_remake_suitable_demands
from app.gap_script_demand.odps_demand_pool import fetch_demand_names
from app.gap_script_demand.odps_writer import build_odps_row, sync_gap_script_demands_to_odps
from app.gap_script_demand.repository import GapScriptDemandRepository
from app.gap_script_demand.types import GapScriptDemandConfig
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_gap_script_demand_summary(summary: dict[str, Any]) -> None:
    """将脚本主驱流程结果打印到控制台。"""
    print("=" * 60)
    print("当下供需gap-脚本主驱 任务结果")
    print("=" * 60)
    print(f"查询日期: {summary.get('query_date')}")
    print(f"ODPS 分区: {summary.get('partition_dt')}")
    print(f"源 strategy: {summary.get('demand_pool_strategy')}")
    print(f"输出 strategy: {summary.get('output_strategy')}")
    print(f"LLM 模型: {summary.get('llm_model')}")
    print(f"批次大小: {summary.get('judge_batch_size')}")

    if summary.get("skip_reason") in {
        "no_demand_names",
        "partition_data_exists",
        "no_matched_demands",
    }:
        print(f"跳过原因: {summary.get('skip_reason')}")
        if summary.get("skip_reason") == "no_matched_demands":
            print(f"ODPS 需求总数: {summary.get('demand_name_total', 0)}")
            print("匹配可改造数: 0")
        print("=" * 60)
        return

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

    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_matched_demands(
    *,
    demand_names: list[str],
    strategy: str,
    partition_dt: str,
    target_table: str,
    sync_odps: bool = True,
) -> tuple[dict[str, Any], dict[str, Any]]:
    flow_config = load_flow_config()
    repository = GapScriptDemandRepository(flow_config.mysql)
    try:
        mysql_save = repository.persist_matched_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, {}

        if not sync_odps:
            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_gap_script_demands_to_odps(
            rows=odps_rows,
            partition_dt=partition_dt,
            target_table=target_table,
        )
        return mysql_save, odps_sync
    finally:
        repository.close()


def run_gap_script_demand_daily_job(
    config: GapScriptDemandConfig,
    *,
    query_date: date | None = None,
    sync_odps: bool = True,
) -> dict[str, Any]:
    """执行脚本主驱流程：拉取当天供需 gap → LLM 筛选可改造项 → 写 MySQL/ODPS。"""
    target_date = query_date or _today_shanghai()
    partition_dt = target_date.strftime("%Y%m%d")

    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,
        "llm_model": config.llm_model,
        "judge_batch_size": config.judge_batch_size,
        "demand_name_total": 0,
        "matched_count": 0,
        "matched_demand_names": [],
        "mysql_save": {},
        "odps_sync": {},
    }

    flow_config = load_flow_config()
    repository = GapScriptDemandRepository(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": "gap_script_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"gap script demand: start judging "
        f"demand_names={len(demand_names)} batch_size={config.judge_batch_size} "
        f"model={config.llm_model}",
        flush=True,
    )
    matched_demand_names = judge_remake_suitable_demands(
        demand_names=demand_names,
        config=config,
    )
    summary["matched_count"] = len(matched_demand_names)
    summary["matched_demand_names"] = matched_demand_names

    if not matched_demand_names:
        summary["skip_reason"] = "no_matched_demands"
        return summary

    print("gap script demand: persisting matched demands", flush=True)
    mysql_save, odps_sync = _persist_matched_demands(
        demand_names=matched_demand_names,
        strategy=config.output_strategy,
        partition_dt=partition_dt,
        target_table=config.demand_pool_source_table,
        sync_odps=sync_odps,
    )
    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_gap_script_demand_for_date(
    query_date: date,
    *,
    sync_odps: bool = True,
) -> dict[str, Any]:
    """测试指定日期的脚本主驱流程。"""
    config = load_gap_script_demand_config()
    return run_gap_script_demand_daily_job(
        config,
        query_date=query_date,
        sync_odps=sync_odps,
    )