增加新的需求来源

app/core/config.py

@@ -179,6 +179,17 @@ class Settings:
     category_filter_body_max_chars: int = 2000
     category_filter_item_sleep_seconds: float = 0.0
 
+    festival_demand_cron_hours: str = "7"
+    festival_demand_cron_minute: int = 0
+    festival_demand_llm_model: str = "anthropic/claude-haiku-4-5"
+    festival_demand_llm_max_attempts: int = 3
+    festival_demand_llm_retry_sleep_seconds: float = 1.0
+    festival_demand_llm_max_tokens: int = 4000
+    festival_demand_llm_temperature: float = 0.0
+    festival_demand_pool_strategy: str = "去年同期阳历"
+    festival_demand_output_strategy: str = "去年同期阳历-节点事件"
+    festival_demand_match_batch_size: int = 200
+
     @classmethod
     def from_env(cls) -> "Settings":
         defaults = cls()
@@ -411,6 +422,46 @@ class Settings:
                 "CATEGORY_FILTER_ITEM_SLEEP_SECONDS",
                 defaults.category_filter_item_sleep_seconds,
             ),
+            festival_demand_cron_hours=_env(
+                "FESTIVAL_DEMAND_CRON_HOURS",
+                defaults.festival_demand_cron_hours,
+            ),
+            festival_demand_cron_minute=_env_int(
+                "FESTIVAL_DEMAND_CRON_MINUTE",
+                defaults.festival_demand_cron_minute,
+            ),
+            festival_demand_llm_model=_env(
+                "FESTIVAL_DEMAND_LLM_MODEL",
+                defaults.festival_demand_llm_model,
+            ),
+            festival_demand_llm_max_attempts=_env_int(
+                "FESTIVAL_DEMAND_LLM_MAX_ATTEMPTS",
+                defaults.festival_demand_llm_max_attempts,
+            ),
+            festival_demand_llm_retry_sleep_seconds=_env_float(
+                "FESTIVAL_DEMAND_LLM_RETRY_SLEEP_SECONDS",
+                defaults.festival_demand_llm_retry_sleep_seconds,
+            ),
+            festival_demand_llm_max_tokens=_env_int(
+                "FESTIVAL_DEMAND_LLM_MAX_TOKENS",
+                defaults.festival_demand_llm_max_tokens,
+            ),
+            festival_demand_llm_temperature=_env_float(
+                "FESTIVAL_DEMAND_LLM_TEMPERATURE",
+                defaults.festival_demand_llm_temperature,
+            ),
+            festival_demand_pool_strategy=_env(
+                "FESTIVAL_DEMAND_POOL_STRATEGY",
+                defaults.festival_demand_pool_strategy,
+            ),
+            festival_demand_output_strategy=_env(
+                "FESTIVAL_DEMAND_OUTPUT_STRATEGY",
+                defaults.festival_demand_output_strategy,
+            ),
+            festival_demand_match_batch_size=_env_int(
+                "FESTIVAL_DEMAND_MATCH_BATCH_SIZE",
+                defaults.festival_demand_match_batch_size,
+            ),
         )
 
 

app/festival_demand/__init__.py

@@ -0,0 +1 @@
+"""节日需求定时任务模块（与热度需求独立）。"""

app/festival_demand/calendar.py

