external_demand/app/festival_demand/demand_generation.py

"""根据节日匹配结果生成需求。"""

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,
    }