external_demand/app/festival_demand/validation.py

"""节日结果过滤：仅校验查询日期是否在 LLM 返回的预热/节日期区间内。"""

from __future__ import annotations

from datetime import date, timedelta
from typing import Any


def _parse_date(value: Any) -> date | None:
    text = str(value or "").strip()
    if not text:
        return None
    try:
        return date.fromisoformat(text)
    except ValueError:
        return None


def is_festival_active_on_date(
    query_date: date,
    festival_start: date,
    festival_end: date,
    prewarm_days: int,
) -> tuple[str, int] | None:
    """判断查询日是否命中；命中返回 (phase, days_to_festival)，否则 None。"""
    if festival_start <= query_date <= festival_end:
        return ("festival", 0)
    if prewarm_days > 0:
        prewarm_start = festival_start - timedelta(days=prewarm_days)
        if prewarm_start <= query_date < festival_start:
            return ("prewarm", (festival_start - query_date).days)
    return None


def filter_festivals_by_query_date(
    festivals: list[dict[str, Any]],
    query_date: date,
    prewarm_by_name: dict[str, int],
) -> list[dict[str, Any]]:
    """保留符合「目标日期 + 提前量」的节日，移除其余项。"""
    kept: list[dict[str, Any]] = []
    for item in festivals:
        name = str(item.get("name") or "").strip()
        if not name or name not in prewarm_by_name:
            continue
        festival_start = _parse_date(item.get("festival_start"))
        festival_end = _parse_date(item.get("festival_end"))
        if not festival_start or not festival_end:
            continue
        if festival_end < festival_start:
            festival_start, festival_end = festival_end, festival_start

        prewarm_days = prewarm_by_name[name]
        active = is_festival_active_on_date(
            query_date,
            festival_start,
            festival_end,
            prewarm_days,
        )
        if not active:
            continue

        phase, days_to_festival = active
        kept.append(
            {
                **item,
                "prewarm_days": prewarm_days,
                "festival_start": festival_start.isoformat(),
                "festival_end": festival_end.isoformat(),
                "phase": phase,
                "days_to_festival": days_to_festival,
            }
        )
    return kept