@@ -0,0 +1,103 @@
+"""节日/节点事件日历与预热配置。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Literal
+
+FestivalKind = Literal["statutory", "non_statutory"]
+EventType = Literal["节日/节气", "纪念日", "特定民生热点", "特定天气时段"]
+
+
+@dataclass(frozen=True)
+class FestivalEvent:
+    """节点事件、类型及预热天数。"""
+
+    name: str
+    event_type: EventType
+    prewarm_days: int
+    kind: FestivalKind = "non_statutory"
+
+
+# 法定节假日：节日活跃期按国务院当年放假区间计算
+_STATUTORY = "statutory"
+_NON_STATUTORY = "non_statutory"
+_FESTIVAL_SOLAR = "节日/节气"
+_MEMORIAL = "纪念日"
+_LIVELIHOOD = "特定民生热点"
+_WEATHER = "特定天气时段"
+
+DEFAULT_FESTIVAL_EVENTS: tuple[FestivalEvent, ...] = (
+    FestivalEvent("元旦", _FESTIVAL_SOLAR, 10, _STATUTORY),
+    FestivalEvent("腊八节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("小年", _FESTIVAL_SOLAR, 7, _NON_STATUTORY),
+    FestivalEvent("除夕", _FESTIVAL_SOLAR, 10, _NON_STATUTORY),
+    FestivalEvent("春节", _FESTIVAL_SOLAR, 10, _STATUTORY),
+    FestivalEvent("元宵节", _FESTIVAL_SOLAR, 7, _NON_STATUTORY),
+    FestivalEvent("龙抬头", _FESTIVAL_SOLAR, 7, _NON_STATUTORY),
+    FestivalEvent("妇女节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("上巳节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("劳动节", _FESTIVAL_SOLAR, 7, _STATUTORY),
+    FestivalEvent("母亲节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("端午节", _FESTIVAL_SOLAR, 7, _STATUTORY),
+    FestivalEvent("儿童节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("父亲节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("建党节", _FESTIVAL_SOLAR, 10, _NON_STATUTORY),
+    FestivalEvent("建军节", _FESTIVAL_SOLAR, 10, _NON_STATUTORY),
+    FestivalEvent("七夕节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("中元节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("教师节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("中秋节", _FESTIVAL_SOLAR, 7, _STATUTORY),
+    FestivalEvent("重阳节", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("国庆节", _FESTIVAL_SOLAR, 7, _STATUTORY),
+    FestivalEvent("立春", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("雨水", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("惊蛰", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("春分", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("清明", _FESTIVAL_SOLAR, 15, _STATUTORY),
+    FestivalEvent("谷雨", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("立夏", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("小满", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("芒种", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("夏至", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("小暑", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("大暑", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("立秋", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("处暑", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("白露", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("秋分", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("寒露", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("霜降", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("立冬", _FESTIVAL_SOLAR, 7, _NON_STATUTORY),
+    FestivalEvent("小雪", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("大雪", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("冬至", _FESTIVAL_SOLAR, 7, _NON_STATUTORY),
+    FestivalEvent("小寒", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("大寒", _FESTIVAL_SOLAR, 5, _NON_STATUTORY),
+    FestivalEvent("周总理逝世", _MEMORIAL, 7, _NON_STATUTORY),
+    FestivalEvent("周总理诞辰", _MEMORIAL, 7, _NON_STATUTORY),
+    FestivalEvent("七七事变纪念日", _MEMORIAL, 5, _NON_STATUTORY),
+    FestivalEvent("日本投降", _MEMORIAL, 5, _NON_STATUTORY),
+    FestivalEvent("毛主席逝世", _MEMORIAL, 30, _NON_STATUTORY),
+    FestivalEvent("918纪念日", _MEMORIAL, 5, _NON_STATUTORY),
+    FestivalEvent("毛主席诞辰", _MEMORIAL, 30, _NON_STATUTORY),
+    FestivalEvent("公祭日", _MEMORIAL, 10, _NON_STATUTORY),
+    FestivalEvent("台湾光复纪念日", _MEMORIAL, 5, _NON_STATUTORY),
+    FestivalEvent("全国两会", _LIVELIHOOD, 30, _NON_STATUTORY),
+    FestivalEvent("315", _LIVELIHOOD, 7, _NON_STATUTORY),
+    FestivalEvent("高考", _LIVELIHOOD, 10, _NON_STATUTORY),
+    FestivalEvent("三伏", _WEATHER, 15, _NON_STATUTORY),
+    FestivalEvent("数九", _WEATHER, 5, _NON_STATUTORY),
+)
+
+
+def festival_events_payload() -> list[dict[str, int | str]]:
+    return [
+        {
+            "name": event.name,
+            "event_type": event.event_type,
+            "prewarm_days": event.prewarm_days,
+            "kind": event.kind,
+        }
+        for event in DEFAULT_FESTIVAL_EVENTS
+    ]

app/festival_demand/config.py

@@ -0,0 +1,102 @@
+"""节日需求流程配置。"""
+
+from __future__ import annotations
+
+import os
+
+from app.core.config import settings
+from app.festival_demand.exceptions import FestivalDemandError
+
+
+def _get_env(name: str, default: str = "") -> str:
+    value = os.getenv(name)
+    if value is None or value == "":
+        return default
+    return value
+
+
+def _get_env_int(name: str, default: int) -> int:
+    raw = os.getenv(name)
+    if raw is None or raw == "":
+        return default
+    try:
+        return int(raw)
+    except ValueError as exc:
+        raise FestivalDemandError(f"invalid integer env {name}={raw!r}") from exc
+
+
+def _get_env_float(name: str, default: float) -> float:
+    raw = os.getenv(name)
+    if raw is None or raw == "":
+        return default
+    try:
+        return float(raw)
+    except ValueError as exc:
+        raise FestivalDemandError(f"invalid float env {name}={raw!r}") from exc
+
+
+def _parse_cron_hours(value: str) -> str:
+    hours = [item.strip() for item in value.split(",") if item.strip()]
+    if not hours:
+        raise FestivalDemandError("festival demand cron hours cannot be empty")
+    normalized: list[str] = []
+    for hour in hours:
+        try:
+            hour_num = int(hour)
+        except ValueError as exc:
+            raise FestivalDemandError(f"invalid festival demand cron hour: {hour!r}") from exc
+        if not 0 <= hour_num <= 23:
+            raise FestivalDemandError(f"festival demand cron hour out of range: {hour_num}")
+        normalized.append(str(hour_num))
+    return ",".join(normalized)
+
+
+def load_festival_demand_config() -> "FestivalDemandConfig":
+    from app.festival_demand.types import FestivalDemandConfig
+
+    llm_model = _get_env("FESTIVAL_DEMAND_LLM_MODEL", settings.festival_demand_llm_model)
+    config = FestivalDemandConfig(
+        cron_hours=_parse_cron_hours(
+            _get_env("FESTIVAL_DEMAND_CRON_HOURS", settings.festival_demand_cron_hours)
+        ),
+        cron_minute=_get_env_int(
+            "FESTIVAL_DEMAND_CRON_MINUTE",
+            settings.festival_demand_cron_minute,
+        ),
+        llm_model=llm_model.strip() or settings.open_router_default_model,
+        llm_max_attempts=_get_env_int(
+            "FESTIVAL_DEMAND_LLM_MAX_ATTEMPTS",
+            settings.festival_demand_llm_max_attempts,
+        ),
+        llm_retry_sleep_seconds=_get_env_float(
+            "FESTIVAL_DEMAND_LLM_RETRY_SLEEP_SECONDS",
+            settings.festival_demand_llm_retry_sleep_seconds,
+        ),
+        llm_max_tokens=_get_env_int(
+            "FESTIVAL_DEMAND_LLM_MAX_TOKENS",
+            settings.festival_demand_llm_max_tokens,
+        ),
+        llm_temperature=_get_env_float(
+            "FESTIVAL_DEMAND_LLM_TEMPERATURE",
+            settings.festival_demand_llm_temperature,
+        ),
+        demand_pool_source_table=_get_env(
+            "DEMAND_POOL_SOURCE_TABLE",
+            settings.demand_pool_source_table,
+        ).strip(),
+        demand_pool_strategy=_get_env(
+            "FESTIVAL_DEMAND_POOL_STRATEGY",
+            settings.festival_demand_pool_strategy,
+        ).strip(),
+        output_strategy=_get_env(
+            "FESTIVAL_DEMAND_OUTPUT_STRATEGY",
+            settings.festival_demand_output_strategy,
+        ).strip(),
+        match_batch_size=_get_env_int(
+            "FESTIVAL_DEMAND_MATCH_BATCH_SIZE",
+            settings.festival_demand_match_batch_size,
+        ),
+    )
+    if config.match_batch_size <= 0:
+        raise FestivalDemandError("FESTIVAL_DEMAND_MATCH_BATCH_SIZE must be positive")
+    return config

app/festival_demand/demand_generation.py

@@ -0,0 +1,145 @@
+"""根据节日匹配结果生成需求。"""
+
+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,
+    }

app/festival_demand/demand_id.py

@@ -0,0 +1,16 @@
+"""节日需求 demand_id 生成。"""
+
+from __future__ import annotations
+
+import hashlib
+
+
+def build_festival_demand_id(
+    *,
+    strategy: str,
+    demand_name: str,
+    partition_dt: str,
+) -> str:
+    """md5(concat(strategy, demand_name, partition_dt))"""
+    raw = f"{strategy}{demand_name}{partition_dt}"
+    return hashlib.md5(raw.encode("utf-8")).hexdigest()

app/festival_demand/detection.py

@@ -0,0 +1,212 @@
+"""调用 LLM 判断当天是否处于节日或预热期内。"""
+
+from __future__ import annotations
+
+import json
+import re
+import time
+from datetime import date
+from typing import Any
+
+from app.core.open_router_llm import OpenRouterCallError, create_chat_completion
+from app.festival_demand.calendar import festival_events_payload
+from app.festival_demand.exceptions import FestivalDemandError
+from app.festival_demand.llm_json import extract_json_object
+from app.festival_demand.types import FestivalDemandConfig
+from app.festival_demand.validation import filter_festivals_by_query_date
+
+SYSTEM_PROMPT = """
+你是一个中国节日与节点事件日历专家。
+
+# 任务
+根据给定的 query_date 和 events 列表，判断 query_date 当天处于哪些事件的活跃期内，并返回这些事件。
+
+# 节日类型（events 中 kind 字段）
+1. statutory（法定节假日）：须查准当年国务院放假调休区间，festival_start 为放假首日，festival_end 为放假末日。
+2. non_statutory（非法定节日）：节日活跃期仅当天，festival_start = festival_end = 该事件当天。
+
+# 预热期（两类通用）
+- 预热只在节日开始之前，节日一到预热立即结束。
+- 设节日开始日 S（法定为 festival_start，非法定为当天），预热天数为 events 中的 prewarm_days（记为 N）：
+  - 预热区间 = [S-N, S-1]（含首尾，不含 S）
+  - 节日期 = [festival_start, festival_end]（含首尾）
+- 仅当 query_date 落在预热区间或节日期内，才返回该事件。
+
+# 你的职责
+- 由你推算当年每个事件的 festival_start、festival_end、event_date（农历、节气、固定纪念日、放假安排等均由你判断）。
+- 严格遵守上述活跃期定义筛选，不要把已过节日的项返回进来。
+
+# 输出格式
+严格只输出一个 JSON 对象，禁止在 JSON 前后或之后追加任何说明、注释、markdown 代码块标记。
+{
+  "date": "YYYY-MM-DD",
+  "festivals": [
+    {
+      "name": "与 events 中 name 完全一致",
+      "kind": "statutory 或 non_statutory",
+      "event_date": "YYYY-MM-DD",
+      "festival_start": "YYYY-MM-DD",
+      "festival_end": "YYYY-MM-DD",
+      "phase": "prewarm 或 festival",
+      "days_to_festival": 0,
+      "reason": "简短中文说明"
+    }
+  ]
+}
+
+# 约束
+- festivals 为空数组表示当天无活跃事件。
+- name 只能来自 events，不得编造。
+- phase=prewarm 时 query_date 必须 < festival_start；phase=festival 时 query_date 必须在 [festival_start, festival_end]。
+"""
+
+
+def _parse_date_field(item: dict[str, Any], *keys: str) -> str | None:
+    for key in keys:
+        value = str(item.get(key) or "").strip()
+        if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
+            return value
+    return None
+
+
+def _parse_festival_item(
+    item: Any,
+    allowed_names: set[str],
+    allowed_kinds: dict[str, str],
+) -> dict[str, Any] | None:
+    if not isinstance(item, dict):
+        return None
+    name = str(item.get("name") or "").strip()
+    if not name or name not in allowed_names:
+        return None
+
+    festival_start = _parse_date_field(item, "festival_start")
+    festival_end = _parse_date_field(item, "festival_end")
+    event_date = _parse_date_field(item, "event_date")
+    if not festival_start and event_date:
+        festival_start = event_date
+    if not festival_end and festival_start:
+        festival_end = festival_start
+    if not festival_start or not festival_end:
+        return None
+
+    kind = str(item.get("kind") or allowed_kinds.get(name, "non_statutory")).strip()
+    if kind not in {"statutory", "non_statutory"}:
+        kind = allowed_kinds.get(name, "non_statutory")
+
+    return {
+        "name": name,
+        "kind": kind,
+        "event_date": event_date or festival_start,
+        "festival_start": festival_start,
+        "festival_end": festival_end,
+        "phase": str(item.get("phase") or "").strip().lower(),
+        "days_to_festival": item.get("days_to_festival"),
+        "reason": str(item.get("reason") or "").strip(),
+    }
+
+
+def _enrich_festivals_with_event_type(
+    festivals: list[dict[str, Any]],
+    event_type_by_name: dict[str, str],
+) -> list[dict[str, Any]]:
+    enriched: list[dict[str, Any]] = []
+    for item in festivals:
+        name = str(item.get("name") or "").strip()
+        enriched.append(
+            {
+                **item,
+                "event_type": event_type_by_name.get(name, ""),
+            }
+        )
+    return enriched
+
+
+def _normalize_detection_result(
+    parsed: dict[str, Any],
+    *,
+    query_date: date,
+    allowed_names: set[str],
+    allowed_kinds: dict[str, str],
+    prewarm_by_name: dict[str, int],
+    event_type_by_name: dict[str, str],
+) -> dict[str, Any]:
+    festivals_raw = parsed.get("festivals")
+    if festivals_raw is None:
+        festivals_raw = parsed.get("active_festivals") or []
+    if not isinstance(festivals_raw, list):
+        raise FestivalDemandError("llm output festivals must be a list")
+
+    parsed_items: list[dict[str, Any]] = []
+    seen_names: set[str] = set()
+    for item in festivals_raw:
+        if isinstance(item, str):
+            continue
+        normalized = _parse_festival_item(item, allowed_names, allowed_kinds)
+        if not normalized or normalized["name"] in seen_names:
+            continue
+        parsed_items.append(normalized)
+        seen_names.add(normalized["name"])
+
+    festivals = filter_festivals_by_query_date(
+        parsed_items,
+        query_date,
+        prewarm_by_name,
+    )
+    festivals = _enrich_festivals_with_event_type(festivals, event_type_by_name)
+
+    return {
+        "date": query_date.isoformat(),
+        "festivals": festivals,
+        "festival_names": [item["name"] for item in festivals],
+    }
+
+
+def detect_active_festivals(
+    query_date: date,
+    config: FestivalDemandConfig,
+) -> dict[str, Any]:
+    """调用 LLM 判断查询日期处于活跃期内的节日列表。"""
+    events = festival_events_payload()
+    allowed_names = {str(item["name"]) for item in events}
+    allowed_kinds = {str(item["name"]): str(item["kind"]) for item in events}
+    prewarm_by_name = {str(item["name"]): int(item["prewarm_days"]) for item in events}
+    event_type_by_name = {str(item["name"]): str(item["event_type"]) for item in events}
+    payload = {
+        "query_date": query_date.isoformat(),
+        "year": query_date.year,
+        "events": events,
+    }
+
+    last_error: Exception | None = None
+    for attempt in range(1, config.llm_max_attempts + 1):
+        try:
+            resp = create_chat_completion(
+                [
+                    {"role": "system", "content": SYSTEM_PROMPT.strip()},
+                    {
+                        "role": "user",
+                        "content": json.dumps(payload, ensure_ascii=False),
+                    },
+                ],
+                model=config.llm_model,
+                temperature=config.llm_temperature,
+                max_tokens=config.llm_max_tokens,
+            )
+            parsed = extract_json_object(str(resp.get("content") or ""))
+            return _normalize_detection_result(
+                parsed,
+                query_date=query_date,
+                allowed_names=allowed_names,
+                allowed_kinds=allowed_kinds,
+                prewarm_by_name=prewarm_by_name,
+                event_type_by_name=event_type_by_name,
+            )
+        except (OpenRouterCallError, FestivalDemandError, ValueError) as exc:
+            last_error = exc
+            if attempt < config.llm_max_attempts:
+                time.sleep(config.llm_retry_sleep_seconds)
+
+    raise FestivalDemandError(
+        f"festival detection failed after {config.llm_max_attempts} attempts: {last_error}"
+    ) from last_error

app/festival_demand/exceptions.py

@@ -0,0 +1,5 @@
+"""节日需求流程异常。"""
+
+
+class FestivalDemandError(Exception):
+    """节日需求流程业务异常。"""

app/festival_demand/llm_json.py

@@ -0,0 +1,58 @@
+"""LLM 回复 JSON 解析工具。"""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+from app.festival_demand.exceptions import FestivalDemandError
+
+
+def strip_markdown_fence(text: str) -> str:
+    raw = text.strip()
+    if not raw.startswith("```"):
+        return raw
+    raw = re.sub(r"^```(?:json)?\s*", "", raw, flags=re.IGNORECASE)
+    raw = re.sub(r"\s*```\s*$", "", raw)
+    return raw.strip()
+
+
+def extract_json_object(text: str) -> dict[str, Any]:
+    """从 LLM 回复中提取第一个 JSON 对象（容忍尾部多余文字）。"""
+    raw = strip_markdown_fence(text)
+    decoder = json.JSONDecoder()
+
+    def _try_decode(candidate: str) -> dict[str, Any] | None:
+        stripped = candidate.strip()
+        if not stripped:
+            return None
+        try:
+            parsed = json.loads(stripped)
+            if isinstance(parsed, dict):
+                return parsed
+        except json.JSONDecodeError:
+            pass
+        try:
+            parsed, _end = decoder.raw_decode(stripped)
+            if isinstance(parsed, dict):
+                return parsed
+        except json.JSONDecodeError:
+            pass
+        return None
+
+    direct = _try_decode(raw)
+    if direct is not None:
+        return direct
+
+    start = 0
+    while True:
+        start = raw.find("{", start)
+        if start == -1:
+            break
+        decoded = _try_decode(raw[start:])
+        if decoded is not None:
+            return decoded
+        start += 1
+
+    raise FestivalDemandError("llm output is not json object")

app/festival_demand/matching.py

@@ -0,0 +1,260 @@
+"""需求词与当天活跃节日/节点事件的 LLM 语义匹配。"""
+
+from __future__ import annotations
+
+import json
+import time
+from typing import Any
+
+from app.core.open_router_llm import OpenRouterCallError, create_chat_completion
+from app.festival_demand.exceptions import FestivalDemandError
+from app.festival_demand.llm_json import extract_json_object
+from app.festival_demand.types import FestivalDemandConfig
+
+MATCH_SYSTEM_PROMPT = """
+你是一个专业的视频内容主题匹配判断助手。你的任务是根据给定的"节点事件"及其类型，判断"视频特征词"是否符合该节点事件的核心主题，从而确定视频是否属于此节点事件。
+
+判断规则
+请严格参照以下节点事件类型与主题对应关系：
+- 类型为"节日/节气"：主题以祝福、知识分享为主（如节日祝福、民俗科普、节气养生知识等）。
+- 类型为"纪念日"：主题以家国仇恨、勿忘国耻、怀念英烈、歌颂伟人事迹为主（如历史事件回顾、英雄人物纪念等）。
+- 类型为"特定天气时段"：主题以描述天气、健康养生为主（如天气变化提醒、时令养生方法等）。
+- 类型为"特定民生热点"：主题以民生、知识分享为主（如政策解读、考试备考、消费警示等）。
+
+输入信息
+用户会提供以下内容：
+- 节点事件名称：<事件名称>
+- 节点事件类型：<从"节日/节气、纪念日、特定天气时段、特定民生热点"中选择其一>
+- 视频特征词列表：多个待判断的特征词
+
+判断逻辑
+1. 首先明确该节点事件所属类型对应的允许主题范围。
+2. 逐条分析每个视频特征词所表达的核心内容。若特征词明显指向上述主题范围（不需要完全一致，语义相近、属正常延伸即可），则判定匹配；若特征词与主题无关（如纯娱乐、无关的日常生活、其他不相干领域），则判定不匹配。
+3. 如遇模棱两可的情况，倾向于严格判断，即只有当特征词明显属于主题方向时才匹配，否则不匹配。
+
+输出要求
+严格只输出一个 JSON 对象，禁止输出 JSON 之外的任何内容。固定格式如下：
+{
+  "matched_demand_names": ["匹配成功的特征词1", "匹配成功的特征词2"]
+}
+
+约束：
+- 只返回判定为匹配的特征词，不匹配的不要放入数组。
+- matched_demand_names 中的每一项必须与用户提供的视频特征词原文完全一致，不得改写、不得编造。
+- 若没有任何特征词匹配，返回 {"matched_demand_names": []}。
+
+示例1：
+节点事件名称：春节
+节点事件类型：节日/节气
+视频特征词列表：拜年祝福、年夜饭做法、搞笑模仿
+输出：{"matched_demand_names": ["拜年祝福", "年夜饭做法"]}
+
+示例2：
+节点事件名称：九一八事变纪念日
+节点事件类型：纪念日
+视频特征词列表：搞笑模仿、宠物日常
+输出：{"matched_demand_names": []}
+
+示例3：
+节点事件名称：三伏
+节点事件类型：特定天气时段
+视频特征词列表：防暑小妙招、三伏贴教程、明星八卦
+输出：{"matched_demand_names": ["防暑小妙招", "三伏贴教程"]}
+
+现在，请根据上述规则处理用户输入。
+"""
+
+
+def _chunked(items: list[str], batch_size: int) -> list[list[str]]:
+    size = max(batch_size, 1)
+    return [items[index : index + size] for index in range(0, len(items), size)]
+
+
+def _build_demand_lookup(demand_names: list[str]) -> dict[str, str]:
+    lookup: dict[str, str] = {}
+    for item in demand_names:
+        demand_name = str(item).strip()
+        if not demand_name:
+            continue
+        lookup.setdefault(demand_name, demand_name)
+        compact_key = "".join(demand_name.split())
+        if compact_key:
+            lookup.setdefault(compact_key, demand_name)
+    return lookup
+
+
+def _resolve_demand_name(demand_name: str, demand_lookup: dict[str, str]) -> str | None:
+    value = demand_name.strip()
+    if not value:
+        return None
+    return demand_lookup.get(value) or demand_lookup.get("".join(value.split()))
+
+
+def _build_batch_user_message(
+    *,
+    festival_name: str,
+    event_type: str,
+    demand_names: list[str],
+) -> str:
+    feature_lines = "\n".join(
+        f"{index}. {demand_name}" for index, demand_name in enumerate(demand_names, start=1)
+    )
+    return f"""节点事件名称：{festival_name}
+节点事件类型：{event_type}
+
+视频特征词列表：
+{feature_lines}"""
+
+
+def _extract_matched_name_list(parsed: Any) -> list[Any] | None:
+    if not isinstance(parsed, dict):
+        return None
+
+    for key in ("matched_demand_names", "matched", "demand_names", "results"):
+        value = parsed.get(key)
+        if isinstance(value, list):
+            return value
+    return None
+
+
+def _normalize_matched_demand_names(
+    parsed: Any,
+    *,
+    festival_name: str,
+    event_type: str,
+    demand_names: list[str],
+    demand_lookup: dict[str, str],
+) -> list[dict[str, Any]]:
+    matched_raw = _extract_matched_name_list(parsed)
+    if matched_raw is None:
+        raise FestivalDemandError("llm output missing matched_demand_names")
+
+    allowed_demand_names = set(demand_names)
+    normalized: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for item in matched_raw:
+        if isinstance(item, str):
+            raw_name = item
+        elif isinstance(item, dict):
+            raw_name = str(
+                item.get("demand_name")
+                or item.get("feature_word")
+                or item.get("视频特征词")
+                or item.get("name")
+                or ""
+            )
+        else:
+            continue
+
+        demand_name = _resolve_demand_name(str(raw_name), demand_lookup)
+        if not demand_name or demand_name not in allowed_demand_names:
+            continue
+        if demand_name in seen:
+            continue
+        seen.add(demand_name)
+        normalized.append(
+            {
+                "demand_name": demand_name,
+                "festival_name": festival_name,
+                "event_type": event_type,
+            }
+        )
+    return normalized
+
+
+def _llm_match_festival_batch(
+    *,
+    festival: dict[str, Any],
+    demand_names: list[str],
+    config: FestivalDemandConfig,
+) -> list[dict[str, Any]]:
+    festival_name = str(festival.get("name") or "").strip()
+    event_type = str(festival.get("event_type") or "").strip()
+    if not festival_name or not demand_names:
+        return []
+
+    demand_lookup = _build_demand_lookup(demand_names)
+    user_message = _build_batch_user_message(
+        festival_name=festival_name,
+        event_type=event_type,
+        demand_names=demand_names,
+    )
+
+    last_error: Exception | None = None
+    for attempt in range(1, config.llm_max_attempts + 1):
+        try:
+            resp = create_chat_completion(
+                [
+                    {"role": "system", "content": MATCH_SYSTEM_PROMPT.strip()},
+                    {"role": "user", "content": user_message},
+                ],
+                model=config.llm_model,
+                temperature=config.llm_temperature,
+                max_tokens=config.llm_max_tokens,
+            )
+            parsed = extract_json_object(str(resp.get("content") or ""))
+            return _normalize_matched_demand_names(
+                parsed,
+                festival_name=festival_name,
+                event_type=event_type,
+                demand_names=demand_names,
+                demand_lookup=demand_lookup,
+            )
+        except (OpenRouterCallError, FestivalDemandError, ValueError) as exc:
+            last_error = exc
+            if attempt < config.llm_max_attempts:
+                time.sleep(config.llm_retry_sleep_seconds)
+
+    raise FestivalDemandError(
+        f"festival demand match failed for {festival_name} "
+        f"after {config.llm_max_attempts} attempts: {last_error}"
+    ) from last_error
+
+
+def match_demand_names_to_festivals(
+    *,
+    demand_names: list[str],
+    festivals: list[dict[str, Any]],
+    config: FestivalDemandConfig,
+) -> list[dict[str, Any]]:
+    """按批次将需求词与活跃节日列表做 LLM 匹配。"""
+    if not demand_names or not festivals:
+        return []
+
+    all_matches: list[dict[str, Any]] = []
+    seen: set[tuple[str, str]] = set()
+    batches = _chunked(demand_names, config.match_batch_size)
+    total_batches = len(batches)
+    for batch_index, batch in enumerate(batches, start=1):
+        for festival in festivals:
+            festival_name = str(festival.get("name") or "").strip()
+            print(
+                f"festival demand: matching batch {batch_index}/{total_batches} "
+                f"festival={festival_name} size={len(batch)}",
+                flush=True,
+            )
+            batch_matches = _llm_match_festival_batch(
+                festival=festival,
+                demand_names=batch,
+                config=config,
+            )
+            for row in batch_matches:
+                dedupe_key = (row["demand_name"], row["festival_name"])
+                if dedupe_key in seen:
+                    continue
+                seen.add(dedupe_key)
+                all_matches.append(row)
+    return all_matches
+
+
+def collect_matched_demand_names(matches: list[dict[str, Any]]) -> list[str]:
+    """收集去重后的 demand_name 列表。"""
+    names: list[str] = []
+    seen: set[str] = set()
+    for item in matches:
+        demand_name = str(item.get("demand_name") or "").strip()
+        if not demand_name or demand_name in seen:
+            continue
+        seen.add(demand_name)
+        names.append(demand_name)
+    return names

app/festival_demand/odps_demand_pool.py

@@ -0,0 +1,75 @@
+"""从 ODPS 需求池拉取 demand_name。"""
+
+from __future__ import annotations
+
+import re
+
+from app.aliyun_odps.client import get_odps_client
+from app.festival_demand.exceptions import FestivalDemandError
+
+IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?$")
+
+
+def _safe_identifier(name: str) -> str:
+    value = name.strip()
+    if not IDENTIFIER_RE.match(value):
+        raise FestivalDemandError(f"invalid sql identifier: {name}")
+    return value
+
+
+def _escape_sql_string(value: str) -> str:
+    return value.replace("'", "''")
+
+
+def _normalize_demand_key(value: str) -> str:
+    return "".join(value.split())
+
+
+def _dedupe_demand_names(demand_names: list[str]) -> list[str]:
+    deduped: list[str] = []
+    seen: set[str] = set()
+    for raw_name in demand_names:
+        demand_name = str(raw_name).strip()
+        if not demand_name:
+            continue
+        keys = {demand_name, _normalize_demand_key(demand_name)}
+        if keys & seen:
+            continue
+        seen.update(keys)
+        deduped.append(demand_name)
+    return deduped
+
+
+def fetch_demand_names(
+    *,
+    partition_dt: str,
+    strategy: str,
+    source_table: str,
+) -> list[str]:
+    """按分区日期与 strategy 从 ODPS 需求池拉取 demand_name。"""
+    table = _safe_identifier(source_table)
+    dt = _escape_sql_string(partition_dt.strip())
+    strategy_value = _escape_sql_string(strategy.strip())
+    if not dt:
+        raise FestivalDemandError("partition_dt is required")
+    if not strategy_value:
+        raise FestivalDemandError("strategy is required")
+
+    sql = f"""
+    SELECT TRIM(demand_name) AS demand_name
+    FROM {table}
+    WHERE dt = '{dt}'
+      AND strategy = '{strategy_value}'
+      AND demand_name IS NOT NULL
+      AND TRIM(demand_name) <> ''
+    """
+
+    odps_client = get_odps_client()
+    instance = odps_client.execute_sql(sql)
+    raw_demand_names: list[str] = []
+    with instance.open_reader(tunnel=True) as reader:
+        for record in reader:
+            demand_name = str(record["demand_name"] or "").strip()
+            if demand_name:
+                raw_demand_names.append(demand_name)
+    return _dedupe_demand_names(raw_demand_names)

app/festival_demand/odps_writer.py

@@ -0,0 +1,94 @@
+"""节日需求写入 ODPS 需求池表。"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from app.aliyun_odps.client import get_odps_client
+from app.festival_demand.exceptions import FestivalDemandError
+
+IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?$")
+
+
+def _safe_identifier(name: str) -> str:
+    value = name.strip()
+    if not IDENTIFIER_RE.match(value):
+        raise FestivalDemandError(f"invalid sql identifier: {name}")
+    return value
+
+
+def _escape_sql_string(value: str) -> str:
+    return value.replace("'", "''")
+
+
+def build_odps_row(
+    *,
+    strategy: str,
+    demand_id: str,
+    demand_name: str,
+    weight: float = 1.0,
+    demand_type: str = "特征点",
+) -> dict[str, Any]:
+    return {
+        "strategy": strategy,
+        "demand_id": demand_id,
+        "demand_name": demand_name,
+        "weight": weight,
+        "type": demand_type,
+        "video_count": 0,
+        "extend": "",
+    }
+
+
+def _build_row_select(row: dict[str, Any]) -> str:
+    strategy = _escape_sql_string(str(row["strategy"]))
+    demand_id = _escape_sql_string(str(row["demand_id"]))
+    demand_name = _escape_sql_string(str(row["demand_name"]))
+    weight = float(row.get("weight") or 1.0)
+    demand_type = _escape_sql_string(str(row.get("type") or "特征点"))
+    extend = _escape_sql_string(str(row.get("extend") or ""))
+    return f"""
+        SELECT
+            '{strategy}' AS strategy,
+            '{demand_id}' AS demand_id,
+            '{demand_name}' AS demand_name,
+            {weight} AS weight,
+            '{demand_type}' AS type,
+            CAST(0 AS BIGINT) AS video_count,
+            array() AS video_list,
+            '{extend}' AS extend
+    """
+
+
+def sync_festival_demands_to_odps(
+    *,
+    rows: list[dict[str, Any]],
+    partition_dt: str,
+    target_table: str,
+) -> dict[str, Any]:
+    """将节日需求追加写入 ODPS 分区（不覆盖分区已有数据）。"""
+    partition_value = partition_dt.strip()
+    if not partition_value:
+        raise FestivalDemandError("partition_dt is required")
+    if not rows:
+        return {
+            "target_table": target_table,
+            "partition_dt": partition_value,
+            "written_count": 0,
+        }
+
+    table_name = _safe_identifier(target_table)
+    odps_client = get_odps_client()
+    select_sql = " UNION ALL ".join(_build_row_select(row) for row in rows)
+    sql = f"""
+    INSERT INTO TABLE {table_name} PARTITION (dt='{_escape_sql_string(partition_value)}')
+    {select_sql}
+    """
+    instance = odps_client.execute_sql(sql)
+    instance.wait_for_success()
+    return {
+        "target_table": target_table,
+        "partition_dt": partition_value,
+        "written_count": len(rows),
+    }

