"""确定性节日/节气/纪念日的代码日期计算与活跃检测兜底。""" from __future__ import annotations from datetime import date, timedelta from typing import Any from app.festival_demand.validation import filter_festivals_by_query_date # 21 世纪节气常数 C(索引 0=小寒 … 23=冬至) _CENTURY_21_C = ( 5.4055, 20.12, 3.87, 18.73, 5.63, 20.646, 4.81, 20.1, 5.52, 21.04, 5.678, 21.37, 7.108, 22.83, 7.5, 23.13, 7.646, 23.042, 8.318, 23.438, 7.438, 22.36, 7.18, 21.94, ) _SOLAR_TERM_INDEX: dict[str, int] = { "小寒": 0, "大寒": 1, "立春": 2, "雨水": 3, "惊蛰": 4, "春分": 5, "清明": 6, "谷雨": 7, "立夏": 8, "小满": 9, "芒种": 10, "夏至": 11, "小暑": 12, "大暑": 13, "立秋": 14, "处暑": 15, "白露": 16, "秋分": 17, "寒露": 18, "霜降": 19, "立冬": 20, "小雪": 21, "大雪": 22, "冬至": 23, } # 固定公历月日(非法定单日事件:festival_start = festival_end = 当天) FIXED_MONTH_DAY: dict[str, tuple[int, int]] = { "妇女节": (3, 8), "儿童节": (6, 1), "建党节": (7, 1), "建军节": (8, 1), "教师节": (9, 10), "周总理逝世": (1, 8), "周总理诞辰": (3, 5), "七七事变纪念日": (7, 7), "日本投降": (8, 15), "毛主席逝世": (9, 9), "918纪念日": (9, 18), "毛主席诞辰": (12, 26), "公祭日": (12, 13), "台湾光复纪念日": (10, 25), "315": (3, 15), } # 固定公历锚点日(法定节日兜底:仅锚点日当天为节日期,放假区间仍由 LLM 补充) FIXED_STATUTORY_ANCHOR: dict[str, tuple[int, int]] = { "元旦": (1, 1), "劳动节": (5, 1), "国庆节": (10, 1), } # 第 N 个星期 X(month 1-12, weekday 0=周一 … 6=周日) _NTH_WEEKDAY_RULES: dict[str, tuple[int, int, int]] = { "母亲节": (5, 6, 2), # 5 月第 2 个周日 "父亲节": (6, 6, 3), # 6 月第 3 个周日 } DETERMINISTIC_EVENT_NAMES: frozenset[str] = frozenset( { *_SOLAR_TERM_INDEX.keys(), *FIXED_MONTH_DAY.keys(), *FIXED_STATUTORY_ANCHOR.keys(), *_NTH_WEEKDAY_RULES.keys(), } ) def is_deterministic_event(name: str) -> bool: return name in DETERMINISTIC_EVENT_NAMES def solar_term_date(year: int, term_name: str) -> date: """计算指定年份的二十四节气公历日期。""" if term_name not in _SOLAR_TERM_INDEX: raise ValueError(f"unknown solar term: {term_name}") term_index = _SOLAR_TERM_INDEX[term_name] century_c = _CENTURY_21_C[term_index] y = year % 100 leap_years = y // 4 day = int(y * 0.2422 + century_c) - leap_years month = term_index // 2 + 1 return date(year, month, day) def nth_weekday_of_month(year: int, month: int, weekday: int, nth: int) -> date: """返回某月第 nth 个指定星期几的日期(weekday: 0=周一 … 6=周日)。""" if nth <= 0: raise ValueError("nth must be positive") first = date(year, month, 1) offset = (weekday - first.weekday()) % 7 candidate = first + timedelta(days=offset) candidate += timedelta(weeks=nth - 1) if candidate.month != month: raise ValueError("nth weekday does not exist in month") return candidate def resolve_event_date(year: int, name: str) -> date | None: """解析确定性事件的当年公历日期。""" if name in _SOLAR_TERM_INDEX: return solar_term_date(year, name) if name in FIXED_MONTH_DAY: month, day = FIXED_MONTH_DAY[name] return date(year, month, day) if name in FIXED_STATUTORY_ANCHOR: month, day = FIXED_STATUTORY_ANCHOR[name] return date(year, month, day) if name in _NTH_WEEKDAY_RULES: month, weekday, nth = _NTH_WEEKDAY_RULES[name] return nth_weekday_of_month(year, month, weekday, nth) return None def detect_deterministic_active_festivals( query_date: date, *, allowed_kinds: dict[str, str], prewarm_by_name: dict[str, int], event_type_by_name: dict[str, str], ) -> list[dict[str, Any]]: """用代码计算确定性事件在 query_date 是否处于预热/节日期。""" candidates: list[dict[str, Any]] = [] year = query_date.year for name in DETERMINISTIC_EVENT_NAMES: if name not in prewarm_by_name: continue try: event_date_value = resolve_event_date(year, name) except (ValueError, OverflowError): continue if event_date_value is None: continue kind = allowed_kinds.get(name, "non_statutory") event_iso = event_date_value.isoformat() candidates.append( { "name": name, "kind": kind, "event_date": event_iso, "festival_start": event_iso, "festival_end": event_iso, } ) filtered = filter_festivals_by_query_date( candidates, query_date, prewarm_by_name, ) enriched: list[dict[str, Any]] = [] for item in filtered: name = str(item.get("name") or "").strip() enriched.append( { **item, "event_type": event_type_by_name.get(name, ""), "detection_source": "code", "reason": "代码兜底:确定性日期", } ) return enriched def merge_festival_detections( *detection_lists: list[dict[str, Any]], ) -> list[dict[str, Any]]: """按名称合并多路检测结果,后出现的结果覆盖先前的同名项。""" merged: dict[str, dict[str, Any]] = {} for detection_list in detection_lists: for item in detection_list: name = str(item.get("name") or "").strip() if not name: continue merged[name] = item return list(merged.values())