skill-video-lapian/scripts/build_report.py

#!/usr/bin/env python3
"""build_report.py — 生成拉片 HTML 报告(暗色主题/手机自适应/相对路径).

用法:
  python3 build_report.py --workdir /tmp/lapian --title "拉片分析:听show丽江酒吧"
前置(可缺失, 缺了自动降级):
  meta.json / transcript.json / frames.json / comments.json
产物: workdir/report.html + workdir/analysis.json
特性:
  - 分段: 优先 LLM(编导方法论点评), 失败回退规则启发式, 永不中断
  - 词频/情绪: 纯本地 n-gram + 情绪词典(无第三方依赖)
  - 模板渲染: 内置 mini-Jinja(stdlib), 无需安装 jinja2
"""
import argparse
import datetime
import json
import os
import re
import sys
import urllib.request

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common import e as esc, fmt_ts  # noqa: E402

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TEMPLATE = os.path.join(ROOT, "templates", "report.html.j2")
TEXT_MODEL = os.environ.get("FEME_TEXT_MODEL", "deepseek-v4-flash")
API_BASE = os.environ.get("FEME_API_BASE", "https://api.fmode.cn/v1")

TAG_CLS = {"薛辉": "tag-blue", "郑经说": "tag-purple", "三把刀": "tag-green",
           "小希": "tag-yellow", "南门": "tag-yellow", "志楠": "tag-green"}
POS_KW = ("同感 真实 说得对 认同 共鸣 喜欢 爱 赞 通透 治愈 舒服 自由 幸福 羡慕 "
          "向往 懂生活 回家 回云南 想去 泪目 豁达 乐观").split()
NEG_KW = ("不认同 杠 胡说 假的 喷 讨厌 反对 无语 尬 难看 翻车 骗 贩卖焦虑 内卷 累 苦"
          ).split()
SENT_STOP = "的了是在和就有不人我都这也你们他们啊吧吗呢把被将还又再便而或如果所以因为但是"
LLM_PROMPT = """你是短视频编导顾问(方法体系: 薛辉·观点前置/排比/画面感/收束升华; 郑经说·人群分层/功利型内容; 三把刀·流量文案vs变现文案; 编导小希·前三秒画面钩子)。对下面这条口播视频逐段拉片。

视频数据: 时长{dur}s, 点赞{digg}, 评论{cmt}, 收藏{coll}, 分享{share}
文案全文: {fulltext}

逐段台词(带时间戳):
{segment_lines}

已抽帧画面参考:
{frame_lines}

严格输出 JSON(不要markdown围栏):
{{"segments":[{{"i":0,"func":"起/承/转/合/收 之一","techniques":["导师·技法名"],"comment":"60-90字点评: 句式/节奏/画面匹配/情绪价值, 点名用了什么技法","reuse":"可直接套用的句式模板, 含{{占位符}}"}}],
"suggestions":[{{"title":"建议标题","body":"40字内可执行动作"}}]}}
要求: segments 数量与输入逐段台词一一对应; techniques 每段1-3个, 用"导师·技法"格式; suggestions 3条。"""


# ----------------------------------------------------------- mini-Jinja ----
_TOKEN = re.compile(r"({{.*?}}|{%.*?%})", re.S)


def _lookup(val, key):
    if isinstance(val, dict):
        return val.get(key, "")
    if isinstance(val, (list, tuple)):
        if key == "length":
            return len(val)
        try:
            return val[int(key)]
        except (ValueError, IndexError):
            return ""
    return getattr(val, key, "")


def _resolve(expr, ctx):
    expr = expr.strip()
    if re.fullmatch(r"-?\d+(\.\d+)?", expr):
        return float(expr)
    if expr in ("true", "True"):
        return True
    if expr in ("false", "False", "none", "None"):
        return False
    parts = expr.split(".")
    val = ctx.get(parts[0], "")
    for p in parts[1:]:
        val = _lookup(val, p)
    return val


