增加新的策略

app/core/config.py

@@ -190,6 +190,17 @@ class Settings:
     festival_demand_output_strategy: str = "去年同期阳历-节点事件"
     festival_demand_match_batch_size: int = 200
 
+    gap_script_demand_cron_hours: str = "10"
+    gap_script_demand_cron_minute: int = 0
+    gap_script_demand_llm_model: str = "openai/gpt-4o"
+    gap_script_demand_llm_max_attempts: int = 3
+    gap_script_demand_llm_retry_sleep_seconds: float = 1.0
+    gap_script_demand_llm_max_tokens: int = 4000
+    gap_script_demand_llm_temperature: float = 0.0
+    gap_script_demand_pool_strategy: str = "当下供需gap"
+    gap_script_demand_output_strategy: str = "当下供需gap-脚本主驱"
+    gap_script_demand_judge_batch_size: int = 40
+
     @classmethod
     def from_env(cls) -> "Settings":
         defaults = cls()
@@ -462,6 +473,46 @@ class Settings:
                 "FESTIVAL_DEMAND_MATCH_BATCH_SIZE",
                 defaults.festival_demand_match_batch_size,
             ),
+            gap_script_demand_cron_hours=_env(
+                "GAP_SCRIPT_DEMAND_CRON_HOURS",
+                defaults.gap_script_demand_cron_hours,
+            ),
+            gap_script_demand_cron_minute=_env_int(
+                "GAP_SCRIPT_DEMAND_CRON_MINUTE",
+                defaults.gap_script_demand_cron_minute,
+            ),
+            gap_script_demand_llm_model=_env(
+                "GAP_SCRIPT_DEMAND_LLM_MODEL",
+                defaults.gap_script_demand_llm_model,
+            ),
+            gap_script_demand_llm_max_attempts=_env_int(
+                "GAP_SCRIPT_DEMAND_LLM_MAX_ATTEMPTS",
+                defaults.gap_script_demand_llm_max_attempts,
+            ),
+            gap_script_demand_llm_retry_sleep_seconds=_env_float(
+                "GAP_SCRIPT_DEMAND_LLM_RETRY_SLEEP_SECONDS",
+                defaults.gap_script_demand_llm_retry_sleep_seconds,
+            ),
+            gap_script_demand_llm_max_tokens=_env_int(
+                "GAP_SCRIPT_DEMAND_LLM_MAX_TOKENS",
+                defaults.gap_script_demand_llm_max_tokens,
+            ),
+            gap_script_demand_llm_temperature=_env_float(
+                "GAP_SCRIPT_DEMAND_LLM_TEMPERATURE",
+                defaults.gap_script_demand_llm_temperature,
+            ),
+            gap_script_demand_pool_strategy=_env(
+                "GAP_SCRIPT_DEMAND_POOL_STRATEGY",
+                defaults.gap_script_demand_pool_strategy,
+            ),
+            gap_script_demand_output_strategy=_env(
+                "GAP_SCRIPT_DEMAND_OUTPUT_STRATEGY",
+                defaults.gap_script_demand_output_strategy,
+            ),
+            gap_script_demand_judge_batch_size=_env_int(
+                "GAP_SCRIPT_DEMAND_JUDGE_BATCH_SIZE",
+                defaults.gap_script_demand_judge_batch_size,
+            ),
         )
 
 

app/gap_script_demand/__init__.py

@@ -0,0 +1 @@
+"""当下供需 gap 脚本主驱筛选定时任务。"""

app/gap_script_demand/config.py

