增加当下供需gap-早安策略，早中晚好按权重取top10写入。

Co-authored-by: Cursor <cursoragent@cursor.com>

agent/llm/qwen.py

@@ -52,14 +52,16 @@ def create_qwen_llm_call(
     ) -> Dict[str, Any]:
         
         try:
-            response = await client.chat.completions.create(
-                model=model,
-                messages=messages,
-                tools=tools,
-                temperature=temperature,
-                max_tokens=max_tokens,
-                **kwargs
-            )
+            create_kwargs: Dict[str, Any] = {
+                "model": model,
+                "messages": messages,
+                "temperature": temperature,
+                "max_tokens": max_tokens,
+                **kwargs,
+            }
+            if tools:
+                create_kwargs["tools"] = tools
+            response = await client.chat.completions.create(**create_kwargs)
 
             # 获取内容
             content = response.choices[0].message.content or ""

examples/demand/data_query_tools.py

@@ -53,6 +53,8 @@ def execute_odps_sql(sql) -> bool:
 
 _STRATEGY_GAP = "当下供需gap"
 _STRATEGY_GAP_FENCI = "当下供需gap-分词"
+_STRATEGY_GAP_ZAOAN = "当下供需gap-早安"
+_ZAOAN_BARE_DEMAND_NAMES = {"早安", "晨安", "早上好", "清晨", "晨起"}
 _HIVE_TABLE = "loghubods.dwd_multi_demand_pool_di"
 _HIVE_DT_FMT = "%Y%m%d"  # 分区格式：yyyymmdd，如 20260519
 _CHINA_TZ = ZoneInfo("Asia/Shanghai")
@@ -282,6 +284,58 @@ def write_dwd_multi_demand_pool_di_to_hive(rows: list[dict]) -> int:
     return len(gap_parts) + len(fenci_parts)
 
 
+def write_dwd_zaoan_demand_pool_to_hive(rows: list[dict]) -> int:
+    """
+    写入策略「当下供需gap-早安」到 loghubods.dwd_multi_demand_pool_di。
+    固定早安词（早安/晨安/早上好/清晨/晨起）demand_name 只用词本身，不拼品类。
+    其余 demand_name = merge_leve2 + ' ' + name。
+    demand_id = md5(strategy + demand_name + type + dt)
+    该策略已由上游筛过，不再套用过滤词 / 昨天分词去重。
+    """
+    if not rows:
+        return 0
+
+    china_today = _hive_partition_dt()
+    select_parts: list[str] = []
+
+    for row in rows:
+        merge_leve2 = str(row.get("merge_leve2") or "").strip()
+        name = str(row.get("name") or "").strip()
+        if not merge_leve2 or not name:
+            continue
+
+        weight = round(float(row.get("score") or 0.0), 6)
+        ext_data = _parse_ext_data(row.get("ext_data"))
+        type_str = str(ext_data.get("type") or "").strip()
+        video_ids = ext_data.get("video_ids") or []
+        if not isinstance(video_ids, list):
+            video_ids = []
+        video_ids = [str(v).strip() for v in video_ids if v is not None and str(v).strip()]
+        video_count = len(video_ids)
+        extend_json = json.dumps({"品类": merge_leve2}, ensure_ascii=False)
+
+        demand_name = name if name in _ZAOAN_BARE_DEMAND_NAMES else f"{merge_leve2} {name}"
+        demand_id = hashlib.md5(
+            f"{_STRATEGY_GAP_ZAOAN}{demand_name}{type_str}{china_today}".encode("utf-8")
+        ).hexdigest()
+        select_parts.append(
+            _build_hive_select_part(
+                _STRATEGY_GAP_ZAOAN, demand_id, demand_name,
+                weight, type_str, video_count, video_ids, extend_json,
+            )
+        )
+
+    if not select_parts:
+        print(f"[hive] {_STRATEGY_GAP_ZAOAN} 无有效行，跳过")
+        return 0
+
+    print(f"[hive] {_STRATEGY_GAP_ZAOAN} 写入: dt={china_today}, rows={len(select_parts)}")
+    ok = _insert_hive_select_parts(select_parts, china_today)
+    if not ok:
+        raise RuntimeError(f"写入 Hive 失败: strategy={_STRATEGY_GAP_ZAOAN}, dt={china_today}")
+    return len(select_parts)
+
+
 def write_feature_point_data_to_hive(names: list[str]) -> int:
     """
     将需求名称写入 Hive 表 feature_point_data（按北京时间当天分区）。

examples/demand/run.py

@@ -239,10 +239,22 @@ def _finish_demand_task(task_id: Optional[int], status: int, task_log: str) -> N
         log(f"[task] 更新 demand_task 失败，task_id={task_id}, status={status}, error={e}")
 
 
-def write_demand_items_to_mysql(execution_id: int, merge_level2: str) -> int:
+def write_demand_items_local(execution_id: int, merge_level2: str) -> int:
     """
     把 result/{execution_id}/execution_id_{execution_id}_demand_items.json