def _parse(tokens, i):
    nodes = []
    while i < len(tokens):
        t = tokens[i]
        if t.startswith("{%"):
            stmt = t[2:-2].strip()
            kw, _, rest = stmt.partition(" ")
            if kw in ("endif", "endfor", "else"):
                return nodes, i
            if kw == "if":
                branches, cond = [], rest.strip()
                while True:
                    body, i = _parse(tokens, i + 1)
                    branches.append((cond, body))
                    tk = tokens[i]
                    tstmt = tk[2:-2].strip()
                    tkw, _, trest = tstmt.partition(" ")
                    if tkw == "elif":
                        cond = trest.strip()
                        continue
                    if tkw == "else":
                        body, i = _parse(tokens, i + 1)
                        branches.append((None, body))
                    i += 1  # consume endif
                    break
                nodes.append(("if", branches))
            elif kw == "for":
                var, itexpr = rest.strip().split(" in ", 1)
                body, i = _parse(tokens, i + 1)
                i += 1  # consume endfor
                nodes.append(("for", var.strip(), itexpr.strip(), body))
            else:
                i += 1
        elif t.startswith("{{"):
            nodes.append(("expr", t[2:-2].strip()))
            i += 1
        else:
            nodes.append(("text", t))
            i += 1
    return nodes, i


def _exec(nodes, ctx, out):
    for node in nodes:
        if node[0] == "text":
            out.append(node[1])
        elif node[0] == "expr":
            v = _resolve(node[1], ctx)
            out.append("" if v is None else str(v))
        elif node[0] == "if":
            for cond, body in node[1]:
                if cond is None or _resolve(cond, ctx):
                    _exec(body, ctx, out)
                    break
        elif node[0] == "for":
            _, var, itexpr, body = node
            it = _resolve(itexpr, ctx)
            items = it if isinstance(it, (list, tuple)) else ([it] if it else [])
            n = len(items)
            for idx, item in enumerate(items, 1):
                sub = dict(ctx)
                sub[var] = item
                sub["loop"] = {"index": idx, "first": idx == 1, "last": idx == n}
                _exec(body, sub, out)


def render_template(template: str, ctx: dict) -> str:
    tokens = _TOKEN.split(template)
    nodes, _ = _parse(tokens, 0)
    out = []
    _exec(nodes, ctx, out)
    return "".join(out)


# ------------------------------------------------------------- analysis ----
def load_json(path):
    if os.path.exists(path):
        try:
            return json.load(open(path, encoding="utf-8"))
        except Exception:
            return None
    return None


def call_llm(prompt: str):
    payload = {"model": TEXT_MODEL, "temperature": 0.3, "max_tokens": 4000,
               "messages": [{"role": "user", "content": prompt}]}
    req = urllib.request.Request(
        f"{API_BASE}/chat/completions", data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json",
                 "Authorization": f"Bearer {os.environ.get('FEME_NEWAPI_TOKEN') or os.environ.get('FMODE_API_KEY','')}"})
    with urllib.request.urlopen(req, timeout=180) as r:
        resp = json.loads(r.read())
    return resp["choices"][0]["message"]["content"]


def parse_llm_json(text: str) -> dict:
    text = text.strip()
    if text.startswith("```"):
        text = text.strip("`").lstrip("json").strip()
    try:
        return json.loads(text)
    except Exception:
        pass
    if "{" in text and "}" in text:
        try:
            return json.loads(text[text.index("{"): text.rindex("}") + 1])
        except Exception:
            pass
    return {}


def tag_cls(label: str) -> str:
    for k, cls in TAG_CLS.items():
        if k in label:
            return cls
    return "tag-blue"


def heuristic_techniques(i: int, n: int, text: str) -> list:
    tags = []
    if i == 0:
        tags.append("薛辉·观点前置")
        tags.append("小希·前三秒钩子")
    if re.search(r"(既不|又不|也不|不急着|不是[^。，]*,?不是)", text):
        tags.append("薛辉·排比")
    if re.search(r"(有人说|不是.*而是|别人.*我们|他们.*我们)", text):
        tags.append("薛辉·反击句式/对比")
    if re.search(r"(块钱|免费|不要钱|吃|喝|买|菜市场|院子)", text):
        tags.append("薛辉·画面感细节")
    if i == n - 1:
        tags.append("薛辉·收束升华")
    return tags or ["志楠·节奏控制"]


