"""根据节日匹配结果生成需求。""" from __future__ import annotations from datetime import date from typing import Any, Literal GenerationType = Literal["year_festival", "festival", "year_festival_feature"] def _append_demand( items: list[dict[str, Any]], flat_names: list[str], seen_names: set[str], *, demand_name: str, generation_type: GenerationType, festival_name: str, feature_name: str = "", event_type: str = "", ) -> None: name = demand_name.strip() if not name: return items.append( { "demand_name": name, "generation_type": generation_type, "festival_name": festival_name, "feature_name": feature_name, "event_type": event_type, } ) if name not in seen_names: seen_names.add(name) flat_names.append(name) def _matched_festivals_in_order(matches: list[dict[str, Any]]) -> list[str]: festivals: list[str] = [] seen: set[str] = set() for item in matches: festival_name = str(item.get("festival_name") or "").strip() if not festival_name or festival_name in seen: continue seen.add(festival_name) festivals.append(festival_name) return festivals def _event_type_by_festival(matches: list[dict[str, Any]]) -> dict[str, str]: mapping: dict[str, str] = {} for item in matches: festival_name = str(item.get("festival_name") or "").strip() if not festival_name: continue mapping.setdefault(festival_name, str(item.get("event_type") or "").strip()) return mapping def generate_demands_from_matches( matches: list[dict[str, Any]], *, query_date: date, ) -> dict[str, Any]: """根据匹配结果生成三种类型的需求。 1. year_festival: 当前年份 + 节日事件,如「2026 建党节」 2. festival: 节日事件本身,如「建党节」 3. year_festival_feature: 当前年份 + 节日事件 + 特征点,如「2026 劳动节 五一劳动节」 """ if not matches: return { "year": query_date.year, "generated_demands": [], "generated_demand_names": [], "generated_demand_count": 0, "demands_by_type": { "year_festival": [], "festival": [], "year_festival_feature": [], }, } year = str(query_date.year) event_type_by_festival = _event_type_by_festival(matches) items: list[dict[str, Any]] = [] flat_names: list[str] = [] seen_names: set[str] = set() for festival_name in _matched_festivals_in_order(matches): event_type = event_type_by_festival.get(festival_name, "") _append_demand( items, flat_names, seen_names, demand_name=f"{year} {festival_name}", generation_type="year_festival", festival_name=festival_name, event_type=event_type, ) _append_demand( items, flat_names, seen_names, demand_name=festival_name, generation_type="festival", festival_name=festival_name, event_type=event_type, ) for item in matches: festival_name = str(item.get("festival_name") or "").strip() feature_name = str(item.get("demand_name") or "").strip() if not festival_name or not feature_name: continue _append_demand( items, flat_names, seen_names, demand_name=f"{year} {festival_name} {feature_name}", generation_type="year_festival_feature", festival_name=festival_name, feature_name=feature_name, event_type=str(item.get("event_type") or "").strip(), ) demands_by_type: dict[str, list[str]] = { "year_festival": [], "festival": [], "year_festival_feature": [], } for item in items: generation_type = str(item.get("generation_type") or "") demand_name = str(item.get("demand_name") or "") if generation_type in demands_by_type and demand_name: demands_by_type[generation_type].append(demand_name) return { "year": query_date.year, "generated_demands": items, "generated_demand_names": flat_names, "generated_demand_count": len(flat_names), "demands_by_type": demands_by_type, }