external_demand/app/festival_demand/fixed_dates.py

"""确定性节日/节气/纪念日的代码日期计算与活跃检测兜底。"""

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())