def fallback_comment(i, n, text) -> str:
    if i == 0:
        return "开场即抛出反常识观点, 承担前三秒留人任务; 句式短促, 信息密度高。"
    if i == n - 1:
        return "结尾回扣开头形成闭环, 用更高维度的定义完成价值升华。"
    return "中段延续主线, 用具体细节支撑观点, 保持节奏与信息密度。"


def chi_words(text: str, top: int = 8) -> list:
    """无分词词典 → 2/3-gram 近似, 去停用字边, 去包含重复。"""
    text = re.sub(r"[^\u4e00-\u9fff]", " ", text)
    grams = {}
    for w in text.split():
        for n in (2, 3):
            for k in range(len(w) - n + 1):
                g = w[k:k + n]
                grams[g] = grams.get(g, 0) + 1
    cands = [(g, c) for g, c in grams.items()
             if c >= 2 and g[0] not in SENT_STOP and g[-1] not in SENT_STOP]
    cands.sort(key=lambda x: -x[1])
    kept = []
    for g, c in cands:
        if not any(g in k for k, _ in kept):
            kept.append((g, c))
        if len(kept) >= top:
            break
    mx = kept[0][1] if kept else 1
    return [{"word": g, "count": c, "pct": max(6, round(c / mx * 100))}
            for g, c in kept]


def sentiment_split(text: str) -> list:
    sents = [s for s in re.split(r"[。！？；\n]", text) if s.strip()]
    pos = sum(1 for s in sents if any(k in s for k in POS_KW))
    neg = sum(1 for s in sents if any(k in s for k in NEG_KW))
    neu = max(len(sents) - pos - neg, 0)
    tot = max(pos + neg + neu, 1)
    rows = [("正面", pos, "var(--green)"), ("中性", neu, "var(--yellow)"),
            ("负面", neg, "var(--red)")]
    out, acc = [], 0
    for i, (label, v, color) in enumerate(rows):
        pct = round(v / tot * 100) if i < 2 else max(0, 100 - acc)
        acc += pct
        out.append({"label": label, "pct": pct, "color": color})
    return out


def comment_insight(text: str) -> str:
    if any(k in text for k in ("回去", "回云南", "回家", "社保")):
        return "在外游子的身份共鸣——最能带动转发"
    if any(k in text for k in ("同感", "真实", "说得对", "就是这样")):
        return "强认同型评论, 印证文案戳中人群"
    if any(k in text for k in ("不是", "而是", "——")):
        return "评论本身有文案意识, 可沉淀为选题素材"
    if any(k in text for k in ("但", "不过", "其实")):
        return "补充视角, 可作为下一条视频的回应点"
    return "氛围型互动"