app/festival_demand/repository.py

@@ -0,0 +1,203 @@
+"""节日需求 MySQL 仓储。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+try:
+    import pymysql
+    from pymysql.cursors import DictCursor
+except ImportError:  # pragma: no cover
+    pymysql = None
+    DictCursor = None
+
+from app.festival_demand.demand_id import build_festival_demand_id
+from app.festival_demand.exceptions import FestivalDemandError
+from app.hot_content.types import MysqlConfig
+
+TABLE_NAME = "festival_demand_pool_di"
+
+
+class FestivalDemandRepository:
+    def __init__(self, config: MysqlConfig):
+        if pymysql is None or DictCursor is None:
+            raise FestivalDemandError("missing dependency: pip install pymysql")
+        self.conn = pymysql.connect(
+            host=config.host,
+            port=config.port,
+            user=config.user,
+            password=config.password,
+            database=config.database,
+            charset=config.charset,
+            autocommit=True,
+            cursorclass=DictCursor,
+        )
+        self._ensure_table()
+
+    def close(self) -> None:
+        self.conn.close()
+
+    def _ensure_table(self) -> None:
+        sql = f"""
+            CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
+                id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '自增主键',
+                strategy VARCHAR(128) NOT NULL COMMENT '策略',
+                demand_id CHAR(32) NOT NULL COMMENT '需求id',
+                demand_name VARCHAR(1024) NOT NULL COMMENT '需求',
+                partition_dt VARCHAR(8) NOT NULL DEFAULT '' COMMENT '业务日期 yyyymmdd',
+                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+                    ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+                PRIMARY KEY (id),
+                UNIQUE KEY uk_festival_demand_pool (strategy, demand_id),
+                KEY idx_strategy_partition (strategy, partition_dt),
+                KEY idx_demand_name (demand_name(191))
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+        """
+        with self.conn.cursor() as cursor:
+            cursor.execute(sql)
+            self._ensure_partition_dt_column(cursor)
+
+    @staticmethod
+    def _ensure_partition_dt_column(cursor: Any) -> None:
+        cursor.execute(
+            f"""
+            SELECT COUNT(*) AS cnt
+            FROM information_schema.COLUMNS
+            WHERE TABLE_SCHEMA = DATABASE()
+              AND TABLE_NAME = '{TABLE_NAME}'
+              AND COLUMN_NAME = 'partition_dt'
+            """
+        )
+        row = cursor.fetchone() or {}
+        if int(row.get("cnt") or 0) > 0:
+            return
+        cursor.execute(
+            f"""
+            ALTER TABLE {TABLE_NAME}
+            ADD COLUMN partition_dt VARCHAR(8) NOT NULL DEFAULT '' COMMENT '业务日期 yyyymmdd'
+            AFTER demand_name,
+            ADD KEY idx_strategy_partition (strategy, partition_dt)
+            """
+        )
+
+    def has_partition_data(self, *, strategy: str, partition_dt: str) -> bool:
+        strategy_value = strategy.strip()
+        partition_value = partition_dt.strip()
+        if not strategy_value or not partition_value:
+            return False
+        sql = f"""
+            SELECT 1
+            FROM {TABLE_NAME}
+            WHERE strategy = %s
+              AND partition_dt = %s
+            LIMIT 1
+        """
+        with self.conn.cursor() as cursor:
+            cursor.execute(sql, (strategy_value, partition_value))
+            return cursor.fetchone() is not None
+
+    def build_demand_rows(
+        self,
+        *,
+        demand_names: list[str],
+        strategy: str,
+        partition_dt: str,
+    ) -> list[dict[str, str]]:
+        strategy_value = strategy.strip()
+        partition_value = partition_dt.strip()
+        if not strategy_value:
+            raise FestivalDemandError("festival demand output strategy is required")
+        if not partition_value:
+            raise FestivalDemandError("partition_dt is required")
+
+        rows: list[dict[str, str]] = []
+        seen_demand_ids: set[str] = set()
+        for raw_name in demand_names:
+            demand_name = str(raw_name).strip()
+            if not demand_name:
+                continue
+            demand_id = build_festival_demand_id(
+                strategy=strategy_value,
+                demand_name=demand_name,
+                partition_dt=partition_value,
+            )
+            if demand_id in seen_demand_ids:
+                continue
+            seen_demand_ids.add(demand_id)
+            rows.append(
+                {
+                    "strategy": strategy_value,
+                    "demand_id": demand_id,
+                    "demand_name": demand_name,
+                    "partition_dt": partition_value,
+                }
+            )
+        return rows
+
+    def insert_demand_rows(self, rows: list[dict[str, str]]) -> int:
+        if not rows:
+            return 0
+        sql = f"""
+            INSERT INTO {TABLE_NAME} (
+                strategy,
+                demand_id,
+                demand_name,
+                partition_dt
+            )
+            VALUES (%s, %s, %s, %s)
+        """
+        payload = [
+            (
+                row["strategy"],
+                row["demand_id"],
+                row["demand_name"],
+                row["partition_dt"],
+            )
+            for row in rows
+        ]
+        with self.conn.cursor() as cursor:
+            cursor.executemany(sql, payload)
+        return len(rows)
+
+    def persist_generated_demands(
+        self,
+        *,
+        demand_names: list[str],
+        strategy: str,
+        partition_dt: str,
+    ) -> dict[str, Any]:
+        """当天已有数据则跳过；否则写入 MySQL 并返回待同步 ODPS 的行。"""
+        strategy_value = strategy.strip()
+        partition_value = partition_dt.strip()
+        base_result = {
+            "table": TABLE_NAME,
+            "strategy": strategy_value,
+            "partition_dt": partition_value,
+            "input_count": len(demand_names),
+            "skipped": False,
+            "saved_count": 0,
+            "rows": [],
+        }
+
+        if self.has_partition_data(strategy=strategy_value, partition_dt=partition_value):
+            return {
+                **base_result,
+                "skipped": True,
+                "skip_reason": "partition_data_exists",
+            }
+
+        rows = self.build_demand_rows(
+            demand_names=demand_names,
+            strategy=strategy_value,
+            partition_dt=partition_value,
+        )
+        if not rows:
+            return base_result
+
+        saved_count = self.insert_demand_rows(rows)
+        return {
+            **base_result,
+            "saved_count": saved_count,
+            "rows": rows,
+        }

app/festival_demand/service.py

@@ -0,0 +1,263 @@
+"""节日需求每日任务入口。"""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+from typing import Any
+
+from app.festival_demand.config import load_festival_demand_config
+from app.festival_demand.demand_generation import generate_demands_from_matches
+from app.festival_demand.detection import detect_active_festivals
+from app.festival_demand.matching import (
+    collect_matched_demand_names,
+    match_demand_names_to_festivals,
+)
+from app.festival_demand.odps_demand_pool import fetch_demand_names
+from app.festival_demand.odps_writer import build_odps_row, sync_festival_demands_to_odps
+from app.festival_demand.repository import FestivalDemandRepository
+from app.festival_demand.types import FestivalDemandConfig
+from app.hot_content.config import load_flow_config
+from app.hot_content.timezone import SHANGHAI_TZ
+
+
+def _today_shanghai() -> date:
+    return datetime.now(SHANGHAI_TZ).date()
+
+
+def print_festival_demand_summary(summary: dict[str, Any]) -> None:
+    """将节日需求流程结果打印到控制台。"""
+    print("=" * 60)
+    print("节日需求任务结果")
+    print("=" * 60)
+    print(f"查询日期: {summary.get('query_date')}")
+    print(f"ODPS 分区: {summary.get('partition_dt')}")
+    print(f"需求池 strategy: {summary.get('demand_pool_strategy')}")
+
+    if summary.get("skip_reason") in {
+        "no_active_festivals",
+        "skip_odps",
+        "no_demand_names",
+        "partition_data_exists",
+    }:
+        print(f"跳过原因: {summary.get('skip_reason')}")
+        print("=" * 60)
+        return
+
+    festival_names = summary.get("festival_names") or []
+    print(f"\n活跃节日 ({len(festival_names)}):")
+    for name in festival_names:
+        print(f"  - {name}")
+
+    print(f"\nODPS 需求词总数: {summary.get('demand_name_total', 0)}")
+    print(f"匹配特征词数: {summary.get('matched_count', 0)}")
+
+    matched_names = summary.get("matched_demand_names") or []
+    if matched_names:
+        print("\n匹配到的特征词:")
+        for name in matched_names:
+            print(f"  * {name}")
+
+    demands_by_type = summary.get("demands_by_type") or {}
+    type_labels = {
+        "year_festival": "方式1: 年份 + 节日",
+        "festival": "方式2: 节日",
+        "year_festival_feature": "方式3: 年份 + 节日 + 特征点",
+    }
+    generated_names = summary.get("generated_demand_names") or []
+    print(f"\n生成需求总数: {summary.get('generated_demand_count', len(generated_names))}")
+    for generation_type, label in type_labels.items():
+        names = demands_by_type.get(generation_type) or []
+        if not names:
+            continue
+        print(f"\n{label} ({len(names)}):")
+        for name in names:
+            print(f"  + {name}")
+
+    if not generated_names:
+        print("\n（暂无生成需求）")
+
+    mysql_save = summary.get("mysql_save") or {}
+    if mysql_save.get("skipped"):
+        print(
+            f"\nMySQL/ODPS 跳过: reason={mysql_save.get('skip_reason')} "
+            f"partition_dt={mysql_save.get('partition_dt')}"
+        )
+    elif mysql_save:
+        print(
+            f"\nMySQL 写入: table={mysql_save.get('table')} "
+            f"saved={mysql_save.get('saved_count', 0)} "
+            f"strategy={mysql_save.get('strategy')}"
+        )
+        odps_sync = summary.get("odps_sync") or {}
+        if odps_sync:
+            print(
+                f"ODPS 写入: table={odps_sync.get('target_table')} "
+                f"partition_dt={odps_sync.get('partition_dt')} "
+                f"written={odps_sync.get('written_count', 0)}"
+            )
+
+    print("=" * 60)
+
+
+def _persist_generated_demands(
+    *,
+    demand_names: list[str],
+    strategy: str,
+    partition_dt: str,
+    target_table: str,
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    flow_config = load_flow_config()
+    repository = FestivalDemandRepository(flow_config.mysql)
+    try:
+        mysql_save = repository.persist_generated_demands(
+            demand_names=demand_names,
+            strategy=strategy,
+            partition_dt=partition_dt,
+        )
+        if mysql_save.get("skipped") or not mysql_save.get("rows"):
+            return mysql_save, {}
+
+        odps_rows = [
+            build_odps_row(
+                strategy=row["strategy"],
+                demand_id=row["demand_id"],
+                demand_name=row["demand_name"],
+            )
+            for row in mysql_save["rows"]
+        ]
+        odps_sync = sync_festival_demands_to_odps(
+            rows=odps_rows,
+            partition_dt=partition_dt,
+            target_table=target_table,
+        )
+        return mysql_save, odps_sync
+    finally:
+        repository.close()
+
+
+def run_festival_demand_daily_job(
+    config: FestivalDemandConfig,
+    *,
+    query_date: date | None = None,
+    skip_odps: bool = False,
+) -> dict[str, Any]:
+    """执行节日需求流程：检测活跃节日 → 拉取 ODPS 需求词 → LLM 批量匹配。"""
+    target_date = query_date or _today_shanghai()
+    partition_dt = target_date.strftime("%Y%m%d")
+
+    detection = detect_active_festivals(target_date, config)
+    festivals = detection.get("festivals") or []
+
+    summary: dict[str, Any] = {
+        "query_date": target_date.isoformat(),
+        "partition_dt": partition_dt,
+        "demand_pool_strategy": config.demand_pool_strategy,
+        "output_strategy": config.output_strategy,
+        "demand_pool_source_table": config.demand_pool_source_table,
+        "festival_count": len(festivals),
+        "festival_names": detection.get("festival_names") or [],
+        "festivals": festivals,
+        "demand_name_total": 0,
+        "match_batch_size": config.match_batch_size,
+        "matched_count": 0,
+        "matched_demand_names": [],
+        "matches": [],
+        "generated_demand_count": 0,
+        "generated_demand_names": [],
+        "generated_demands": [],
+        "demands_by_type": {
+            "year_festival": [],
+            "festival": [],
+            "year_festival_feature": [],
+        },
+        "mysql_save": {},
+        "odps_sync": {},
+    }
+
+    if not festivals:
+        summary["skip_reason"] = "no_active_festivals"
+        return summary
+
+    if skip_odps:
+        summary["skip_reason"] = "skip_odps"
+        return summary
+
+    flow_config = load_flow_config()
+    repository = FestivalDemandRepository(flow_config.mysql)
+    try:
+        if repository.has_partition_data(
+            strategy=config.output_strategy,
+            partition_dt=partition_dt,
+        ):
+            summary["skip_reason"] = "partition_data_exists"
+            summary["mysql_save"] = {
+                "table": "festival_demand_pool_di",
+                "strategy": config.output_strategy,
+                "partition_dt": partition_dt,
+                "skipped": True,
+                "skip_reason": "partition_data_exists",
+            }
+            return summary
+    finally:
+        repository.close()
+
+    demand_names = fetch_demand_names(
+        partition_dt=partition_dt,
+        strategy=config.demand_pool_strategy,
+        source_table=config.demand_pool_source_table,
+    )
+    summary["demand_name_total"] = len(demand_names)
+
+    if not demand_names:
+        summary["skip_reason"] = "no_demand_names"
+        return summary
+
+    print(
+        f"festival demand: start matching "
+        f"festivals={len(festivals)} demand_names={len(demand_names)} "
+        f"batch_size={config.match_batch_size}",
+        flush=True,
+    )
+    matches = match_demand_names_to_festivals(
+        demand_names=demand_names,
+        festivals=festivals,
+        config=config,
+    )
+    matched_demand_names = collect_matched_demand_names(matches)
+    generated = generate_demands_from_matches(matches, query_date=target_date)
+
+    summary["matched_count"] = len(matched_demand_names)
+    summary["matched_demand_names"] = matched_demand_names
+    summary["matches"] = matches
+    summary["generated_demand_count"] = generated["generated_demand_count"]
+    summary["generated_demand_names"] = generated["generated_demand_names"]
+    summary["generated_demands"] = generated["generated_demands"]
+    summary["demands_by_type"] = generated["demands_by_type"]
+
+    if generated["generated_demand_names"]:
+        print("festival demand: persisting generated demands", flush=True)
+        mysql_save, odps_sync = _persist_generated_demands(
+            demand_names=generated["generated_demand_names"],
+            strategy=config.output_strategy,
+            partition_dt=partition_dt,
+            target_table=config.demand_pool_source_table,
+        )
+        summary["mysql_save"] = mysql_save
+        summary["odps_sync"] = odps_sync
+        if mysql_save.get("skipped"):
+            summary["skip_reason"] = mysql_save.get("skip_reason")
+    return summary
+
+
+def test_festivals_for_date(
+    query_date: date,
+    *,
+    skip_odps: bool = False,
+) -> dict[str, Any]:
+    """测试指定日期的节日需求流程。"""
+    config = load_festival_demand_config()
+    return run_festival_demand_daily_job(
+        config,
+        query_date=query_date,
+        skip_odps=skip_odps,
+    )

app/festival_demand/types.py

@@ -0,0 +1,20 @@
+"""节日需求流程类型定义。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class FestivalDemandConfig:
+    cron_hours: str
+    cron_minute: int
+    llm_model: str
+    llm_max_attempts: int
+    llm_retry_sleep_seconds: float
+    llm_max_tokens: int
+    llm_temperature: float
+    demand_pool_source_table: str
+    demand_pool_strategy: str
+    output_strategy: str
+    match_batch_size: int