-    写入 MySQL 表 demand_content
+    写成本地 JSON，不写入 MySQL / Hive。
+    """
+    return write_demand_items_to_mysql(
+        execution_id=execution_id,
+        merge_level2=merge_level2,
+        write_to_db=False,
+    )
+
+
+def write_demand_items_to_mysql(execution_id: int, merge_level2: str, write_to_db: bool = True) -> int:
+    """
+    把 result/{execution_id}/execution_id_{execution_id}_demand_items.json
+    写入 MySQL 表 demand_content。write_to_db=False 时只写本地文件。
     """
     # create_demand_item(s) 使用 Path.cwd()/result 作为输出目录。
     # 为了兼容“从不同目录启动脚本”的情况，这里同时尝试 cwd 和脚本目录两种结果位置。
@@ -317,6 +329,16 @@ def write_demand_items_to_mysql(execution_id: int, merge_level2: str) -> int:
         log("[mysql] 生成行为空，跳过写入")
         return 0
 
+    local_path = Path(__file__).parent / "result" / f"{merge_level2}.json"
+    local_path.parent.mkdir(parents=True, exist_ok=True)
+    with open(local_path, "w", encoding="utf-8") as f:
+        json.dump(rows, f, ensure_ascii=False, indent=4)
+    log(f"[local] 已写入 {local_path}，rows={len(rows)}")
+
+    if not write_to_db:
+        log("[mysql] skip：按要求不入库")
+        return len(rows)
+
     affected = mysql_db.insert_many("demand_content", rows)
     log(f"[mysql] 写入 demand_content 完成，rows={len(rows)}, affected={affected}")
 
@@ -327,9 +349,6 @@ def write_demand_items_to_mysql(execution_id: int, merge_level2: str) -> int:
         log(f"[hive] 写入 dwd_multi_demand_pool_di 完成，rows={hive_written}, dt={dt_value}")
     except Exception as e:
         log(f"[hive] 写入 dwd_multi_demand_pool_di 异常（MySQL 已成功）：{e}")
-    # with open(f'/Users/shimeng/Desktop/py/Agent/examples/demand/result/{merge_level2}.json', 'w',
-    #           encoding='utf-8') as f:
-    #     json.dump(rows, f, ensure_ascii=False, indent=4)
     return len(rows)
 
 
@@ -393,7 +412,13 @@ def write_global_tree_demand_items_to_hive(execution_id: int, merge_level2: str)
         return 0
 
 
-async def run_once(execution_id, merge_level2, count: int = 30, task_id: Optional[int] = None) -> str:
+async def run_once(
+        execution_id,
+        merge_level2,
+        count: int = 30,
+        task_id: Optional[int] = None,
+        write_to_db: bool = True,
+) -> str:
     task_log_text = ""
     task_status = 0
 
@@ -413,6 +438,8 @@ async def run_once(execution_id, merge_level2, count: int = 30, task_id: Optiona
     run_config = copy.deepcopy(RUN_CONFIG)
     run_config.temperature = float(prompt.config.get("temperature", run_config.temperature))
     run_config.max_iterations = int(prompt.config.get("max_iterations", run_config.max_iterations))
+    if count and int(count) > 50:
+        run_config.max_iterations = max(run_config.max_iterations, 400)
     run_config.tools = ENABLED_TOOLS.copy()
     # 禁用反思/总结经验相关流程（避免进入 reflection 侧分支）
     run_config.enable_research_flow = False
@@ -465,9 +492,11 @@ async def run_once(execution_id, merge_level2, count: int = 30, task_id: Optiona
 
             log(f"[cost] total_tokens={total_tokens}, total_cost=${total_cost:.6f}")
 
-            # agent 执行完成后：全局树写 Hive，其他写 MySQL
+            # agent 执行完成后：全局树写 Hive，其他写 MySQL；write_to_db=False 时只落本地
             try:
-                if str(merge_level2).strip() == "全局树":
+                if not write_to_db:
+                    write_demand_items_local(execution_id=execution_id, merge_level2=merge_level2)
+                elif str(merge_level2).strip() == "全局树":
                     write_global_tree_demand_items_to_hive(execution_id=execution_id, merge_level2=merge_level2)
                 else:
                     # element_names -> name（逗号分隔）；reason -> demand_content.reason；desc -> demand_content.suggestion；dt -> demand_content.dt
@@ -509,6 +538,7 @@ async def main(
         count,
         execution_id: Optional[int] = None,
         task_id: Optional[int] = None,
+        write_to_db: bool = True,
 ) -> dict:
     if execution_id is None:
         if platform_type == "piaoquan":
@@ -522,7 +552,13 @@ async def main(
     if not execution_id:
         return {"execution_id": None, "final_text": ""}
 
-    final_text = await run_once(execution_id, cluster_name, count=count, task_id=task_id)
+    final_text = await run_once(
+        execution_id,
+        cluster_name,
+        count=count,
+        task_id=task_id,
+        write_to_db=write_to_db,
+    )
     return {"execution_id": execution_id, "final_text": final_text}
 
 

examples/demand/run_local.py

@@ -0,0 +1,70 @@
+"""本地跑需求生成：只写文件，不入库。
+
+在项目根目录执行：
+
+    .venv/bin/python examples/demand/run_local.py
+
+默认：品类「早中晚好」，生成约 100 条。结果文件：
+
+    examples/demand/result/早中晚好.json
+    examples/demand/result/<execution_id>/execution_id_<execution_id>_demand_items.json
+
+可选参数：
+
+    .venv/bin/python examples/demand/run_local.py --cluster 早中晚好 --count 100
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+DEMAND_DIR = Path(__file__).resolve().parent
+sys.path.insert(0, str(ROOT))
+os.chdir(DEMAND_DIR)
+
+from examples.demand.run import main as run_demand
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="本地生成需求并写入 JSON，不入库")
+    parser.add_argument("--cluster", default="早中晚好", help="二级品类名")
+    parser.add_argument("--platform", default="piaoquan", choices=["piaoquan", "changwen", "zengzhang"])
+    parser.add_argument("--count", type=int, default=100, help="目标需求数量")
+    return parser.parse_args()
+
+
+async def _run() -> None:
+    args = parse_args()
+    print(
+        f"[local] cluster={args.cluster} platform={args.platform} "
+        f"count={args.count} write_to_db=False",
+        flush=True,
+    )
+    result = await run_demand(
+        args.cluster,
+        args.platform,
+        args.count,
+        write_to_db=False,
+    )
+    execution_id = result.get("execution_id")
+    named_path = DEMAND_DIR / "result" / f"{args.cluster}.json"
+    items_path = (
+        DEMAND_DIR
+        / "result"
+        / str(execution_id)
+        / f"execution_id_{execution_id}_demand_items.json"
+    )
+    print(f"[local] execution_id={execution_id}", flush=True)
+    print(f"[local] 需求文件: {named_path}", flush=True)
+    print(f"[local] Agent 原始结果: {items_path}", flush=True)
+    if not execution_id:
+        raise SystemExit("执行失败：未拿到 execution_id（请检查品类数据和数据库连接）")
+
+
+if __name__ == "__main__":
+    asyncio.run(_run())