def classify_comment(text: str) -> str:
    if any(k in text for k in NEG_KW):
        return "neg"
    if any(k in text for k in POS_KW):
        return "pos"
    return "neu"


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--workdir", required=True)
    ap.add_argument("--title", default="")
    args = ap.parse_args()
    wd = args.workdir

    meta = load_json(os.path.join(wd, "meta.json")) or {}
    tr = load_json(os.path.join(wd, "transcript.json")) or {}
    frames = load_json(os.path.join(wd, "frames.json")) or []
    comments = load_json(os.path.join(wd, "comments.json")) or []

    data = tr.get("data") or {}
    segs = data.get("segments") or []
    fulltext = data.get("text") or "".join(s.get("text", "") for s in segs)
    if not segs and fulltext:  # 无时间戳转写 → 整体一段
        segs = [{"bg": 0, "ed": int(meta.get("duration_s", 60) * 1000),
                 "speaker": "0", "text": fulltext}]
    if not segs:
        raise SystemExit("[report] 没有转写内容, 先跑 transcribe.py")
    duration_s = meta.get("duration_s") or (segs[-1]["ed"] / 1000)

    # --- LLM 逐段点评(失败回退启发式) ---
    llm_engine, llm_out = "规则启发式", {}
    try:
        seg_lines = "\n".join(
            f"[{fmt_ts(s['bg'])}-{fmt_ts(s['ed'])}] {s.get('text','')}" for s in segs)
        frame_lines = "\n".join(
            f"{f['t_label']}: {f.get('caption','')}" for f in frames) or "(无)"
        st = meta.get("stats") or {}
        prompt = LLM_PROMPT.format(
            dur=round(duration_s), digg=st.get("digg", "?"), cmt=st.get("comment", "?"),
            coll=st.get("collect", "?"), share=st.get("share", "?"),
            fulltext=fulltext[:1200], segment_lines=seg_lines, frame_lines=frame_lines)
        raw = call_llm(prompt)
        llm_out = parse_llm_json(raw)
        if llm_out.get("segments"):
            llm_engine = TEXT_MODEL
            print(f"[report] LLM 点评 {len(llm_out['segments'])} 段 ({llm_engine})")
        else:
            print("[report] LLM 输出无法解析, 回退规则启发式")
    except Exception as ex:
        print(f"[report] LLM 调用失败({ex}), 回退规则启发式")

    llm_segs = {int(s.get("i", i)): s for i, s in enumerate(llm_out.get("segments") or [])}

    # --- 组装分析段 ---
    n = len(segs)
    analysis_segs = []
    for i, s in enumerate(segs):
        t0, t1 = int(s.get("bg", 0)), int(s.get("ed", 0))
        text = s.get("text", "")
        L = llm_segs.get(i) or {}
        techniques = [t for t in (L.get("techniques") or
                                  heuristic_techniques(i, n, text)) if t]
        frame = min(frames, key=lambda f: abs(f["t_ms"] - t0),
                    default=None) if frames else None
        if frame and abs(frame["t_ms"] - t0) > 8000:
            frame = None
        analysis_segs.append({
            "i": i, "bg": t0, "ed": t1,
            "t_start": fmt_ts(t0), "t_end": fmt_ts(t1), "t_start_s": t0 // 1000,
            "text": esc(text), "preview": esc(text[:24] + ("…" if len(text) > 24 else "")),
            "techniques": [{"label": esc(t), "cls": tag_cls(t)} for t in techniques],
            "func": esc(L.get("func") or ""),
            "comment_html": esc(L.get("comment") or fallback_comment(i, n, text)),
            "reuse": esc(L.get("reuse") or ""),
            "frame": frame,
        })

    # --- 整体结构表 ---
    stage_names = ["起", "承", "转", "合", "收"]
    structure_rows = []
    for idx, sg in enumerate(analysis_segs):
        stage = sg["func"] or stage_names[min(int(idx / n * len(stage_names)),
                                              len(stage_names) - 1)]
        structure_rows.append({
            "stage": esc(stage), "time": f"{sg['t_start']}—{sg['t_end']}",
            "func": esc(re.sub(r"^.{0,3}[:：]", "", sg["func"]) if sg["func"]
                        else ("开场留人" if idx == 0 else
                              ("收束升华" if idx == n - 1 else "承接展开"))),
            "techniques": " ".join(t["label"] for t in sg["techniques"]) or "—",
        })
    summary_html = esc(llm_out.get("structure_summary") or
                       f"{n}段完成全片, 各段技法如上; 逐段细节见上方可折叠卡片。")

    # --- 词频/情绪 ---
    wf = chi_words(fulltext)
    sentiment = sentiment_split(fulltext)
    word_count = len(re.sub(r"\s", "", fulltext))
    wpm = round(word_count / max(duration_s / 60, 0.1))

    # --- 评论区 ---
    comment_stats, top_comments, comment_keywords, comment_total = [], [], [], 0
    if comments:
        cls_counts = {"pos": 0, "neu": 0, "neg": 0}
        for c in comments:
            cls_counts[classify_comment(c.get("text", ""))] += 1
        comment_total = len(comments)
        rows = [("正面认同", cls_counts["pos"], "var(--green)"),
                ("中性补充", cls_counts["neu"], "var(--yellow)"),
                ("争议", cls_counts["neg"], "var(--red)")]
        acc = 0
        for i, (label, v, color) in enumerate(rows):
            pct = round(v / comment_total * 100) if i < 2 else max(0, 100 - acc)
            acc += pct
            comment_stats.append({"label": label, "pct": pct, "color": color})
        top_comments = [{"nickname": esc(c.get("nickname", "匿名")),
                         "likes": c.get("likes", 0), "text": esc(c.get("text", "")),
                         "insight": esc(comment_insight(c.get("text", "")))}
                        for c in comments[:5]]
        comment_keywords = chi_words(" ".join(c.get("text", "") for c in comments), top=6)

    # --- 数据卡片 ---
    st = meta.get("stats") or {}
    stats_cards = [{"num": f"{duration_s:.0f}s", "label": "总时长"}]
    if st.get("digg"):
        stats_cards += [{"num": f"{st['digg']:,}", "label": "👍 点赞"},
                        {"num": f"{st['comment']:,}", "label": "💬 评论"},
                        {"num": f"{st['collect']:,}", "label": "🔖 收藏"},
                        {"num": f"{st['share']:,}", "label": "↗ 分享"},
                        {"num": f"{st['digg'] / max(st['comment'], 1):.0f}:1",
                         "label": "赞评比"}]
    else:
        stats_cards += [{"num": len(comments), "label": "评论采集"},
                        {"num": len(frames), "label": "抽帧"},
                        {"num": n, "label": "分段"},
                        {"num": word_count, "label": "总字数"},
                        {"num": wpm, "label": "字/分"}]

    # --- 建议 ---
    suggestions = [{"title": esc(s.get("title", f"建议{ i + 1 }")),
                    "body": esc(s.get("body", ""))}
                   for i, s in enumerate(llm_out.get("suggestions") or [])]
    if not suggestions:
        hw = wf[0]["word"] if wf else "核心词"
        pos_pct = next((r["pct"] for r in (comment_stats or sentiment)
                        if r["label"] in ("正面认同", "正面")), 0)
        suggestions = [
            {"title": "强化记忆锚点", "body": f"高频词「{hw}」贯穿全片, 封面/标题/评论区置顶都应重复它, 形成账号记忆点。"},
            {"title": "延续情绪线", "body": f"正面情绪约{pos_pct}%, 下一条保持同一情绪基调并升级一个具体场景细节。"},
            {"title": "补强行动指令", "body": "结尾闭环后加一句轻互动引导(如「你家乡呢?」), 把共鸣转化为评论量。"},
        ]

    # --- 汇总上下文(所有动态值已转义) ---
    author = meta.get("author") or "本地视频"
    title = args.title or f"拉片分析:{author}"
    today = datetime.date.today().strftime("%Y%m%d")
    aweme_id = str(meta.get("aweme_id") or "local")
    ctx = {
        "meta": {
            "title": esc(title),
            "subtitle": esc(f"{(meta.get('desc') or '')[:42]} · {duration_s:.0f}秒 · "
                            f"{n}段 · 方法论对照拆解"),
            "author": esc(author), "author_signature": esc(meta.get("author_signature", "")),
            "tags": [esc(t) for t in (meta.get("tags") or [])],
            "publish_time": esc(meta.get("create_time", "")),
            "aweme_id": esc(aweme_id), "platform": esc(meta.get("platform", "douyin")),
            "generated_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
            "llm_engine": esc(llm_engine),
            "report_id": esc(f"lapian-{meta.get('platform','douyin')}-{today}-{aweme_id[:10]}"),
            "word_count": word_count, "wpm": wpm,
            "sentence_count": len([s for s in re.split(r"[。！？]", fulltext) if s.strip()]),
        },
        "stats_cards": stats_cards,
        "segments": analysis_segs,
        "structure_rows": structure_rows,
        "summary_html": summary_html,
        "wordfreq": wf,
        "sentiment": sentiment,
        "comments": bool(comments),
        "comment_total": comment_total,
        "comment_stats": comment_stats,
        "top_comments": top_comments,
        "comment_keywords": comment_keywords,
        "suggestions": suggestions,
    }

    html = render_template(open(TEMPLATE, encoding="utf-8").read(), ctx)
    out_html = os.path.join(wd, "report.html")
    open(out_html, "w", encoding="utf-8").write(html)
    json.dump({"meta": {k: v for k, v in ctx["meta"].items()},
               "segments": [{k: (v if k != "text" else v) for k, v in sg.items()
                             if k not in ("frame",)} for sg in analysis_segs],
               "suggestions": suggestions,
               "llm_engine": llm_engine},
              open(os.path.join(wd, "analysis.json"), "w", encoding="utf-8"),
              ensure_ascii=False, indent=2, default=str)
    print(f"[report] {out_html} ({os.path.getsize(out_html)/1024:.0f} KB)")


if __name__ == "__main__":
    main()