Create-knowledge-find-decode/acquisition/query.py

"""生成 query 的 demo 引擎（只到「生成 query」为止：不搜索、不解构）。

① 实质×创作阶段×需求点(LLM)  ② 形式×载体位置(LLM)  ③ 搜索词扩展(待接入真 sug)
④ 多轴正交组合(机械)：实质×形式×阶段×动作×作用×知识类型
取自 scope_trees 节点 + 人工定义轴；设计见 开发文档/query构造.md。
"""
from __future__ import annotations

import json
from pathlib import Path
from typing import Optional

from acquisition.suggest import suggest
from core.config import Settings
from core.llm import chat_json
from core.prompts import load_prompt

ROOT = Path(__file__).resolve().parent.parent
TREES = ROOT / "scope_trees" / "trees_index.json"

# ② 载体位置：载体 × 位置 的交叉（短视频/图片有封面，文章无封面）；剧本/小说/长文属于实质，不在此
CARRIERS = ["短视频", "图片", "文章"]
POSITIONS = ["开头", "中间", "收尾", "封面"]
CARRIER_POS = [f"{c}{p}" for c in CARRIERS for p in POSITIONS if not (c == "文章" and p == "封面")]
# 分组形式（给前端「查看全部」展示 载体 × 位置 的交叉）
CARRIER_POS_GROUPED = {c: [p for p in POSITIONS if not (c == "文章" and p == "封面")] for c in CARRIERS}
# ④ 需求点（人工拟定）：每个创作阶段下的细分需求，喂给 LLM 从中选 + 前端按钮展示
DEMAND = {
    "灵感": ["找方向", "找素材", "拆案例"],
    "选题": ["选题", "爆款选题", "什么内容火"],
    "脚本": ["开头钩子", "结构", "标题", "文案"],
}
# 制作屏蔽词：④ 过滤 + 前端标记（创作 vs 制作边界）
BLOCK = ["剪辑", "调色", "参数", "导出", "软件", "生成", "插件", "渲染", "压制"]

# ④ 多轴正交「组合」query（机械拼接）的人工定义轴
ACTIONS = ["构思", "策划", "组织", "撰写", "改编", "润色"]            # 动作（待修改）
STAGES = ["灵感", "选题", "脚本"]                                    # 阶段
KTYPE_SUFFIX = {"what": "有哪些", "why": "为什么", "how": "怎么做"}   # 知识类型→句尾后缀
MODALITIES = ["图文", "视频"]                                       # 知识模态（搜索筛选维度，不进 query 串）


def sample_nodes(source_type: str, depths=(3, 4), limit: int = 20,
                 under: Optional[str] = None) -> list[str]:
    """读 trees_index.json，取某棵树的中层节点名（按出现序去重、封顶）。
    under 给定时只取 path 含该分支的节点（如实质取 实质树/理念、手法取 形式树/架构）。"""
    idx = json.loads(TREES.read_text("utf-8"))
    out: list[str] = []
    seen: set[str] = set()
    for n in idx:
        if n.get("source_type") != source_type:
            continue
        path = [x for x in (n.get("path") or "").split("/") if x]
        if under and under not in path:
            continue
        if len(path) in depths:
            name = n.get("name") or (path[-1] if path else "")
            if name and name not in seen:
                seen.add(name)
                out.append(name)
                if len(out) >= limit:
                    break
    return out


def tactic2_form_llm(form_nodes: list[str], settings: Settings) -> list[dict]:
    """② 形式树 × 载体位置 → LLM 正交生成自然 query（LLM 自行把书面形式标签理解成创作手法）。
    返回 [{形式, 载体位置, query}]，供前端表格从左到右展示正交。1 次批量调用。"""
    user = json.dumps({"形式树": form_nodes, "载体位置": CARRIER_POS, "屏蔽制作词": BLOCK}, ensure_ascii=False)
    try:
        res = chat_json(load_prompt("form_query_gen"), user, settings=settings, timeout=120)
        rows = res.get("rows") or []
    except Exception:
        rows = []
    return [{"形式": r.get("形式", ""), "载体位置": r.get("载体位置", ""), "query": r.get("query", "")}
            for r in rows if isinstance(r, dict) and r.get("query")]