examples/demand/web_api.py

@@ -24,6 +24,12 @@ from examples.demand.changwen_prepare import changwen_prepare
 from examples.demand.mysql import mysql_db
 from examples.demand.piaoquan_prepare import piaoquan_prepare
 from examples.demand.run import _create_demand_task, main as run_demand
+from examples.demand.zaozhongwanhao_extra import (
+    CLUSTER_NAME as ZAOZHONGWANHAO_CLUSTER,
+    maybe_finish_zaoan_from_cluster,
+    maybe_run_zaozhongwanhao_extra,
+    schedule_has_zaozhongwanhao,
+)
 
 app = FastAPI(title="demand web api")
 
@@ -170,13 +176,17 @@ async def demand_scheduled_run_once() -> None:
     任务批处理（串行）：
     遍历配置列表 -> 查当天 demand_task -> 匹配 cluster_name/name & platform_type/platform
     若存在 status=0 或 1 的记录则跳过；否则执行一次 demand_start_sync。
+    「早中晚好」：有品类产出则按权重取 top10；没有则只生成 10 条。再筛早安策略入库。
     """
     # ODPS 查询是阻塞调用，放到线程里避免阻塞事件循环
     demand_schedule_cluster_platform_list = await asyncio.to_thread(_get_demand_schedule_cluster_platform_list)
-    if not demand_schedule_cluster_platform_list:
-        return
+    demand_schedule_cluster_platform_list = demand_schedule_cluster_platform_list or []
 
     now = datetime.now(BEIJING_TZ)
+    has_zao = schedule_has_zaozhongwanhao(demand_schedule_cluster_platform_list)
+    await maybe_run_zaozhongwanhao_extra(demand_schedule_cluster_platform_list, now=now)
+
+    zao_execution_id = None
     for item in demand_schedule_cluster_platform_list:
         cluster_name = item.get("cluster_name")
         platform_type = item.get("platform_type")
@@ -189,7 +199,12 @@ async def demand_scheduled_run_once() -> None:
             continue
 
         print(f"[scheduler] run: cluster={cluster_name}, platform={platform_type}, count={count}")
-        await demand_start_sync(cluster_name=cluster_name, platform_type=platform_type, count=count)  # 串行执行
+        result = await demand_start_sync(cluster_name=cluster_name, platform_type=platform_type, count=count)  # 串行执行
+        if str(cluster_name).strip() == ZAOZHONGWANHAO_CLUSTER and isinstance(result, dict):
+            zao_execution_id = result.get("execution_id") or zao_execution_id
+
+    if has_zao:
+        await maybe_finish_zaoan_from_cluster(execution_id=zao_execution_id, now=now)
 
 
 _demand_scheduler: Optional[Any] = None

examples/demand/zaozhongwanhao_extra.py

@@ -0,0 +1,485 @@
+"""早中晚好额外需求：筛选保留项 + 固定早安词，写入策略「当下供需gap-早安」。"""
+
+from __future__ import annotations
+
+import json
+import re
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Optional
+from zoneinfo import ZoneInfo
+
+from examples.demand.db_manager import exist_cluster_tree
+
+CLUSTER_NAME = "早中晚好"
+PLATFORM_TYPE = "piaoquan"
+EXTRA_COUNT = 10
+TOP_N = 10
+QWEN_FILTER_MODEL = "qwen3.6-plus"
+OPENROUTER_QWEN_FILTER_MODEL = "qwen/qwen3.6-plus"
+STRATEGY_ZAOAN = "当下供需gap-早安"
+FIXED_ZAOAN_DEMANDS = ["早安", "晨安", "早上好", "清晨", "晨起"]
+BEIJING_TZ = ZoneInfo("Asia/Shanghai")
+
+FILTER_PROMPT = """你是一个专业的视频内容解构词筛选器。你的任务是对我提供的所有“解构词”进行逐一审核，剔除那些与 “早安祝福” 主题无关或关联性过弱的词，保留能够有效体现、联想或支撑该主题的词。
+筛选标准（必须同时满足以下两点，方可保留）： 主题相关性：该解构词必须能让人直接或明显联想到“早安祝福”的场景、情感、对象或常见元素（如：晨光、问候、美好祝愿、积极心态、崭新一天、健康平安、亲友关怀等）。
+意义有效性：即使该词在品类上看似合理，但如果放入早安祝福语境下无法产生具体、积极、正向的意义，或显得生硬、无关、空洞，则应判定为“不匹配”，予以剔除。
+判定示例（仅供理解，不限于此）： 保留：阳光、微笑、晨风、好运、平安、咖啡香、鸟鸣、崭新、希望、温暖、感恩、加油、活力、朝霞、问候语……
+剔除：暴雨、深夜、疲惫、焦虑、折扣、促销、施工、故障、诉讼、账单、统计表……（即使某些词属于常见品类，但无法正向关联早安祝福）
+输出格式要求： 只输出 保留名单，按行列出每个解构词（可保留原始“品类 解构词”格式）。
+不输出剔除名单，不输出解释说明，不添加额外评语。
+待筛选的解构词列表
+"""
+
+_BULLET_RE = re.compile(r"^[\s\-*\d.、）)]+")
+
+
+def schedule_has_zaozhongwanhao(schedule_items: list[dict] | None) -> bool:
+    for item in schedule_items or []:
+        if str(item.get("cluster_name") or "").strip() != CLUSTER_NAME:
+            continue
+        if str(item.get("platform_type") or "").strip() != PLATFORM_TYPE:
+            continue
+        return True
+    return False
+
+
+def extra_output_dir(now: Optional[datetime] = None) -> Path:
+    current = now or datetime.now(BEIJING_TZ)
+    dt = current.astimezone(BEIJING_TZ).strftime("%Y%m%d")
+    return Path(__file__).parent / "result" / f"{CLUSTER_NAME}_extra" / dt
+
+
+def extra_already_done_today(now: Optional[datetime] = None) -> bool:
+    return (extra_output_dir(now) / "zaoan_written.txt").exists()
+
+
+def _join_element_names(element_names: object) -> str:
+    if element_names is None:
+        return ""
+    if isinstance(element_names, list):
+        return " ".join(str(x).strip() for x in element_names if x is not None and str(x).strip())
+    return str(element_names).strip()
+
+
+def _decomp_term(item: dict) -> str:
+    joined = _join_element_names(item.get("element_names"))
+    if not joined:
+        return ""
+    return f"{CLUSTER_NAME} {joined}"
+
+
+def _find_demand_items_path(execution_id: int) -> Optional[Path]:
+    name = f"execution_id_{execution_id}_demand_items.json"
+    candidates = [
+        Path.cwd() / "result" / str(execution_id) / name,
+        Path(__file__).parent / "result" / str(execution_id) / name,
+    ]
+    for path in candidates:
+        if path.exists():
+            return path
+    return None
+
+
+def _load_demand_items(execution_id: int) -> list[dict]:
+    path = _find_demand_items_path(execution_id)
+    if not path:
+        return []
+    with open(path, "r", encoding="utf-8") as f:
+        loaded = json.load(f)
+    items = loaded["items"] if isinstance(loaded, dict) and isinstance(loaded.get("items"), list) else loaded
+    if not isinstance(items, list):
+        return []
+    return [item for item in items if isinstance(item, dict)]
+
+
+def _item_weight(item: dict, score_map: dict) -> float:
+    from examples.demand.run import _avg_score_for_joined_name, _join_element_names_to_name
+
+    name = _join_element_names_to_name(item.get("element_names")) or _join_element_names(item.get("element_names"))
+    if not name:
+        return 0.0
+    comma_score = _avg_score_for_joined_name(name, score_map)
+    space_score = _avg_score_for_joined_name(name.replace(" ", ","), score_map)
+    return max(comma_score, space_score)
+
+
+def select_top_n_by_weight(items: list[dict], execution_id: Optional[int] = None, n: int = TOP_N) -> list[dict]:
+    if not items:
+        return []
+    from examples.demand.run import _load_name_score_map
+
+    score_map = _load_name_score_map(execution_id) if execution_id else {}
+    ranked = sorted(items, key=lambda item: _item_weight(item, score_map), reverse=True)
+    return ranked[: max(int(n), 0)]
+
+
+def _normalize_kept_line(line: str) -> str:
+    text = (line or "").strip()
+    if not text:
+        return ""
+    text = text.strip("`\"'“”‘’")
+    text = _BULLET_RE.sub("", text).strip()
+    return text
+
+
+def _parse_kept_lines(content: str) -> list[str]:
+    kept: list[str] = []
+    seen: set[str] = set()
+    for raw in (content or "").splitlines():
+        line = _normalize_kept_line(raw)
+        if not line:
+            continue
+        if line in seen:
+            continue
+        seen.add(line)
+        kept.append(line)
+    return kept
+
+
+def _item_matches_kept(item: dict, kept_lines: list[str]) -> bool:
+    term = _decomp_term(item)
+    joined = _join_element_names(item.get("element_names"))
+    for line in kept_lines:
+        if line == term or (joined and line == joined):
+            return True
+        if joined and (line.endswith(f" {joined}") or line.endswith(f"\t{joined}")):
+            return True
+    return False
+
+
+async def _filter_with_qwen(terms: list[str]) -> tuple[str, list[str]]:
+    import os
+
+    messages = [
+        {
+            "role": "user",
+            "content": FILTER_PROMPT + "\n".join(terms),
+        }
+    ]
+    content = ""
+    if os.getenv("QWEN_API_KEY"):
+        from agent.llm.qwen import create_qwen_llm_call
+
+        llm_call = create_qwen_llm_call(model=QWEN_FILTER_MODEL)
+        try:
+            result = await llm_call(
+                messages=messages,
+                model=QWEN_FILTER_MODEL,
+                temperature=0.1,
+                extra_body={"enable_thinking": False},
+            )
+        except Exception:
+            result = await llm_call(
+                messages=messages,
+                model=QWEN_FILTER_MODEL,
+                temperature=0.1,
+            )
+        content = str((result or {}).get("content") or "").strip()
+    else:
+        from agent.llm.openrouter import create_openrouter_llm_call
+
+        print(f"[scheduler-extra] 未配置 QWEN_API_KEY，改用 OpenRouter {OPENROUTER_QWEN_FILTER_MODEL}")
+        llm_call = create_openrouter_llm_call(model=OPENROUTER_QWEN_FILTER_MODEL)
+        result = await llm_call(
+            messages=messages,
+            model=OPENROUTER_QWEN_FILTER_MODEL,
+            temperature=0.1,
+        )
+        content = str((result or {}).get("content") or "").strip()
+    return content, _parse_kept_lines(content)
+
+
+def _write_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(payload, f, ensure_ascii=False, indent=2)
+
+
+def _merge_kept_and_fixed(kept_items: list[dict]) -> list[dict]:
+    merged: list[dict] = []
+    seen: set[str] = set()
+    for item in kept_items or []:
+        name = _join_element_names(item.get("element_names"))
+        if not name or name in seen:
+            continue
+        seen.add(name)
+        merged.append(item)
+    for name in FIXED_ZAOAN_DEMANDS:
+        if name in seen:
+            continue
+        seen.add(name)
+        merged.append(
+            {
+                "element_names": [name],
+                "reason": "固定早安祝福需求",
+                "desc": f"用户需要「{name}」相关的早安祝福内容",
+                "type": "元素",
+            }
+        )
+    return merged
+
+
+def _build_zaoan_rows(items: list[dict], execution_id: Optional[int] = None) -> list[dict]:
+    from examples.demand.run import (
+        _avg_score_for_joined_name,
+        _join_element_names_to_name,
+        _load_name_score_map,
+        _resolve_video_ids_by_name_and_execution_id,
+        _safe_truncate,
+    )
+
+    dt_value = datetime.now(BEIJING_TZ).strftime("%Y%m%d")
+    score_map = _load_name_score_map(execution_id) if execution_id else {}
+    rows: list[dict] = []
+    for item in items:
+        name = _join_element_names_to_name(item.get("element_names"))
+        if not name:
+            name = _join_element_names(item.get("element_names"))
+        if not name:
+            continue
+        name = name.replace(",", " ")
+        type_str = str(item.get("type") or "元素").strip() or "元素"
+        reason = item.get("reason") or "固定早安祝福需求"
+        desc_value = item.get("desc") or f"用户需要「{name}」相关的早安祝福内容"
+        score = _avg_score_for_joined_name(name.replace(" ", ","), score_map) if score_map else 0.0
+        video_ids: list[str] = []
+        if execution_id:
+            try:
+                video_ids = _resolve_video_ids_by_name_and_execution_id(
+                    name=name.replace(" ", ","),
+                    execution_id=execution_id,
+                )
+            except Exception as e:
+                print(f"[scheduler-extra] 解析 video_ids 失败 name={name}: {e}")
+        ext_data = {
+            "reason": reason,
+            "desc": desc_value,
+            "type": type_str,
+            "video_ids": video_ids,
+            "strategy": STRATEGY_ZAOAN,
+        }
+        rows.append(
+            {
+                "merge_leve2": CLUSTER_NAME[:32],
+                "name": _safe_truncate(name, 64),
+                "reason": reason,
+                "suggestion": desc_value,
+                "score": float(score),
+                "ext_data": json.dumps(ext_data, ensure_ascii=False),
+                "dt": dt_value,
+            }
+        )
+    return rows
+
+
+def persist_zaoan_strategy(
+        kept_items: list[dict],
+        execution_id: Optional[int] = None,
+        now: Optional[datetime] = None,
+) -> dict:
+    """保留需求 + 固定 5 词，写入 demand_content 和 Hive 策略「当下供需gap-早安」。"""
+    from examples.demand.data_query_tools import write_dwd_zaoan_demand_pool_to_hive
+    from examples.demand.mysql import mysql_db
+
+    current = now or datetime.now(BEIJING_TZ)
+    out_dir = extra_output_dir(current)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    merged_items = _merge_kept_and_fixed(kept_items)
+    rows = _build_zaoan_rows(merged_items, execution_id=execution_id)
+    _write_json(out_dir / "zaoan_demand_items.json", merged_items)
+    _write_json(out_dir / "zaoan_rows.json", rows)
+
+    if not rows:
+        message = "早安策略行为空，跳过入库"
+        print(f"[scheduler-extra] {message}")
+        return {"ok": False, "message": message, "mysql": 0, "hive": 0}
+
+    mysql_affected = mysql_db.insert_many("demand_content", rows)
+    print(f"[scheduler-extra] MySQL demand_content 写入 rows={len(rows)}, affected={mysql_affected}")
+
+    hive_written = write_dwd_zaoan_demand_pool_to_hive(rows)
+    print(f"[scheduler-extra] Hive {STRATEGY_ZAOAN} 写入 rows={hive_written}")
+
+    marker = (
+        f"dt={rows[0]['dt']}\n"
+        f"execution_id={execution_id}\n"
+        f"mysql={mysql_affected}\n"
+        f"hive={hive_written}\n"
+        f"names={','.join(r['name'] for r in rows)}\n"
+    )
+    (out_dir / "zaoan_written.txt").write_text(marker, encoding="utf-8")
+    return {
+        "ok": True,
+        "mysql": mysql_affected,
+        "hive": hive_written,
+        "count": len(rows),
+        "output_dir": str(out_dir),
+        "names": [r["name"] for r in rows],
+    }
+
+
+async def filter_and_persist_zaoan(
+        items: list[dict],
+        execution_id: Optional[int] = None,
+        now: Optional[datetime] = None,
+) -> dict:
+    """按权重取 top10，Qwen 筛选后加上固定早安词写入策略。"""
+    current = now or datetime.now(BEIJING_TZ)
+    out_dir = extra_output_dir(current)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    top_items = select_top_n_by_weight(items, execution_id=execution_id, n=TOP_N)
+    _write_json(out_dir / "demand_items.json", items)
+    _write_json(out_dir / "top10_demand_items.json", top_items)
+    terms = [term for term in (_decomp_term(item) for item in top_items) if term]
+    (out_dir / "terms.txt").write_text("\n".join(terms) + ("\n" if terms else ""), encoding="utf-8")
+    print(f"[scheduler-extra] 候选={len(items)} 按权重取 top{TOP_N}={len(top_items)}")
+
+    if not terms:
+        print("[scheduler-extra] top10 为空，仍写入固定早安需求")
+        (out_dir / "kept.txt").write_text("", encoding="utf-8")
+        _write_json(out_dir / "kept_demand_items.json", [])
+        persist = persist_zaoan_strategy([], execution_id=execution_id, now=current)
+        return {
+            "ok": persist.get("ok", False),
+            "execution_id": execution_id,
+            "raw_count": len(items),
+            "top_count": 0,
+            "kept_count": 0,
+            "persist": persist,
+        }
+
+    print(f"[scheduler-extra] 用 {QWEN_FILTER_MODEL} 筛选 {len(terms)} 个解构词")
+    try:
+        raw_content, kept_lines = await _filter_with_qwen(terms)
+    except Exception as e:
+        print(f"[scheduler-extra] Qwen 筛选失败: {e}")
+        (out_dir / "qwen_error.txt").write_text(str(e), encoding="utf-8")
+        raw_content, kept_lines = "", []
+    (out_dir / "qwen_raw.txt").write_text(raw_content + ("\n" if raw_content else ""), encoding="utf-8")
+    (out_dir / "kept.txt").write_text("\n".join(kept_lines) + ("\n" if kept_lines else ""), encoding="utf-8")
+
+    kept_items = [item for item in top_items if _item_matches_kept(item, kept_lines)]
+    _write_json(out_dir / "kept_demand_items.json", kept_items)
+    persist = persist_zaoan_strategy(kept_items, execution_id=execution_id, now=current)
+    print(
+        f"[scheduler-extra] 完成 execution_id={execution_id} "
+        f"raw={len(items)} top={len(top_items)} kept={len(kept_items)}"
+    )
+    return {
+        "ok": True,
+        "execution_id": execution_id,
+        "raw_count": len(items),
+        "top_count": len(top_items),
+        "kept_count": len(kept_items),
+        "output_dir": str(out_dir),
+        "persist": persist,
+    }
+
+
+async def run_zaozhongwanhao_extra(now: Optional[datetime] = None) -> dict:
+    """
+    当天没有「早中晚好」品类时：只生成 10 条（不进普通需求表），
+    再筛选并写入「当下供需gap-早安」。
+    """
+    current = now or datetime.now(BEIJING_TZ)
+    out_dir = extra_output_dir(current)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    if not exist_cluster_tree(CLUSTER_NAME):
+        message = "获取聚类树失败，跳过额外生成"
+        print(f"[scheduler-extra] {CLUSTER_NAME}: {message}")
+        (out_dir / "error.txt").write_text(message, encoding="utf-8")
+        return {"ok": False, "message": message}
+
+    from examples.demand.run import main as run_demand
+
+    print(f"[scheduler-extra] 当天无品类，只生成 {EXTRA_COUNT} 条需求（不入库）")
+    result = await run_demand(
+        CLUSTER_NAME,
+        PLATFORM_TYPE,
+        EXTRA_COUNT,
+        write_to_db=False,
+    )
+    execution_id = result.get("execution_id") if isinstance(result, dict) else None
+    if not execution_id:
+        message = "额外生成失败：未拿到 execution_id"
+        print(f"[scheduler-extra] {message}")
+        (out_dir / "error.txt").write_text(message, encoding="utf-8")
+        return {"ok": False, "message": message, "execution_id": None}
+
+    items = _load_demand_items(execution_id)
+    return await filter_and_persist_zaoan(items, execution_id=execution_id, now=current)
+
+
+async def run_zaoan_from_generated_demands(
+        execution_id: Optional[int] = None,
+        now: Optional[datetime] = None,
+) -> dict:
+    """当天品类已产生多条需求时：按权重取 top10，再筛选写入早安策略。"""
+    current = now or datetime.now(BEIJING_TZ)
+    if not execution_id:
+        from examples.demand.run import get_execution_id_by_merge_level2
+
+        execution_id = get_execution_id_by_merge_level2(CLUSTER_NAME)
+    if not execution_id:
+        message = "未找到当天「早中晚好」execution_id，无法按权重取 top10"
+        print(f"[scheduler-extra] {message}")
+        persist = persist_zaoan_strategy([], execution_id=None, now=current)
+        return {"ok": persist.get("ok", False), "message": message, "persist": persist}
+
+    items = _load_demand_items(execution_id)
+    print(f"[scheduler-extra] 品类已产出 {len(items)} 条，按权重取 top{TOP_N}")
+    return await filter_and_persist_zaoan(items, execution_id=execution_id, now=current)
+
+
+async def maybe_run_zaozhongwanhao_extra(schedule_items: list[dict] | None, now: Optional[datetime] = None) -> None:
+    """仅处理「当天列表没有早中晚好」：补生成 10 条。有品类时等正常任务跑完再取 top10。"""
+    current = now or datetime.now(BEIJING_TZ)
+    out_dir = extra_output_dir(current)
+    if extra_already_done_today(current):
+        print(f"[scheduler-extra] 当天早安策略已写入，跳过：{out_dir}")
+        return
+    if schedule_has_zaozhongwanhao(schedule_items):
+        print(f"[scheduler-extra] 当天已有「{CLUSTER_NAME}」，等正常任务产出后按权重取 top{TOP_N}")
+        return
+
+    kept_path = out_dir / "kept_demand_items.json"
+    if kept_path.exists():
+        print(f"[scheduler-extra] 复用已有筛选结果写入早安策略：{kept_path}")
+        try:
+            kept_items = json.loads(kept_path.read_text(encoding="utf-8"))
+            if not isinstance(kept_items, list):
+                kept_items = []
+            persist_zaoan_strategy(kept_items, now=current)
+        except Exception as e:
+            print(f"[scheduler-extra] 复用筛选结果写入失败: {e}")
+        return
+
+    print(f"[scheduler-extra] 当天没有「{CLUSTER_NAME}」，只产生 {EXTRA_COUNT} 条后写入 {STRATEGY_ZAOAN}")
+    try:
+        await run_zaozhongwanhao_extra(now=current)
+    except Exception as e:
+        print(f"[scheduler-extra] 额外生成失败: {e}")
+        out_dir.mkdir(parents=True, exist_ok=True)
+        (out_dir / "error.txt").write_text(str(e), encoding="utf-8")
+
+
+async def maybe_finish_zaoan_from_cluster(
+        execution_id: Optional[int] = None,
+        now: Optional[datetime] = None,
+) -> None:
+    """品类正常任务之后：按权重取 top10 写入早安策略。"""
+    current = now or datetime.now(BEIJING_TZ)
+    out_dir = extra_output_dir(current)
+    if extra_already_done_today(current):
+        print(f"[scheduler-extra] 当天早安策略已写入，跳过：{out_dir}")
+        return
+    try:
+        await run_zaoan_from_generated_demands(execution_id=execution_id, now=current)
+    except Exception as e:
+        print(f"[scheduler-extra] 按权重取 top{TOP_N} 失败: {e}")
+        out_dir.mkdir(parents=True, exist_ok=True)
+        (out_dir / "error.txt").write_text(str(e), encoding="utf-8")