"""节日结果过滤:仅校验查询日期是否在 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