def tactic3_suggest(seeds: list[dict], settings: Settings) -> list[dict]:
    """③ 搜索词扩展（仅小红书）：每个种子 → keyword_v2 → 从相关帖挖候选搜索词。
    seeds=[{query, 来源}]；来源标明种子出处（实质+意图 / 形式+需求词）。"""
    from acquisition.crawler import RateLimiter
    rl = RateLimiter(min_interval_seconds=1.0)
    out = []
    for s in seeds:
        q, origin = s["query"], s.get("来源", "")
        try:
            cands = suggest(q, settings=settings, rate_limiter=rl, limit=12)
        except Exception as exc:
            cands = [f"(失败: {str(exc)[:40]})"]
        out.append({"seed": q, "来源": origin, "候选": cands})
    return out


def tactic4_llm(topics: list[str], settings: Settings) -> list[dict]:
    """④ LLM 正交清洗：实质 × 创作阶段 × 需求点(人工拟定 DEMAND) → 自然 query（滤制作）。
    返回逐条带正交三轴的行：[{实质, 阶段, 需求点, query}]，供前端表格展示。1 次批量调用。"""
    user = json.dumps({"实质": topics, "需求点表": DEMAND, "屏蔽制作词": BLOCK}, ensure_ascii=False)
    try:
        res = chat_json(load_prompt("query_gen"), user, settings=settings, timeout=120)
        rows = res.get("rows") or []
    except Exception:
        rows = []
    return [{"实质": r.get("实质", ""), "阶段": r.get("阶段", ""),
             "需求点": r.get("需求点", ""), "query": r.get("query", "")}
            for r in rows if isinstance(r, dict) and r.get("query")]


def _nonleaf_d4(source_type: str, limit: int, under: Optional[str] = None) -> list[str]:
    """取某棵树的【4级非叶子节点】名（实质 79 / 形式 47 / 作用 16…），按序采样封顶。
    under 给定时只取该分支（如形式限 架构，避开 呈现 里的剪辑/后期等制作节点）。"""
    idx = json.loads(TREES.read_text("utf-8"))
    paths = {(n.get("path") or "") for n in idx if n.get("source_type") == source_type
             and (not under or under in (n.get("path") or "").split("/"))}
    d4 = [p for p in paths if len([x for x in p.split("/") if x]) == 4]
    nonleaf = sorted(p for p in d4 if any(o != p and o.startswith(p + "/") for o in paths))
    return [p.split("/")[-1] for p in nonleaf][:limit]


def tactic_multiaxis(n: int = 36) -> list[dict]:
    """④ 多轴正交组合（机械拼接）：实质×形式×阶段×动作×作用×知识类型 → 拼成「组合 query」。
    实质/形式/作用 取自分类树（4级非叶子），阶段/动作/知识类型 人工定义；模态不进 query。
    组合空间 ~320 万，这里 round-robin 取不同轴做【采样】，避免爆炸。无 LLM、纯机械。"""
    sz = _nonleaf_d4("实质", 8)
    xs = _nonleaf_d4("形式", 6, under="架构")   # 限创作手法（架构），避开呈现里的制作节点
    zy = _nonleaf_d4("作用", 6)
    ktypes = list(KTYPE_SUFFIX.items())                       # [(what,有哪些),…]
    stage_act = [(s, a) for s in STAGES for a in ACTIONS] + [("", "")]  # +「无动作」变体
    rows = []
    for i in range(n):
        s_ = sz[i % len(sz)]
        f_ = xs[i % len(xs)]
        st, ac = stage_act[i % len(stage_act)]
        zy_ = zy[i % len(zy)]
        kt, suf = ktypes[i % len(ktypes)]
        seg = (st + ac) if ac else ""                         # 脚本撰写 / 空
        parts = [s_, f_] + ([seg] if seg else []) + [zy_, suf]
        rows.append({"实质": s_, "形式": f_, "阶段": st or "/", "动作": ac or "/",
                     "作用": zy_, "知识类型": kt, "query": " ".join(parts)})
    return rows