"""节日需求每日任务入口。""" 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, )