app/festival_demand/validation.py

@@ -0,0 +1,74 @@
+"""节日结果过滤：仅校验查询日期是否在 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

app/scheduler.py

@@ -23,6 +23,9 @@ from app.hot_content.timezone import SHANGHAI_TZ
 from app.hot_content.types import FlowConfig
 from app.hot_content.wxindex_words import run_wxindex_words_daily_job
 from app.hot_content.wxindex_heat_pattern import run_wxindex_heat_pattern_daily_job
+from app.festival_demand.config import load_festival_demand_config
+from app.festival_demand.service import run_festival_demand_daily_job, print_festival_demand_summary
+from app.festival_demand.types import FestivalDemandConfig
 
 
 def _import_blocking_scheduler() -> Any:
@@ -115,6 +118,21 @@ def run_wxindex_heat_pattern_job(config: FlowConfig) -> None:
         repository.close()
 
 
+def run_festival_demand_job(config: FestivalDemandConfig) -> None:
+    try:
+        summary = run_festival_demand_daily_job(config)
+        print_festival_demand_summary(summary)
+        print(
+            json.dumps(
+                {"job": "festival_demand", "summary": summary},
+                ensure_ascii=False,
+                indent=2,
+            )
+        )
+    except Exception as exc:
+        print(f"festival demand failed: {exc}", file=sys.stderr)
+
+
 def register_hot_content_job(scheduler: Any, config: FlowConfig) -> None:
     scheduler.add_job(
         run_hot_content_job,
@@ -179,23 +197,42 @@ def register_wxindex_heat_pattern_job(scheduler: Any, config: FlowConfig) -> Non
     )
 
 
+def register_festival_demand_job(scheduler: Any, config: FestivalDemandConfig) -> None:
+    scheduler.add_job(
+        run_festival_demand_job,
+        trigger="cron",
+        hour=config.cron_hours,
+        minute=config.cron_minute,
+        timezone=SHANGHAI_TZ,
+        args=[config],
+        id="festival_demand",
+        name="节日需求：活跃节日检测 + ODPS 需求词匹配",
+        replace_existing=True,
+        coalesce=True,
+        max_instances=1,
+    )
+
+
 def start_scheduler() -> None:
     BlockingScheduler = _import_blocking_scheduler()
     scheduler = BlockingScheduler(timezone=SHANGHAI_TZ)
     config = load_flow_config()
+    festival_config = load_festival_demand_config()
     register_hot_content_job(scheduler, config)
     register_decode_result_job(scheduler, config)
     register_wxindex_words_refresh_job(scheduler, config)
     register_wxindex_heat_pattern_job(scheduler, config)
+    register_festival_demand_job(scheduler, festival_config)
     print(
         "scheduler started, timezone=Asia/Shanghai, "
         "jobs=['hot_content_flow', 'decode_result_flow', 'wxindex_words_refresh', "
-        "'wxindex_heat_pattern'], "
+        "'wxindex_heat_pattern', 'festival_demand'], "
         f"hot_cron={config.hot_flow_cron_hours}:{config.hot_flow_cron_minute:02d}, "
         f"decode_result_interval={config.decode_result_interval_seconds}s, "
         f"wxindex_words_cron={config.wxindex_words_cron_hours}:{config.wxindex_words_cron_minute:02d}, "
         f"wxindex_heat_pattern_cron="
-        f"{config.wxindex_heat_pattern_cron_hours}:{config.wxindex_heat_pattern_cron_minute:02d}"
+        f"{config.wxindex_heat_pattern_cron_hours}:{config.wxindex_heat_pattern_cron_minute:02d}, "
+        f"festival_demand_cron={festival_config.cron_hours}:{festival_config.cron_minute:02d}"
     )
     scheduler.start()
 
@@ -205,7 +242,15 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--once", action="store_true", help="执行一次，不启动调度器")
     parser.add_argument(
         "--job",
-        choices=("all", "hot-content", "decode-result", "postprocess", "wxindex-refresh", "wxindex-heat-pattern"),
+        choices=(
+            "all",
+            "hot-content",
+            "decode-result",
+            "postprocess",
+            "wxindex-refresh",
+            "wxindex-heat-pattern",
+            "festival-demand",
+        ),
         default="all",
         help="--once 时选择执行哪个任务",
     )
@@ -255,6 +300,9 @@ def main() -> None:
             run_wxindex_words_refresh_job(config)
         if args.job in {"wxindex-heat-pattern"}:
             run_wxindex_heat_pattern_job(config)
+        if args.job in {"festival-demand"}:
+            festival_config = load_festival_demand_config()
+            run_festival_demand_job(festival_config)
         return
     start_scheduler()
 

docker-compose.yml

@@ -30,6 +30,12 @@ services:
       WXINDEX_WORDS_CRON_MINUTE: ${WXINDEX_WORDS_CRON_MINUTE:-0}
       WXINDEX_HEAT_PATTERN_CRON_HOUR: ${WXINDEX_HEAT_PATTERN_CRON_HOUR:-11}
       WXINDEX_HEAT_PATTERN_CRON_MINUTE: ${WXINDEX_HEAT_PATTERN_CRON_MINUTE:-0}
+      FESTIVAL_DEMAND_CRON_HOURS: ${FESTIVAL_DEMAND_CRON_HOURS:-7}
+      FESTIVAL_DEMAND_CRON_MINUTE: ${FESTIVAL_DEMAND_CRON_MINUTE:-0}
+      FESTIVAL_DEMAND_LLM_MODEL: ${FESTIVAL_DEMAND_LLM_MODEL:-}
+      FESTIVAL_DEMAND_POOL_STRATEGY: ${FESTIVAL_DEMAND_POOL_STRATEGY:-去年同期阳历}
+      FESTIVAL_DEMAND_OUTPUT_STRATEGY: ${FESTIVAL_DEMAND_OUTPUT_STRATEGY:-去年同期阳历-节点事件}
+      FESTIVAL_DEMAND_MATCH_BATCH_SIZE: ${FESTIVAL_DEMAND_MATCH_BATCH_SIZE:-200}
       # 业务阈值
       WXINDEX_SCORE_THRESHOLD: ${WXINDEX_SCORE_THRESHOLD:-1000000}
       DEMAND_POOL_SOURCE_TABLE: ${DEMAND_POOL_SOURCE_TABLE:-dwd_multi_demand_pool_di}