@@ -0,0 +1,102 @@
+"""当下供需 gap 脚本主驱流程配置。"""
+
+from __future__ import annotations
+
+import os
+
+from app.core.config import settings
+from app.gap_script_demand.exceptions import GapScriptDemandError
+
+
+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 GapScriptDemandError(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 GapScriptDemandError(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 GapScriptDemandError("gap script demand cron hours cannot be empty")
+    normalized: list[str] = []
+    for hour in hours:
+        try:
+            hour_num = int(hour)
+        except ValueError as exc:
+            raise GapScriptDemandError(f"invalid gap script demand cron hour: {hour!r}") from exc
+        if not 0 <= hour_num <= 23:
+            raise GapScriptDemandError(f"gap script demand cron hour out of range: {hour_num}")
+        normalized.append(str(hour_num))
+    return ",".join(normalized)
+
+
+def load_gap_script_demand_config() -> "GapScriptDemandConfig":
+    from app.gap_script_demand.types import GapScriptDemandConfig
+
+    llm_model = _get_env("GAP_SCRIPT_DEMAND_LLM_MODEL", settings.gap_script_demand_llm_model)
+    config = GapScriptDemandConfig(
+        cron_hours=_parse_cron_hours(
+            _get_env("GAP_SCRIPT_DEMAND_CRON_HOURS", settings.gap_script_demand_cron_hours)
+        ),
+        cron_minute=_get_env_int(
+            "GAP_SCRIPT_DEMAND_CRON_MINUTE",
+            settings.gap_script_demand_cron_minute,
+        ),
+        llm_model=llm_model.strip() or settings.open_router_default_model,
+        llm_max_attempts=_get_env_int(
+            "GAP_SCRIPT_DEMAND_LLM_MAX_ATTEMPTS",
+            settings.gap_script_demand_llm_max_attempts,
+        ),
+        llm_retry_sleep_seconds=_get_env_float(
+            "GAP_SCRIPT_DEMAND_LLM_RETRY_SLEEP_SECONDS",
+            settings.gap_script_demand_llm_retry_sleep_seconds,
+        ),
+        llm_max_tokens=_get_env_int(
+            "GAP_SCRIPT_DEMAND_LLM_MAX_TOKENS",
+            settings.gap_script_demand_llm_max_tokens,
+        ),
+        llm_temperature=_get_env_float(
+            "GAP_SCRIPT_DEMAND_LLM_TEMPERATURE",
+            settings.gap_script_demand_llm_temperature,
+        ),
+        demand_pool_source_table=_get_env(
+            "DEMAND_POOL_SOURCE_TABLE",
+            settings.demand_pool_source_table,
+        ).strip(),
+        demand_pool_strategy=_get_env(
+            "GAP_SCRIPT_DEMAND_POOL_STRATEGY",
+            settings.gap_script_demand_pool_strategy,
+        ).strip(),
+        output_strategy=_get_env(
+            "GAP_SCRIPT_DEMAND_OUTPUT_STRATEGY",
+            settings.gap_script_demand_output_strategy,
+        ).strip(),
+        judge_batch_size=_get_env_int(
+            "GAP_SCRIPT_DEMAND_JUDGE_BATCH_SIZE",
+            settings.gap_script_demand_judge_batch_size,
+        ),
+    )
+    if config.judge_batch_size <= 0:
+        raise GapScriptDemandError("GAP_SCRIPT_DEMAND_JUDGE_BATCH_SIZE must be positive")
+    return config

app/gap_script_demand/demand_id.py

@@ -0,0 +1,16 @@
+"""脚本主驱需求 demand_id 生成。"""
+
+from __future__ import annotations
+
+import hashlib
+
+
+def build_gap_script_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/gap_script_demand/exceptions.py

@@ -0,0 +1,5 @@
+"""当下供需 gap 脚本主驱流程异常。"""
+
+
+class GapScriptDemandError(Exception):
+    """当下供需 gap 脚本主驱流程业务异常。"""

app/gap_script_demand/judging.py

@@ -0,0 +1,197 @@
+"""品类+解构词 画面改造可改造性 LLM 批量判定。"""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+from app.core.open_router_llm import OpenRouterCallError, create_chat_completion
+from app.gap_script_demand.exceptions import GapScriptDemandError
+from app.gap_script_demand.llm_json import extract_json_object
+from app.gap_script_demand.types import GapScriptDemandConfig
+
+JUDGE_SYSTEM_PROMPT = """
+你是一个专业的视频内容可改造性评估专家。你的任务是对给定的“品类 + 解构词”组合进行判定，筛选出其中适合用于画面改造的条目。
+背景定义（已内置，无需输出解释）
+- 画面改造：指替换原视频的画面与语音包，达到类似洗稿的效果。
+- 解构词：以“品类 解构词”形式呈现的内容标签。
+筛选目标
+仅保留那些能够通过替换画面和语音包，生成一个主题明确、内容可独立成立的新视频的解构词组合。
+淘汰标准（满足任意一条即判定为“不匹配”）
+1. 画面主导型内容：原视频的核心价值几乎完全依赖视觉呈现，缺乏可被文字/语音替代的叙事或信息逻辑。典型示例包括但不限于：
+  - 美景分享（如风景航拍、城市掠影）
+  - 表演类（魔术、舞蹈、杂技、花式运动）
+  - 旅行记录（无解说/无主题的Vlog）
+  - 纯音乐/演奏（无歌词或无故事线）
+  - 祝福类（节日祝贺、生日祝福画面）
+  - 宠物日常（无情节的萌宠片段） 此类内容即使替换画面和语音，也无法保留原有吸引力或会彻底改变内容性质，故不可改造。
+2. 无意义/弱主题解构词：解构词本身过于宽泛、模糊或与品类关联度极低，无法推导出具体的可替换内容框架。例如：
+  - 解构词为“日常”“合集”“精选”“欣赏”“感受”等无法指向具体场景、动作或叙事的词汇；
+  - 解构词与品类组合后，仍不能明确该视频要表达什么事件、过程或观点（如“美食 好吃”“旅行 好看”）。
+判定原则
+- 综合判断：必须同时参考品类和解构词，两者共同决定主题明确性。例如：
+  - “美食 教程” → 主题明确（有操作流程），可改造 ✅
+  - “美食 展示” → 仅画面陈列，无过程，不可改造 ❌
+  - “舞蹈 教学” → 有动作拆解和语音指导，可改造 ✅
+  - “舞蹈 表演” → 纯视觉观赏，不可改造 ❌
+- 主题可迁移性：如果替换画面和语音后，新视频仍能传递相同的信息价值（如知识、步骤、故事、观点），则判定为可改造；否则为不可改造。
+
+输出要求
+严格只输出一个 JSON 对象，禁止输出 JSON 之外的任何内容。固定格式如下：
+{
+  "matched_demand_names": ["可改造的品类 解构词1", "可改造的品类 解构词2"]
+}
+
+约束：
+- 只返回判定为可改造（匹配）的条目，不匹配的不要放入数组。
+- matched_demand_names 中的每一项必须与用户提供的原文完全一致，不得改写、不得编造。
+- 若没有任何条目可改造，返回 {"matched_demand_names": []}。
+""".strip()
+
+
+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(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"""请对以下“品类 解构词”组合逐条判定是否适合画面改造：
+
+{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,
+    *,
+    demand_names: list[str],
+    demand_lookup: dict[str, str],
+) -> list[str]:
+    matched_raw = _extract_matched_name_list(parsed)
+    if matched_raw is None:
+        raise GapScriptDemandError("llm output missing matched_demand_names")
+
+    allowed_demand_names = set(demand_names)
+    normalized: list[str] = []
+    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("name")
+                or item.get("品类 解构词")
+                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)
+    return normalized
+
+
+def _llm_judge_batch(
+    *,
+    demand_names: list[str],
+    config: GapScriptDemandConfig,
+) -> list[str]:
+    if not demand_names:
+        return []
+
+    demand_lookup = _build_demand_lookup(demand_names)
+    user_message = _build_batch_user_message(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": JUDGE_SYSTEM_PROMPT},
+                    {"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,
+                demand_names=demand_names,
+                demand_lookup=demand_lookup,
+            )
+        except (OpenRouterCallError, GapScriptDemandError, ValueError) as exc:
+            last_error = exc
+            if attempt < config.llm_max_attempts:
+                time.sleep(config.llm_retry_sleep_seconds)
+
+    raise GapScriptDemandError(
+        f"gap script demand judge failed after {config.llm_max_attempts} attempts: {last_error}"
+    ) from last_error
+
+
+def judge_remake_suitable_demands(
+    *,
+    demand_names: list[str],
+    config: GapScriptDemandConfig,
+) -> list[str]:
+    """按批次用 LLM 筛选适合画面改造的品类+解构词。"""
+    if not demand_names:
+        return []
+
+    matched: list[str] = []
+    seen: set[str] = set()
+    batches = _chunked(demand_names, config.judge_batch_size)
+    total_batches = len(batches)
+    for batch_index, batch in enumerate(batches, start=1):
+        print(
+            f"gap script demand: judging batch {batch_index}/{total_batches} "
+            f"size={len(batch)}",
+            flush=True,
+        )
+        batch_matched = _llm_judge_batch(demand_names=batch, config=config)
+        for demand_name in batch_matched:
+            if demand_name in seen:
+                continue
+            seen.add(demand_name)
+            matched.append(demand_name)
+    return matched

app/gap_script_demand/llm_json.py

@@ -0,0 +1,58 @@
+"""LLM 回复 JSON 解析工具。"""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+from app.gap_script_demand.exceptions import GapScriptDemandError
+
+
+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 GapScriptDemandError("llm output is not json object")

app/gap_script_demand/odps_demand_pool.py

@@ -0,0 +1,75 @@
+"""从 ODPS 需求池拉取当下供需 gap 数据。"""
+
+from __future__ import annotations
+
+import re
+
+from app.aliyun_odps.client import get_odps_client
+from app.gap_script_demand.exceptions import GapScriptDemandError
+
+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 GapScriptDemandError(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 GapScriptDemandError("partition_dt is required")
+    if not strategy_value:
+        raise GapScriptDemandError("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/gap_script_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.gap_script_demand.exceptions import GapScriptDemandError
+
+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 GapScriptDemandError(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_gap_script_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 GapScriptDemandError("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": table_name,
+        "partition_dt": partition_value,
+        "written_count": len(rows),
+    }

app/gap_script_demand/repository.py

@@ -0,0 +1,179 @@
+"""当下供需 gap 脚本主驱 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.gap_script_demand.demand_id import build_gap_script_demand_id
+from app.gap_script_demand.exceptions import GapScriptDemandError
+from app.hot_content.types import MysqlConfig
+
+TABLE_NAME = "gap_script_demand_pool_di"
+
+
+class GapScriptDemandRepository:
+    def __init__(self, config: MysqlConfig):
+        if pymysql is None or DictCursor is None:
+            raise GapScriptDemandError("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_gap_script_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)
+
+    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 GapScriptDemandError("gap script demand output strategy is required")
+        if not partition_value:
+            raise GapScriptDemandError("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_gap_script_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_matched_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/gap_script_demand/service.py

@@ -0,0 +1,216 @@
+"""当下供需 gap 脚本主驱每日任务入口。"""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+from typing import Any
+
+from app.gap_script_demand.config import load_gap_script_demand_config
+from app.gap_script_demand.judging import judge_remake_suitable_demands
+from app.gap_script_demand.odps_demand_pool import fetch_demand_names
+from app.gap_script_demand.odps_writer import build_odps_row, sync_gap_script_demands_to_odps
+from app.gap_script_demand.repository import GapScriptDemandRepository
+from app.gap_script_demand.types import GapScriptDemandConfig
+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_gap_script_demand_summary(summary: dict[str, Any]) -> None:
+    """将脚本主驱流程结果打印到控制台。"""
+    print("=" * 60)
+    print("当下供需gap-脚本主驱 任务结果")
+    print("=" * 60)
+    print(f"查询日期: {summary.get('query_date')}")
+    print(f"ODPS 分区: {summary.get('partition_dt')}")
+    print(f"源 strategy: {summary.get('demand_pool_strategy')}")
+    print(f"输出 strategy: {summary.get('output_strategy')}")
+    print(f"LLM 模型: {summary.get('llm_model')}")
+    print(f"批次大小: {summary.get('judge_batch_size')}")
+
+    if summary.get("skip_reason") in {
+        "no_demand_names",
+        "partition_data_exists",
+        "no_matched_demands",
+    }:
+        print(f"跳过原因: {summary.get('skip_reason')}")
+        if summary.get("skip_reason") == "no_matched_demands":
+            print(f"ODPS 需求总数: {summary.get('demand_name_total', 0)}")
+            print("匹配可改造数: 0")
+        print("=" * 60)
+        return
+
+    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}")
+
+    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_matched_demands(
+    *,
+    demand_names: list[str],
+    strategy: str,
+    partition_dt: str,
+    target_table: str,
+    sync_odps: bool = True,
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    flow_config = load_flow_config()
+    repository = GapScriptDemandRepository(flow_config.mysql)
+    try:
+        mysql_save = repository.persist_matched_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, {}
+
+        if not sync_odps:
+            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_gap_script_demands_to_odps(
+            rows=odps_rows,
+            partition_dt=partition_dt,
+            target_table=target_table,
+        )
+        return mysql_save, odps_sync
+    finally:
+        repository.close()
+
+
+def run_gap_script_demand_daily_job(
+    config: GapScriptDemandConfig,
+    *,
+    query_date: date | None = None,
+    sync_odps: bool = True,
+) -> dict[str, Any]:
+    """执行脚本主驱流程：拉取当天供需 gap → LLM 筛选可改造项 → 写 MySQL/ODPS。"""
+    target_date = query_date or _today_shanghai()
+    partition_dt = target_date.strftime("%Y%m%d")
+
+    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,
+        "llm_model": config.llm_model,
+        "judge_batch_size": config.judge_batch_size,
+        "demand_name_total": 0,
+        "matched_count": 0,
+        "matched_demand_names": [],
+        "mysql_save": {},
+        "odps_sync": {},
+    }
+
+    flow_config = load_flow_config()
+    repository = GapScriptDemandRepository(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": "gap_script_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"gap script demand: start judging "
+        f"demand_names={len(demand_names)} batch_size={config.judge_batch_size} "
+        f"model={config.llm_model}",
+        flush=True,
+    )
+    matched_demand_names = judge_remake_suitable_demands(
+        demand_names=demand_names,
+        config=config,
+    )
+    summary["matched_count"] = len(matched_demand_names)
+    summary["matched_demand_names"] = matched_demand_names
+
+    if not matched_demand_names:
+        summary["skip_reason"] = "no_matched_demands"
+        return summary
+
+    print("gap script demand: persisting matched demands", flush=True)
+    mysql_save, odps_sync = _persist_matched_demands(
+        demand_names=matched_demand_names,
+        strategy=config.output_strategy,
+        partition_dt=partition_dt,
+        target_table=config.demand_pool_source_table,
+        sync_odps=sync_odps,
+    )
+    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_gap_script_demand_for_date(
+    query_date: date,
+    *,
+    sync_odps: bool = True,
+) -> dict[str, Any]:
+    """测试指定日期的脚本主驱流程。"""
+    config = load_gap_script_demand_config()
+    return run_gap_script_demand_daily_job(
+        config,
+        query_date=query_date,
+        sync_odps=sync_odps,
+    )

app/gap_script_demand/types.py

@@ -0,0 +1,20 @@
+"""当下供需 gap 脚本主驱流程类型定义。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class GapScriptDemandConfig:
+    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
+    judge_batch_size: int

app/scheduler.py

@@ -26,6 +26,12 @@ from app.hot_content.wxindex_heat_pattern import run_wxindex_heat_pattern_daily_
 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
+from app.gap_script_demand.config import load_gap_script_demand_config
+from app.gap_script_demand.service import (
+    print_gap_script_demand_summary,
+    run_gap_script_demand_daily_job,
+)
+from app.gap_script_demand.types import GapScriptDemandConfig
 
 
 def _import_blocking_scheduler() -> Any:
@@ -133,6 +139,21 @@ def run_festival_demand_job(config: FestivalDemandConfig) -> None:
         print(f"festival demand failed: {exc}", file=sys.stderr)
 
 
+def run_gap_script_demand_job(config: GapScriptDemandConfig) -> None:
+    try:
+        summary = run_gap_script_demand_daily_job(config)
+        print_gap_script_demand_summary(summary)
+        print(
+            json.dumps(
+                {"job": "gap_script_demand", "summary": summary},
+                ensure_ascii=False,
+                indent=2,
+            )
+        )
+    except Exception as exc:
+        print(f"gap script demand failed: {exc}", file=sys.stderr)
+
+
 def register_hot_content_job(scheduler: Any, config: FlowConfig) -> None:
     scheduler.add_job(
         run_hot_content_job,
@@ -213,26 +234,45 @@ def register_festival_demand_job(scheduler: Any, config: FestivalDemandConfig) -
     )
 
 
+def register_gap_script_demand_job(scheduler: Any, config: GapScriptDemandConfig) -> None:
+    scheduler.add_job(
+        run_gap_script_demand_job,
+        trigger="cron",
+        hour=config.cron_hours,
+        minute=config.cron_minute,
+        timezone=SHANGHAI_TZ,
+        args=[config],
+        id="gap_script_demand",
+        name="当下供需gap：脚本主驱画面改造筛选",
+        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()
+    gap_script_config = load_gap_script_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)
+    register_gap_script_demand_job(scheduler, gap_script_config)
     print(
         "scheduler started, timezone=Asia/Shanghai, "
         "jobs=['hot_content_flow', 'decode_result_flow', 'wxindex_words_refresh', "
-        "'wxindex_heat_pattern', 'festival_demand'], "
+        "'wxindex_heat_pattern', 'festival_demand', 'gap_script_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"festival_demand_cron={festival_config.cron_hours}:{festival_config.cron_minute:02d}"
+        f"festival_demand_cron={festival_config.cron_hours}:{festival_config.cron_minute:02d}, "
+        f"gap_script_demand_cron={gap_script_config.cron_hours}:{gap_script_config.cron_minute:02d}"
     )
     scheduler.start()
 
@@ -250,6 +290,7 @@ def parse_args() -> argparse.Namespace:
             "wxindex-refresh",
             "wxindex-heat-pattern",
             "festival-demand",
+            "gap-script-demand",
         ),
         default="all",
         help="--once 时选择执行哪个任务",
@@ -303,6 +344,9 @@ def main() -> None:
         if args.job in {"festival-demand"}:
             festival_config = load_festival_demand_config()
             run_festival_demand_job(festival_config)
+        if args.job in {"gap-script-demand"}:
+            gap_script_config = load_gap_script_demand_config()
+            run_gap_script_demand_job(gap_script_config)
         return
     start_scheduler()
 

docker-compose.yml

@@ -36,6 +36,12 @@ services:
       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}
+      GAP_SCRIPT_DEMAND_CRON_HOURS: ${GAP_SCRIPT_DEMAND_CRON_HOURS:-10}
+      GAP_SCRIPT_DEMAND_CRON_MINUTE: ${GAP_SCRIPT_DEMAND_CRON_MINUTE:-0}
+      GAP_SCRIPT_DEMAND_LLM_MODEL: ${GAP_SCRIPT_DEMAND_LLM_MODEL:-openai/gpt-4o}
+      GAP_SCRIPT_DEMAND_POOL_STRATEGY: ${GAP_SCRIPT_DEMAND_POOL_STRATEGY:-当下供需gap}
+      GAP_SCRIPT_DEMAND_OUTPUT_STRATEGY: ${GAP_SCRIPT_DEMAND_OUTPUT_STRATEGY:-当下供需gap-脚本主驱}
+      GAP_SCRIPT_DEMAND_JUDGE_BATCH_SIZE: ${GAP_SCRIPT_DEMAND_JUDGE_BATCH_SIZE:-40}
       # 业务阈值
       WXINDEX_SCORE_THRESHOLD: ${WXINDEX_SCORE_THRESHOLD:-1000000}
       DEMAND_POOL_SOURCE_TABLE: ${DEMAND_POOL_SOURCE_TABLE:-dwd_multi_demand_pool_di}