#!/usr/bin/env python3 """Build a source-labelled audience-to-topic brief. Default: committed input only. --refresh uses only anonymous `video info` and first-page `comment list` reads. It neither installs a skill nor opens the user's Bilibili credentials/configuration. """ from __future__ import annotations import argparse import datetime as dt import html import json import os from pathlib import Path import re import shutil import subprocess import tempfile ROOT = Path(__file__).resolve().parent REPOSITORY = "kukemc/bili-cli" REVISION = "e17ab4da33a5b32a5fb4706e0a6dbb9ed994085a" VERSION = "1.2.1" INPUT = ROOT / "audience-input.json" TOPICS = ( ("setup", "安装失败时，先判断是哪一层", ("安装", "报错", "错误", "windows", "npm", "网络"), "用三个真实失败入口演示环境、依赖与Gateway的区别，逐项给出成功回执。", "诊断清单 + 一次失败到恢复的演示"), ("connect", "接入聊天渠道以后，怎样确认真的通了", ("飞书", "微信", "钉钉", "接入", "回调"), "把创建应用、接入、收消息与回消息分开验收，使用测试群与虚构消息。", "最小测试群回执 + 接入检查表"), ("cost", "先算清一次任务的模型与媒体成本", ("费用", "收费", "免费", "token", "额度", "api"), "区分聊天订阅、API额度和视频服务费用；展示自己的计费记录，不猜测观众预算。", "一次任务的带范围成本表"), ("use", "安装完成后，交付第一份能保存的成果", ("怎么用", "用途", "工作", "用处", "做什么"), "用公开测试素材完成一份可下载成果，给出输入、操作、输出与人工验收。", "可复现输入 + 实际成果文件"), ) def is_promotion(text): return bool(re.search( r"抱走|商单|学习咨询|点击获取|http|敲.{0,6}666|欢迎.{0,6}(?:选购|购买|下单|咨询)|" r"低至.{0,8}(?:折|元)|(?:扫码|私信).{0,8}(?:获取|领取|购买)", text, re.I)) def qualified_question(text, *, promotion=False): # Cost/topic nouns alone are not questions. A full-content promotion flag # survives excerpt truncation; legacy reviewed inputs still use their text. if promotion or len(text.strip()) < 5 or is_promotion(text): return False return bool(re.search(r"[？?]|怎么|为什么|怎样|如何|多少|能否|是否|请问|请教|求助|" r"不知道|报错|无法|不能|不支持|(?:遇到|出现|提示).{0,12}错误", text)) def is_qualified(row): return qualified_question(row["text"], promotion=row.get("promotion", False)) def validate_snapshot(data): if data.get("mode") not in {"live-public", "fixture"}: raise ValueError("Input mode must explicitly be live-public or fixture") if data.get("upstream", {}).get("revision") != REVISION: raise ValueError("Unexpected upstream revision") video = data.get("video", {}) comments = data.get("comments", []) if not isinstance(comments, list) or not comments or len(comments) > 20: raise ValueError("Expected 1–20 bounded comment inputs") if data["mode"] == "live-public": bvid = video.get("bvid", "") if not re.fullmatch(r"BV[0-9A-Za-z]{10}", bvid): raise ValueError("Public snapshot needs a valid BV id") if video.get("url") != f"https://www.bilibili.com/video/{bvid}": raise ValueError("Unexpected public video URL") if data.get("collection", {}).get("anonymous") is not True: raise ValueError("Public collection must explicitly be anonymous") elif video.get("url") or video.get("bvid") or video.get("stat"): raise ValueError("A fixture must not impersonate a real video or its metrics") ids = set() for row in comments: ident, text = row.get("id"), row.get("text") if not isinstance(ident, str) or ident in ids or not isinstance(text, str) or not text.strip(): raise ValueError("Missing/duplicate comment id or empty text") if len(text) > 180: raise ValueError("Comment excerpts must be bounded") if "promotion" in row and type(row["promotion"]) is not bool: raise ValueError("Comment promotion flag must be a boolean when present") if data["mode"] == "live-public" and not ident.isdigit(): raise ValueError("Public comment id must be an actual rpid") if "user" in row or "mid" in row or "name" in row: raise ValueError("This brief does not publish audience identities") ids.add(ident) return data def build_topics(snapshot): validate_snapshot(snapshot) topics = [] for ident, title, words, angle, artifact in TOPICS: matches = [row for row in snapshot["comments"] if is_qualified(row) and any(word in row["text"].lower() for word in words)] if matches: topics.append({"id": ident, "title": title, "angle": angle, "artifact": artifact, "evidence": [row["id"] for row in matches], "count": len(matches)}) return sorted(topics, key=lambda row: (-row["count"], row["id"]))[:3] def read_cli(binary, arguments, state_dir): # Each argv is constructed by this file. No user-supplied shell or write command. env = {**os.environ, "BILI_CLI_HOME": str(state_dir), "BILI_CLI_TIMEOUT": "25", "PYTHONIOENCODING": "utf-8", "PYTHONDONTWRITEBYTECODE": "1"} result = subprocess.run([str(binary), "--compact", *arguments], env=env, capture_output=True, text=True, encoding="utf-8", timeout=35, check=False) try: payload = json.loads(result.stdout) except ValueError as error: raise ValueError("Bili CLI did not return JSON; existing snapshot is preserved") from error if result.returncode or payload.get("ok") is not True: error = payload.get("error", {}) raise ValueError(f"Anonymous read failed ({error.get('code', 'unknown')}): {error.get('message', '')}") return payload["data"] def collect(binary, bvid): if not re.fullmatch(r"BV[0-9A-Za-z]{10}", bvid): raise ValueError("Pass a BV id, not a command or arbitrary URL") with tempfile.TemporaryDirectory(prefix="bili-anonymous-") as directory: state = Path(directory) version = read_cli(binary, ["--version"], state) if version.get("version") != VERSION: raise ValueError(f"Expected bili-agent {VERSION}; use the pinned source") video = read_cli(binary, ["video", "info", bvid], state) comments = read_cli(binary, ["comment", "list", bvid, "--page", "1", "--sort", "hot"], state) if video.get("bvid") != bvid: raise ValueError("Video id mismatch") if (state / "credential.json").exists(): raise ValueError("Anonymous read unexpectedly wrote a credential") rows = [] for row in comments.get("items", [])[:12]: if not row.get("content"): continue content = str(row["content"]) promotion = is_promotion(content) # Inspect before truncation; never export the full content. excerpt = " ".join(content.split()[:20])[:40] rows.append({"id": str(row["rpid"]), "text": excerpt, "promotion": promotion}) snapshot = {"mode": "live-public", "fetchedAt": dt.datetime.now(dt.timezone.utc).isoformat(), "upstream": {"repository": REPOSITORY, "revision": REVISION, "version": VERSION}, "video": {key: video.get(key) for key in ("bvid", "url", "title", "duration", "pubdate", "stat")}, "comments": rows, "collection": {"anonymous": True, "page": 1, "sort": "hot", "scope": "公开热门评论第一页中的至多12条节选；推广标记按截短前完整文本计算；省略用户身份和楼中楼，不是完整观众调查"}} return validate_snapshot(snapshot) def render(snapshot): topics = build_topics(snapshot) esc = lambda value: html.escape(str(value), quote=True) fixture = snapshot["mode"] == "fixture" badge = "虚构观众提问 · fixture演练" if fixture else "真实公开数据 · 匿名读取快照" scope = "本页输入为明确虚构的课堂素材，不是实抓评论，不代表真实受众或平台趋势。" if fixture else snapshot["collection"]["scope"] qualified = sum(is_qualified(row) for row in snapshot["comments"]) by_id = {row["id"]: row for row in snapshot["comments"]} cards, markdown = [], ["# 观众提问 → 下期选题简报", "", badge, "", scope, "", f'输入{len(snapshot["comments"])}条，合格提问{qualified}条，形成{len(topics)}条选题提案。', ""] for index, topic in enumerate(topics, 1): quotes = "".join(f'
  • {esc(ident)} {esc(by_id[ident]["text"])}
    • ' for ident in topic["evidence"][:2]) cards.append(f'
    0{index}

    样本内 {topic["count"]} 条相关输入 · 可重复归类

    {esc(topic["title"])}

    {esc(topic["angle"])}

    最小成果
    {esc(topic["artifact"])}

    查看选题依据
    ') markdown += [f'## {index}. {topic["title"]}', "", topic["angle"], f'最小成果：{topic["artifact"]}', '依据ID：' + '、'.join(topic["evidence"]), ""] if not cards: markdown += ["样本不足，暂不生成选题；资料推广、口令和没有明确问题的回复不作为需求依据。", ""] cards.append('

    样本不足，暂不生成选题

    本次未发现可归类的观众问题。资料推广、口令和无具体问题的回复不会被当成需求；不从视频播放量推断选题有市场。

    切换到明确虚构的提问演练 →

    ') source = ('无真实视频链接；输入完全虚构。' if fixture else f'{esc(snapshot["video"]["title"])}') refs = "".join(f'
  • {esc(row["id"])} {esc(row["text"])} — {"纳入问题归类" if is_qualified(row) else "未纳入：没有明确问题或属于推广/口令"}
    • ' for row in snapshot["comments"]) document = f'''观众提问 → 下期选题简报

    COGINE / OPENCLAW CONTENT LAB

    {badge}

    先听见问题，
    再决定下一期拍什么。

    {esc(scope)}

    样本：{len(snapshot["comments"])}条 · 合格提问：{qualified}条 · 主题：{len(topics)}个 · 采样／制作日期：{esc(snapshot["fetchedAt"])}

    真实公开样本 · 虚构提问演练

    {''.join(cards)}

    这是编辑简报，不是受众画像

    关键词归类可复现，一条留言可属于多个主题。条数是本样本内的留言数，不是独立人数、需求比例、增长率或转化预测。选题角度由本案例编辑规则提出，尚未发布或验证流量。

    刷新时先用完整留言判定推广，再只保存有限节选与布尔标记；原有未带标记的已审核输入仍按节选判定。费用等主题词本身不算疑问或求助。规则不能替代人工审核。

    来源：{source}

    全部输入与出处
      {refs}

    输入JSON · Markdown简报 · 复现脚本

    ''' markdown += ["## 来源与边界", "", scope, snapshot["video"].get("url") or "无真实视频链接（fixture）", "", "刷新先按完整留言判定推广，只保存节选与布尔标记；原有无标记输入按节选判定。费用等主题词本身不算疑问或求助，规则仍需人工审核。", "", "归类为规则输出，选题角度为编辑提案；未发布、未验证增长。"] return document, "\n".join(markdown) + "\n" def preview(snapshot): count = len(snapshot["comments"]) questions = sum(is_qualified(row) for row in snapshot["comments"]) topics = len(build_topics(snapshot)) label = "真实公开样本" if snapshot["mode"] == "live-public" else "明确虚构的fixture" outcome = "样本不足，不硬造需求" if not topics else "规则归类，编辑提案待验证" return f'''COGINE / BILIBILI CONTENT LAB先听见问题，再决定拍什么{label} · {html.escape(snapshot["fetchedAt"][:10])} · 匿名读取{count} 条输入 → {questions} 条提问 → {topics} 条选题{outcome}另附独立标记的虚构提问演练；未发布、未验证转化。''' def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--refresh", action="store_true") parser.add_argument("--bili", type=Path, help="Pinned temporary venv's bili executable") parser.add_argument("--video", default="BV1dCEX64EyY") parser.add_argument("--output-dir", type=Path, default=ROOT) args = parser.parse_args() standalone = args.output_dir.resolve() != ROOT.resolve() if standalone and args.output_dir.exists(): parser.error("--output-dir must name a new directory") if args.refresh and not args.bili: parser.error("--refresh requires --bili; no global executable is selected implicitly") snapshot = collect(args.bili, args.video) if args.refresh else json.loads(INPUT.read_text(encoding="utf-8")) document, brief = render(snapshot) # Validate everything before changing existing outputs. fixture = json.loads((ROOT / "fixture-input.json").read_text(encoding="utf-8")) if fixture.get("mode") != "fixture": raise ValueError("The separate demonstration must remain labelled fixture") fixture_document, fixture_brief = render(fixture) outputs = {"index.html": document, "topic-brief.md": brief, "preview.svg": preview(snapshot), "fixture.html": fixture_document, "fixture-brief.md": fixture_brief} args.output_dir.mkdir(parents=True, exist_ok=True) if standalone: for name in ("fixture-input.json", "LICENSE.upstream.txt", "reproduce.py"): shutil.copyfile(ROOT / name, args.output_dir / name) if args.refresh or standalone: outputs["audience-input.json"] = json.dumps(snapshot, ensure_ascii=False, indent=2) + "\n" for name, text in outputs.items(): with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=args.output_dir, delete=False) as handle: handle.write(text) temporary = handle.name os.replace(temporary, args.output_dir / name) print(f'Built {len(build_topics(snapshot))} topic proposals from {snapshot["mode"]}; no publishing.') if __name__ == "__main__": main()