"""当下供需 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, )