diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 03f00b9..fdf2c80 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,9 +15,11 @@ jobs: uses: actions/setup-python@v5 with: python-version: "3.11" - - name: Install (editable, deps from pyproject) - run: pip install -e . + - name: Install (editable + dev deps from pyproject) + run: pip install -e ".[dev]" - name: Syntax check (compile all scripts) run: python -m compileall scripts - name: Smoke test (console entry point) run: viralens --help + - name: Unit tests + run: pytest diff --git a/README.md b/README.md index b812eed..808fe70 100644 --- a/README.md +++ b/README.md @@ -239,7 +239,7 @@ creators.py ──▶ fetch_multi.py ──▶ data/_videos.json ( - **Signal scanner** (`scan_signals`): turns each video into a universal feature vector, then auto-tests every dimension (title patterns, length buckets, daypart, cover metrics…) for high/low-play separation, ranks by effect size, and reports which levers are *universal* vs *creator-specific*. -- **L2 — text** (`subtitle`, `comments`): subtitles + hot comments → `jieba` keyword analysis. +- **L2 — text**: subtitles + hot comments → `jieba` keyword analysis (cross-creator comment-engagement test: `compare_meme.py`). - **Cross-creator / cross-zone gate**: a pattern earns a ✅ only if it survives the *same test* on multiple independent creators **and** more than one zone. diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..d1d00f2 --- /dev/null +++ b/conftest.py @@ -0,0 +1,5 @@ +"""Let pytest import the flat scripts/ modules without installing the package.""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent / "scripts")) diff --git a/packaging/viralens.spec b/packaging/viralens.spec index 9858320..3080b2c 100644 --- a/packaging/viralens.spec +++ b/packaging/viralens.spec @@ -50,9 +50,9 @@ hiddenimports += [ "fetch_multi", "fetch_bilibili", "fetch_youtube", "compare_form", "creator_profile", "scan_signals", "charts", "export_data", "build_report", "diagnose", "analyze_video", "import_private", - "creators", "features", "benchmarks", - "classify_and_stats", "comments", "compare_meme", "fetch_covers", - "fetch_videos", "resolve_creators", "subtitle", + "creators", "features", "benchmarks", "shared_markers", "schema", + "compare_meme", "fetch_covers", + "resolve_creators", ] # —— 只读资源:网页界面 + 配置模板 —— 放进打包根目录,app.py 用 runtime.ASSET_DIR 找它们 —— diff --git a/pyproject.toml b/pyproject.toml index 9cee7d9..c3b8182 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,16 +19,17 @@ classifiers = [ "Topic :: Scientific/Engineering :: Information Analysis", ] dependencies = [ - "bilibili-api-python>=17.0.0", + "bilibili-api-python>=17.0.0,<18.0.0", "jieba>=0.42.1", "matplotlib>=3.7.0", "Pillow>=9.0.0", "numpy>=1.23.0", ] -# 仅在「打包成桌面 app」时需要:原生窗口后端。源码 CLI / 网页界面都不依赖它。 +# gui = 打包成桌面 app 时的原生窗口后端;dev = 跑测试。源码 CLI / 网页界面都不依赖它们。 [project.optional-dependencies] gui = ["pywebview>=5.0"] +dev = ["pytest>=7.0"] [project.urls] Homepage = "https://github.com/HarryXin0919/viralens" @@ -45,3 +46,6 @@ py-modules = ["viralens"] [tool.setuptools.package-dir] "" = "scripts" + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/requirements.txt b/requirements.txt index 310a1c6..63a9435 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,6 @@ # viralens — runtime dependencies # Install: pip install -r requirements.txt -bilibili-api-python>=17.0.0 # async Bilibili public-data access +bilibili-api-python>=17.0.0,<18.0.0 # async Bilibili public-data access (cap: 18.x may break API) jieba>=0.42.1 # Chinese word segmentation (comment keyword analysis) matplotlib>=3.7.0 # static README charts (PNG) Pillow>=9.0.0 # cover-image analysis (optional step: fetch_covers.py) diff --git a/scripts/analyze_video.py b/scripts/analyze_video.py index c6659d1..37600c6 100644 --- a/scripts/analyze_video.py +++ b/scripts/analyze_video.py @@ -16,9 +16,9 @@ 按需调用:用户在诊断页点「分析开头+配乐」才下这一条,不批量下 700 条。 命令行单测: - python analyze_video.py bidao BV1mT421Y7mE - python analyze_video.py mrbeast 0BjlBnfHcHM - python analyze_video.py bidao BV1mT421Y7mE --force # 忽略缓存重算 + python analyze_video.py # 如 demo_b1 BVxxxxxxxxxx + python analyze_video.py # YouTube 视频 id + python analyze_video.py --force # 忽略缓存重算 """ import base64 import json diff --git a/scripts/classify_and_stats.py b/scripts/classify_and_stats.py deleted file mode 100644 index 2ecc88e..0000000 --- a/scripts/classify_and_stats.py +++ /dev/null @@ -1,119 +0,0 @@ -""" -viralens · classify_and_stats.py -读 bidao_videos.json,自动给视频打类型标签,按多个维度算对比统计。 -纯本地,零成本,零网络。 - -跑: python classify_and_stats.py - -输出: - - data/classified.json (每个视频加了 type / 派生指标字段) - - 终端打印多维对比表 -""" -import json -import time -from pathlib import Path -from collections import defaultdict -from statistics import median, mean - -from runtime import DATA # 源码=仓库/data,打包成 app 时=用户数据目录 -IN = DATA / "bidao_videos.json" -OUT = DATA / "classified.json" - -NOW = time.time() - - -def year_of(v): - return (v.get("created_iso") or "")[:4] - - -def classify(v): - """规则分类。可能有误判,跑完人工校正。""" - title = v["title"] - desc = v.get("description", "") - tid = v.get("tid") - age_days = (NOW - (v.get("created_ts") or NOW)) / 86400 - - if age_days < 7: - return "新发布(数据未成熟)" - if tid != 201: - return "官方活动/其他分区" - if "×" in title: # 【UP主×品牌】格式 - return "商单合作" - if "科研大赏" in desc: - return "论文盘点(搞笑科研大赏)" - if "消防" in desc: - return "机构合作" - if "打脸" in title: - return "回应/补充" - return "正经科普" - - -def fmt(n): - return f"{n:,.0f}" if n else "-" - - -def main(): - videos = json.loads(IN.read_text(encoding="utf-8")) - - for v in videos: - v["type"] = classify(v) - play = v.get("play") or 0 - # 相对指标:每万播放的评论/弹幕数 → 反映"讨论激发力",不受粉丝量影响 - v["comment_per_10k"] = round((v.get("comment") or 0) / play * 10000, 1) if play else 0 - v["danmaku_per_10k"] = round((v.get("danmaku") or 0) / play * 10000, 1) if play else 0 - - OUT.write_text(json.dumps(videos, ensure_ascii=False, indent=2), encoding="utf-8") - - groups = defaultdict(list) - for v in videos: - groups[v["type"]].append(v) - - # === 维度 A:类型对比(哪种形式天花板最高) === - print("=" * 72) - print("【维度 A】类型对比 — 按组内平均播放降序") - print("=" * 72) - order = sorted(groups.items(), key=lambda kv: -mean([x["play"] for x in kv[1]])) - for typ, vs in order: - plays = [x["play"] for x in vs] - durs = [x["duration_sec"] for x in vs] - cpm = [x["comment_per_10k"] for x in vs] - print(f"\n■ {typ} (n={len(vs)})") - print(f" 播放 均值 {fmt(mean(plays))} | 中位 {fmt(median(plays))} | 范围 {fmt(min(plays))} ~ {fmt(max(plays))}") - print(f" 时长 均值 {mean(durs)/60:.1f} 分 | 范围 {min(durs)/60:.1f} ~ {max(durs)/60:.1f} 分") - print(f" 评论率 均值 {mean(cpm):.1f} 条/万播放") - - # === 维度 B:正经科普 2025 高低对比 === - print("\n" + "=" * 72) - print("【维度 B】正经科普 · 2025年 · 高低播放对比(控制时间→粉丝量相近)") - print("=" * 72) - sci2025 = [v for v in videos if v["type"] == "正经科普" and year_of(v) == "2025"] - sci2025.sort(key=lambda x: -x["play"]) - print(f" 2025 正经科普共 {len(sci2025)} 个") - print("\n 🔴 高播放 Top 5:") - for v in sci2025[:5]: - print(f" {fmt(v['play']):>11} {v['duration_sec']/60:>4.1f}分 评{v['comment_per_10k']:>5} {v['title'][:30]}") - print("\n 🔵 低播放 Bottom 5:") - for v in sci2025[-5:]: - print(f" {fmt(v['play']):>11} {v['duration_sec']/60:>4.1f}分 评{v['comment_per_10k']:>5} {v['title'][:30]}") - - # === 维度 D:时间演化 === - print("\n" + "=" * 72) - print("【维度 D】正经科普 · 按年份(看涨粉 + 内容演化)") - print("=" * 72) - by_year = defaultdict(list) - for v in videos: - if v["type"] == "正经科普": - by_year[year_of(v)].append(v["play"]) - for yr in sorted(by_year): - ps = by_year[yr] - print(f" {yr}: n={len(ps):>2} 播放中位 {fmt(median(ps)):>11} 均值 {fmt(mean(ps)):>11}") - - print("\n" + "=" * 72) - print("✅ 已写 data/classified.json") - print(" 注:维度 C(实验型 vs 思辨型)需要字幕,等 03 脚本") - print(" 分类可能有误判,把上面结果发我,我们一起校正") - print("=" * 72) - - -if __name__ == "__main__": - main() diff --git a/scripts/comments.py b/scripts/comments.py deleted file mode 100644 index 8e20b6f..0000000 --- a/scripts/comments.py +++ /dev/null @@ -1,173 +0,0 @@ -""" -viralens · comments.py -抓 pilot 10 个视频(2025正经科普 高5+低5)的热评,做三类成分分析: - 共同 = 跨视频高频词(创作者标签性反馈,如"涨知识""哈哈哈") - 典型 = 单视频 TF-IDF top 词(这条视频独有的记忆点) - 差异 = 高播放组 vs 低播放组 词频差(观众"买账/不买账"时各说什么) - -依赖: bilibili-api-python jieba -SESSDATA: 从 config_local.py 读(不进 git) -跑: python comments.py -输出: data/comments_raw.json + data/comment_components.json + 终端三类成分 -""" -import asyncio -import json -import sys -import re -from pathlib import Path -from collections import Counter - -from bilibili_api import comment, Credential -from bilibili_api.comment import CommentResourceType, OrderType -import jieba -import jieba.analyse - -if hasattr(sys.stdout, "reconfigure"): - sys.stdout.reconfigure(encoding="utf-8") # Win 控制台默认 GBK,强制 UTF-8 - -from runtime import DATA # 源码=仓库/data,打包成 app 时=用户数据目录 -try: - from config_local import SESSDATA -except ImportError: - SESSDATA = "" - -PER_VIDEO = 40 # 每视频抓多少条热评 - -# 停用词:虚词 + 太通用的口水词(创作者自己的名字建议按需加,否则会在词频里霸榜) -STOP = set(( - "的 了 是 我 你 他 她 它 们 在 也 都 和 与 就 不 没 有 这 那 个 啊 吧 吗 呢 嘛 哦 噢 " - "把 被 给 让 对 跟 还 又 很 太 真 哈 一个 这个 那个 什么 怎么 这样 就是 但是 而且 " - "所以 因为 如果 视频 一下 知道 觉得 感觉 现在 已经 一直 还是 可能 这种 自己 " - "然后 其实 真的 这么 那么 一样 时候 看到 出来 起来 一点 不是 没有 这里 那里 我们 " - "你们 他们 一种 不能 可以 这些 那些 的话 一定 应该 已经" -).split()) - - -def clean_tokens(text): - text = re.sub(r"[^一-龥a-zA-Z]", " ", text) # 只留中英文 - return [t for t in jieba.cut(text) if len(t) >= 2 and t.lower() not in STOP] - - -async def fetch_hot(aid, cred, want): - """抓按点赞排序的热评,返回 (replies, 错误说明)""" - reps, err = [], "" - try: - page = 1 - while len(reps) < want and page <= 5: - r = await comment.get_comments( - oid=aid, type_=CommentResourceType.VIDEO, - page_index=page, order=OrderType.LIKE, credential=cred) - batch = r.get("replies") or [] - if not batch: - break - reps.extend(batch) - page += 1 - await asyncio.sleep(0.4) - except Exception as e: - err = f"{type(e).__name__}: {e}" - return reps[:want], err - - -async def main(): - if not SESSDATA: - print("❌ 没读到 SESSDATA。把 config_local.example.py 复制为 config_local.py 并填入") - return - - classified = json.loads((DATA / "classified.json").read_text(encoding="utf-8")) - sci = [v for v in classified if v.get("type") == "正经科普" - and (v.get("created_iso") or "")[:4] == "2025"] - sci.sort(key=lambda x: -x["play"]) - pilot = [("高", v) for v in sci[:5]] + [("低", v) for v in sci[-5:]] - - cred = Credential(sessdata=SESSDATA) - raw = {} - docs = [] - print("抓评论中(每个约 1-2 秒)...\n") - for group, v in pilot: - reps, err = await fetch_hot(v["aid"], cred, PER_VIDEO) - if not reps: - print(f" ✗ [{group}] {v['title'][:20]}: {err or '无评论'}") - continue - items = [{ - "msg": r.get("content", {}).get("message", ""), - "like": r.get("like", 0), - "uname": r.get("member", {}).get("uname", ""), - } for r in reps] - raw[v["bvid"]] = {"title": v["title"], "group": group, "play": v["play"], "comments": items} - text = " ".join(i["msg"] for i in items) - docs.append({"group": group, "bvid": v["bvid"], "title": v["title"], - "tokens": clean_tokens(text), "text": text}) - print(f" ✓ [{group}] {v['title'][:20]:<20} 抓到 {len(items):>3} 条热评") - - if not docs: - print("\n⚠️ 一条评论都没抓到。把上面的 ✗ 报错贴给我,我改接口。") - return - - (DATA / "comments_raw.json").write_text( - json.dumps(raw, ensure_ascii=False, indent=2), encoding="utf-8") - - # ① 共同成分:文档频率(出现在多少个视频里) - df = Counter() - for d in docs: - for w in set(d["tokens"]): - df[w] += 1 - common = [(w, c) for w, c in df.most_common(30) if c >= max(3, len(docs) // 2)] - - # ② 典型成分:每视频 TF-IDF top - typical = {} - for d in docs: - kws = jieba.analyse.extract_tags(d["text"], topK=8) - typical[d["bvid"]] = {"title": d["title"], "group": d["group"], "keywords": kws} - - # ③ 差异成分:高 vs 低 词频(归一化到每千词),只看出现在≥2视频的词 - def group_counter(g): - c, total = Counter(), 0 - for d in docs: - if d["group"] == g: - c.update(d["tokens"]) - total += len(d["tokens"]) - return c, (total or 1) - - hi_c, hi_n = group_counter("高") - lo_c, lo_n = group_counter("低") - diff = [] - for w in set(hi_c) | set(lo_c): - if df[w] < 2: - continue - hf = hi_c[w] / hi_n * 1000 - lf = lo_c[w] / lo_n * 1000 - diff.append((w, hf - lf)) - hi_over = sorted(diff, key=lambda x: -x[1])[:12] - lo_over = sorted(diff, key=lambda x: x[1])[:12] - - out = { - "common": [{"w": w, "videos": c} for w, c in common], - "typical": typical, - "high_over": [{"w": w, "delta_per_1k": round(d, 2)} for w, d in hi_over], - "low_over": [{"w": w, "delta_per_1k": round(d, 2)} for w, d in lo_over], - } - (DATA / "comment_components.json").write_text( - json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8") - - print("\n" + "=" * 60) - print("① 共同成分 — 创作者标签性反馈(出现在 ≥半数 视频)") - print("=" * 60) - print(" " + " ".join(f"{w}×{c}" for w, c in common[:18])) - - print("\n" + "=" * 60) - print("③ 差异成分 — 高 vs 低播放组 观众语言(差值/千词)") - print("=" * 60) - print(" 🔴 高组更爱说: " + " ".join(f"{w}(+{d:.1f})" for w, d in hi_over)) - print(" 🔵 低组更爱说: " + " ".join(f"{w}({d:.1f})" for w, d in lo_over)) - - print("\n" + "=" * 60) - print("② 典型成分 — 每条视频的记忆点(TF-IDF)") - print("=" * 60) - for t in typical.values(): - print(f" [{t['group']}] {t['title'][:22]:<22} {' '.join(t['keywords'])}") - - print("\n✅ 已写 data/comments_raw.json + comment_components.json — 把整段贴给我解读") - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/scripts/compare_form.py b/scripts/compare_form.py index 870a8b4..ec8e8e6 100644 --- a/scripts/compare_form.py +++ b/scripts/compare_form.py @@ -17,19 +17,8 @@ from runtime import DATA # 源码=仓库/data,打包成 app 时=用户数据目录 from creators import CREATORS -# 偏离核心形式的粗标记(宁紧勿松,误判靠人眼复核打印的标题) -OFF_MARKERS = { - "商单": ["×", "合作", "赞助", "广告", "推广"], - "vlog": ["vlog"], - "直播": ["直播回放", "录播", "直播录"], - "访谈": ["专访", "对谈", "采访", "对话"], - "音乐": ["翻唱", "弹唱", "音乐区"], - # 英文(给 YouTube 等英文标题用;只用多字符安全词,绝不误伤中文标题) - # 不包含 "livestream"/"live stream":有些 Entertainment 创作者把 "secretly in X livestream" - # 当招牌挑战在用,误伤太大。真正的直播录像偏题标题通常含 "(Live)"/"VOD"/"Live Recording"。 - "EN": ["sponsored", "#ad", "(ad)", "[ad]", "paid promotion", - "podcast", "q&a", "interview"], -} +# 偏离核心形式的关键词 —— 单一数据源在 shared_markers.py(对比报告用保守口径版)。 +from shared_markers import OFF_MARKERS_COMPARE as OFF_MARKERS def off_tag(v): diff --git a/scripts/creators.py b/scripts/creators.py index e21299f..81ccc06 100644 --- a/scripts/creators.py +++ b/scripts/creators.py @@ -37,3 +37,38 @@ CREATORS = _LOCAL_CREATORS except Exception: pass + + +def validate_creators(creators=None): + """校验 CREATORS 配置,返回问题清单(空 = 没问题)。一次列出所有问题, + 而不是抓取时才因 KeyError 崩在第一条。常见拼错(zon→zone)、缺 alias、 + alias 重复(输出文件互相覆盖)、YouTube 缺 channel 都会被这里抓出来。""" + creators = CREATORS if creators is None else creators + if not isinstance(creators, list) or not creators: + return ["CREATORS 必须是非空列表"] + known = {"name", "alias", "zone", "platform", "uid", "channel", "min_duration_sec"} + problems, seen = [], set() + for i, c in enumerate(creators): + if not isinstance(c, dict): + problems.append(f"CREATORS[{i}]: 每条必须是 dict") + continue + where = f"CREATORS[{i}]「{c.get('name', '?')}」" + for req in ("name", "alias", "zone"): + if not c.get(req): + problems.append(f"{where}: 缺必填字段 '{req}'") + alias = c.get("alias") + if isinstance(alias, str) and alias: + if alias in seen: + problems.append(f"{where}: alias '{alias}' 重复 —— 输出文件会互相覆盖") + seen.add(alias) + elif alias is not None and not isinstance(alias, str): + problems.append(f"{where}: alias 必须是字符串") + plat = (c.get("platform") or "bilibili").lower() + if plat not in ("bilibili", "youtube"): + problems.append(f"{where}: 未知 platform '{plat}'(支持 bilibili / youtube)") + if plat == "youtube" and not c.get("channel"): + problems.append(f"{where}: youtube 创作者需要 'channel' 字段(@handle / UC.. / 频道名)") + for k in c: + if k not in known: + problems.append(f"{where}: 未知字段 '{k}'(拼错了?已知:{', '.join(sorted(known))})") + return problems diff --git a/scripts/diagnose.py b/scripts/diagnose.py index 6728fc3..d19aa7b 100644 --- a/scripts/diagnose.py +++ b/scripts/diagnose.py @@ -6,8 +6,8 @@ 全部中英双语。供 app.py 的 /api/diag 调用;也能命令行单测: python diagnose.py # 列个例子 - python diagnose.py bidao BV1gcfWYqEsf - python diagnose.py mrbeast + python diagnose.py # 如 demo_b1 BVxxxxxxxxxx + python diagnose.py 核心思路:不泛泛而谈。每条视频都跟「这个创作者自己的爆款」对照 —— 你自己的高播放视频封面平均饱和度多少?这条够不够?差在哪?该怎么调? diff --git a/scripts/features.py b/scripts/features.py index f332004..c8ccb36 100644 --- a/scripts/features.py +++ b/scripts/features.py @@ -9,17 +9,8 @@ import re from datetime import datetime -# —— 偏离"招牌形式"的粗标记(通用:商单 / vlog / 访谈 / 直播 / 音乐)—— -OFF_MARKERS = { - "商单": ["×", "合作", "赞助", "广告", "推广", "联名"], - "vlog": ["vlog"], - "直播": ["直播回放", "录播", "直播录"], - "访谈": ["专访", "对谈", "采访", "对话"], - "音乐": ["翻唱", "弹唱", "音乐区"], - # 英文(给 YouTube 等英文标题用;只用多字符安全词,绝不误伤中文标题) - "EN": ["sponsored", "#ad", "(ad)", "[ad]", "paid promotion", - "podcast", "q&a", "interview", "livestream", "live stream"], -} +# 偏离"招牌形式"的关键词 —— 单一数据源在 shared_markers.py(features 用全口径版)。 +from shared_markers import OFF_MARKERS_FULL as OFF_MARKERS # 标题里的"夸张/钩子"词(通用 clickbait 信号) SUPERLATIVE = ["最", "史上", "第一", "唯一", "居然", "竟然", "震惊", "千万", "全网", "没人", "真相", "终于", "99%", "100%", "绝了", "炸裂", "崩溃"] diff --git a/scripts/fetch_bilibili.py b/scripts/fetch_bilibili.py index 7a79ac5..56bed77 100644 --- a/scripts/fetch_bilibili.py +++ b/scripts/fetch_bilibili.py @@ -11,10 +11,16 @@ from bilibili_api import user, Credential +from schema import VideoRecord + if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") +class BilibiliError(Exception): + """对外抛出的可读错误,由 fetch_multi 调度器统一捕获打印(对应 YouTube 的 YouTubeError)。""" + + def parse_length(s): """'12:34' / '1:02:03' → 秒。""" if not s: @@ -27,42 +33,67 @@ def parse_length(s): return 0 -async def fetch_creator(c, sessdata, num=40, tries=4): - """标准适配器接口(异步):给一个 creators.py 条目 → 标准视频表 list[dict]。""" +PAGE_MAX = 50 # B 站 get_videos 每页上限 + + +def _to_record(c, v) -> VideoRecord: + """一条 B 站原始视频 → 标准视频表记录(字段契约见 schema.VideoRecord)。""" + ts = v.get("created", 0) + bvid = v.get("bvid") + return { + "creator": c["name"], "alias": c["alias"], "zone": c["zone"], + "platform": "bilibili", + "vid": bvid, "bvid": bvid, "aid": v.get("aid"), + "title": v.get("title"), "description": v.get("description", ""), + "cover_url": v.get("pic"), + "duration_sec": parse_length(v.get("length", "")), + "created_ts": ts, + "created_iso": datetime.fromtimestamp(ts).isoformat() if ts else None, + "play": v.get("play"), "comment": v.get("comment"), + "danmaku": v.get("video_review"), "tid": v.get("typeid"), + } + + +async def _get_page(u, ps, pn, name, tries): + """抓一页,带 412 风控退避重试;失败统一抛 BilibiliError。""" + for attempt in range(tries): + try: + return await u.get_videos(ps=ps, pn=pn) + except Exception as e: + if attempt == tries - 1: + raise BilibiliError( + f"{name}:调 B 站 API 失败(网络 / 风控 412?):{type(e).__name__}: {e}") from None + wait = 5 * (attempt + 1) # 5s,10s,15s 退避,清掉 412 风控 + print(f" ...{name} 第 {pn} 页第 {attempt + 1} 次失败(412?),{wait}s 后重试") + await asyncio.sleep(wait) + + +async def fetch_creator(c, sessdata, num=40, tries=4) -> "list[VideoRecord]": + """标准适配器接口(异步):给一个 creators.py 条目 → 标准视频表。 + **分页**抓取直到取够 num 条或没有更多(B 站每页上限 50;旧版只取第一页,num>50 会漏数据)。""" if not sessdata: - raise RuntimeError("没读到 SESSDATA(config_local.py)") + raise BilibiliError("没读到 SESSDATA —— 在 config_local.py 里填上(浏览器 Cookie 里的 SESSDATA)") uid = c.get("uid") if not uid: - raise RuntimeError("缺 UID,先跑 resolve_creators.py") + raise BilibiliError( + f"{c.get('name', '?')} 缺 uid —— 先跑 resolve_creators.py 查出 UP 主 ID 填进 creators.py") cred = Credential(sessdata=sessdata) u = user.User(uid=uid, credential=cred) - raw = None - for attempt in range(tries): - try: - raw = await u.get_videos(ps=num, pn=1) + + raw_videos = [] + pn = 1 + while len(raw_videos) < num: + ps = min(PAGE_MAX, num - len(raw_videos)) + raw = await _get_page(u, ps, pn, c["name"], tries) + page = (raw.get("list", {}) or {}).get("vlist", []) or [] + raw_videos.extend(page) + total = (raw.get("page") or {}).get("count") # 该 UP 主总投稿数(可提前停) + if len(page) < ps: # 这一页没满 → 没有更多了 break - except Exception: - if attempt == tries - 1: - raise - wait = 5 * (attempt + 1) # 5s,10s,15s 退避,清掉 412 风控 - print(f" ...{c['name']} 第{attempt + 1}次失败(412?),{wait}s 后重试") - await asyncio.sleep(wait) - vlist = raw.get("list", {}).get("vlist", []) - videos = [] - for v in vlist: - ts = v.get("created", 0) - bvid = v.get("bvid") - videos.append({ - "creator": c["name"], "alias": c["alias"], "zone": c["zone"], - "platform": "bilibili", - "vid": bvid, "bvid": bvid, "aid": v.get("aid"), - "title": v.get("title"), "description": v.get("description", ""), - "cover_url": v.get("pic"), - "duration_sec": parse_length(v.get("length", "")), - "created_ts": ts, - "created_iso": datetime.fromtimestamp(ts).isoformat() if ts else None, - "play": v.get("play"), "comment": v.get("comment"), - "danmaku": v.get("video_review"), "tid": v.get("typeid"), - }) + if total is not None and len(raw_videos) >= total: + break + pn += 1 + + videos = [_to_record(c, v) for v in raw_videos[:num]] videos.sort(key=lambda x: x["play"] or 0, reverse=True) return videos diff --git a/scripts/fetch_multi.py b/scripts/fetch_multi.py index 019451b..983f2a1 100644 --- a/scripts/fetch_multi.py +++ b/scripts/fetch_multi.py @@ -31,7 +31,7 @@ except ImportError: YOUTUBE_API_KEY = "" -from creators import CREATORS +from creators import CREATORS, validate_creators NUM_VIDEOS = 40 # 每个创作者抓最近多少个 @@ -50,6 +50,12 @@ async def fetch_for(c): async def main(): force = "--force" in sys.argv + problems = validate_creators() + if problems: + print("✗ creators.py 配置有误,请先修正:") + for p in problems: + print(f" - {p}") + return # 界面勾选了哪些平台就只抓哪些(app.py 通过环境变量传入;空=全抓,命令行单跑时不受限) only = [p.strip().lower() for p in os.environ.get("VIRALENS_PLATFORMS", "").split(",") if p.strip()] if only: diff --git a/scripts/fetch_videos.py b/scripts/fetch_videos.py deleted file mode 100644 index 6e6aa8b..0000000 --- a/scripts/fetch_videos.py +++ /dev/null @@ -1,113 +0,0 @@ -""" -viralens · fetch_videos.py -抓一个 B 站 UP 主最近 N 个视频的元数据,输出 JSON 给后续分析用。 -(这是早期单人脚本;多创作者请用 fetch_multi.py。) - -依赖: - python -m pip install bilibili-api-python aiohttp - -使用: - 1. 浏览器登录 b站(www.bilibili.com) - 2. F12 → Application/存储 → Cookies → https://www.bilibili.com → 找 SESSDATA - 复制 Value(看起来像 "abc123%2Cxxxx%2Cyyy...") - 3. 把 Value 粘贴到下方 SESSDATA 变量(引号内) - 4. 跑: python fetch_videos.py - -输出: ../data/_videos.json -""" -import asyncio -import json -import sys -from pathlib import Path -from datetime import datetime - -from bilibili_api import user, Credential - -# ============ 配置 ============ -UID = 0 # 换成你要抓的 UP 主 UID(resolve_creators.py 可按名字查) -NUM_VIDEOS = 30 # 抓最近多少个 -try: - from config_local import SESSDATA # ← SESSDATA 统一放 config_local.py(已 gitignore,不进 git) -except ImportError: - SESSDATA = "" -from runtime import DATA -OUTPUT = DATA / "bidao_videos.json" -# ============================= - - -def parse_length(s: str) -> int: - """B 站 length 字段是 'MM:SS' 或 'HH:MM:SS',转成秒""" - if not s: - return 0 - parts = [int(x) for x in s.split(":")] - if len(parts) == 2: - return parts[0] * 60 + parts[1] - if len(parts) == 3: - return parts[0] * 3600 + parts[1] * 60 + parts[2] - return 0 - - -async def main(): - if not SESSDATA: - print("❌ 没读到 SESSDATA。把 config_local.example.py 复制为 config_local.py 并填入你的 SESSDATA") - sys.exit(1) - - cred = Credential(sessdata=SESSDATA) - u = user.User(uid=UID, credential=cred) - - print(f"📡 抓取 UID={UID} 最近 {NUM_VIDEOS} 个视频...") - try: - raw = await u.get_videos(ps=NUM_VIDEOS, pn=1) - except Exception as e: - print(f"❌ 调用失败: {e}") - print(" 检查 SESSDATA 是否过期、网络是否能访问 b站") - sys.exit(1) - - vlist = raw.get("list", {}).get("vlist", []) - if not vlist: - print("❌ 没拿到视频。原始返回(前 500 字):") - print(str(raw)[:500]) - sys.exit(1) - - videos = [] - for v in vlist: - created_ts = v.get("created", 0) - videos.append({ - "bvid": v.get("bvid"), - "aid": v.get("aid"), - "title": v.get("title"), - "description": v.get("description", ""), - "cover_url": v.get("pic"), - "duration_sec": parse_length(v.get("length", "")), - "duration_raw": v.get("length"), - "created_ts": created_ts, - "created_iso": datetime.fromtimestamp(created_ts).isoformat() if created_ts else None, - "play": v.get("play"), - "comment": v.get("comment"), - "danmaku": v.get("video_review"), - "tid": v.get("typeid"), - "subtitle": v.get("subtitle", ""), - }) - - # 按播放量降序 - videos.sort(key=lambda x: x["play"] or 0, reverse=True) - - OUTPUT.parent.mkdir(parents=True, exist_ok=True) - with open(OUTPUT, "w", encoding="utf-8") as f: - json.dump(videos, f, ensure_ascii=False, indent=2) - - # 打印概览 - print(f"\n✅ 抓到 {len(videos)} 个视频") - print(f" 写入: {OUTPUT}") - print(f"\n📊 播放量分布:") - print(f" 最高: {videos[0]['play']:>10,} ← {videos[0]['title'][:35]}") - mid = videos[len(videos) // 2] - print(f" 中位: {mid['play']:>10,} ← {mid['title'][:35]}") - print(f" 最低: {videos[-1]['play']:>10,} ← {videos[-1]['title'][:35]}") - print(f"\n📅 时间跨度:") - print(f" 最早: {videos[-1]['created_iso'][:10] if videos[-1]['created_iso'] else '?'}") - print(f" 最近: {videos[0]['created_iso'][:10] if videos[0]['created_iso'] else '?'}") - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/scripts/schema.py b/scripts/schema.py new file mode 100644 index 0000000..07f6e35 --- /dev/null +++ b/scripts/schema.py @@ -0,0 +1,32 @@ +""" +viralens · schema.py —— 标准「视频记录」的字段契约(文档化用)。 + +所有平台适配器(fetch_bilibili / fetch_youtube)都吐出同一形状的 dict。以前这个 schema +只散落在各处的 dict 字面量里,传错字段很难发现。这里用一个 TypedDict 把它写下来,既给 +人看(字段一览),也能给类型检查器用。total=False:不同平台填的字段略有差异(B 站有 +danmaku、YouTube 有 like),都视为可选。 +""" +from __future__ import annotations + +from typing import Optional, TypedDict + + +class VideoRecord(TypedDict, total=False): + creator: str # 显示名 + alias: str # 输出文件名用的英文别名 + zone: str # 分区 / 赛道 + platform: str # "bilibili" | "youtube" + vid: str # 平台视频 id(B 站=bvid,YouTube=videoId) + bvid: str # B 站 BV 号 + aid: int # B 站 av 号 + title: str + description: str + cover_url: str + duration_sec: int + created_ts: int # 发布 Unix 时间戳 + created_iso: Optional[str] + play: Optional[int] # 播放 / views + comment: Optional[int] + danmaku: Optional[int] # B 站弹幕数(YouTube 无) + like: Optional[int] # YouTube 点赞(B 站这里不填) + tid: Optional[int] # B 站分区 id diff --git a/scripts/shared_markers.py b/scripts/shared_markers.py new file mode 100644 index 0000000..7c80f44 --- /dev/null +++ b/scripts/shared_markers.py @@ -0,0 +1,39 @@ +""" +viralens · shared_markers.py —— 「偏离招牌形式」关键词的**唯一数据源**。 + +历史上 features.py 和 compare_form.py 各自定义了一份 OFF_MARKERS,内容已经漂移 +(features 多了 "联名" 和 "livestream"/"live stream"),两套分析口径不一致。这里收口成 +一处,保留两份**故意不同**的命名常量: + +- OFF_MARKERS_FULL —— features.extract() 的全维度扫描用(口径较全)。 +- OFF_MARKERS_COMPARE —— compare_form 的头尾对比报告用(故意更保守,见下方注释)。 + +只含常量、不含逻辑,所以谁都能 import,绝无循环依赖。改关键词只改这里一处。 +""" + +# 全口径:特征工程 / 信号扫描用(features.py) +OFF_MARKERS_FULL = { + "商单": ["×", "合作", "赞助", "广告", "推广", "联名"], + "vlog": ["vlog"], + "直播": ["直播回放", "录播", "直播录"], + "访谈": ["专访", "对谈", "采访", "对话"], + "音乐": ["翻唱", "弹唱", "音乐区"], + # 英文(给 YouTube 等英文标题用;只用多字符安全词,绝不误伤中文标题) + "EN": ["sponsored", "#ad", "(ad)", "[ad]", "paid promotion", + "podcast", "q&a", "interview", "livestream", "live stream"], +} + +# 保守口径:头尾形式对比报告用(compare_form.py)。比 FULL 少两类,**刻意为之**: +# - 去掉 "联名":对比报告里宁可漏判,避免把联名款正片误标偏题。 +# - 去掉 "livestream"/"live stream":有些 Entertainment 创作者把 +# "secretly in X livestream" 当招牌挑战在用,误伤太大;真正的直播录像偏题标题 +# 通常含 "(Live)"/"VOD"/"Live Recording",不靠这两个词。 +OFF_MARKERS_COMPARE = { + "商单": ["×", "合作", "赞助", "广告", "推广"], + "vlog": ["vlog"], + "直播": ["直播回放", "录播", "直播录"], + "访谈": ["专访", "对谈", "采访", "对话"], + "音乐": ["翻唱", "弹唱", "音乐区"], + "EN": ["sponsored", "#ad", "(ad)", "[ad]", "paid promotion", + "podcast", "q&a", "interview"], +} diff --git a/scripts/subtitle.py b/scripts/subtitle.py deleted file mode 100644 index 30e2875..0000000 --- a/scripts/subtitle.py +++ /dev/null @@ -1,143 +0,0 @@ -""" -viralens · subtitle.py -抓 pilot 视频(2025正经科普 高5+低5)的 B站字幕,算"实验 vs 思辨"密度、语速、开场钩子。 -验证假设:高播放组 是不是 思辨密度更高、实验奇观更少。 - -依赖: bilibili-api-python aiohttp (已装) -SESSDATA: 从 01 脚本复制粘贴到下方 - -跑: python subtitle.py -输出: data/subtitle_features.json + 终端高低组对比 -""" -import asyncio -import json -import sys -from pathlib import Path -from statistics import mean - -import aiohttp -from bilibili_api import video, Credential - -if hasattr(sys.stdout, "reconfigure"): - sys.stdout.reconfigure(encoding="utf-8") # Win 控制台默认 GBK,强制 UTF-8 防 ✓/emoji 崩 - -from runtime import DATA # 源码=仓库/data,打包成 app 时=用户数据目录 -try: - from config_local import SESSDATA # ← 统一从 config_local.py 读(已 gitignore,不进 git) -except ImportError: - SESSDATA = "" - -# 粗糙词典:pilot 看趋势用,后续会换成更精细的 -EXP_WORDS = ["实验", "我们来", "你看", "倒进", "倒入", "加热", "点燃", "装置", - "材料", "试一下", "试试", "测量", "拍摄", "高速", "帧", "操作"] -THINK_WORDS = ["为什么", "其实", "真相", "你以为", "本质", "原因", "意味着", - "证明", "逻辑", "悖论", "假设", "定律", "并不是", "误解", "概念"] -HOOK_WORDS = ["为什么", "竟然", "居然", "真的", "你知道", "想象", "到底", "难道", "？", "?"] - - -async def get_subtitle_body(v): - """返回字幕 body 列表 [{from,to,content}] 或 (None, 错误说明)""" - info = await v.get_info() - cid = info.get("cid") or info["pages"][0]["cid"] - - subs = [] - err_trace = "" - # 试两种接口(版本差异) - for getter in ("get_player_info", "get_subtitle"): - try: - fn = getattr(v, getter) - raw = await fn(cid=cid) - subs = raw.get("subtitle", {}).get("subtitles", []) or raw.get("subtitles", []) - if subs: - break - except Exception as e: - err_trace += f"[{getter}:{e}] " - if not subs: - return None, f"无字幕({err_trace or '列表为空'})" - - # 优先中文字幕 - subs.sort(key=lambda s: 0 if "zh" in s.get("lan", "") else 1) - url = subs[0].get("subtitle_url", "") - if url.startswith("//"): - url = "https:" + url - if not url: - return None, "字幕无下载链接" - - async with aiohttp.ClientSession(headers={"User-Agent": "Mozilla/5.0"}) as sess: - async with sess.get(url) as r: - data = await r.json() - return data.get("body", []), None - - -def count(text, words): - return sum(text.count(w) for w in words) - - -async def main(): - if not SESSDATA: - print("❌ 请先粘 SESSDATA(从 01 脚本复制)") - return - - classified = json.loads((DATA / "classified.json").read_text(encoding="utf-8")) - sci = [v for v in classified if v.get("type") == "正经科普" - and (v.get("created_iso") or "")[:4] == "2025"] - sci.sort(key=lambda x: -x["play"]) - pilot = [("高", v) for v in sci[:5]] + [("低", v) for v in sci[-5:]] - - cred = Credential(sessdata=SESSDATA) - results = [] - print("抓字幕中(每个约 1-2 秒)...\n") - for group, v in pilot: - vid = video.Video(bvid=v["bvid"], credential=cred) - try: - body, err = await get_subtitle_body(vid) - except Exception as e: - body, err = None, f"异常 {e}" - if err: - print(f" ✗ [{group}] {v['title'][:22]}: {err}") - continue - - full = "".join(seg["content"] for seg in body) - opening = "".join(seg["content"] for seg in body if seg.get("from", 0) <= 30) - n = len(full) or 1 - dur_min = v["duration_sec"] / 60 - rec = { - "group": group, "bvid": v["bvid"], "title": v["title"], - "play": v["play"], "chars": len(full), - "speed_cpm": round(len(full) / dur_min), # 语速:字/分钟 - "exp_per_1k": round(count(full, EXP_WORDS) / n * 1000, 1), - "think_per_1k": round(count(full, THINK_WORDS) / n * 1000, 1), - "hook_open": count(opening, HOOK_WORDS), - } - rec["think_exp_ratio"] = round(rec["think_per_1k"] / (rec["exp_per_1k"] + 0.1), 2) - results.append(rec) - print(f" ✓ [{group}] {v['title'][:22]:<22} 思辨{rec['think_per_1k']:>4} 实验{rec['exp_per_1k']:>4} " - f"比{rec['think_exp_ratio']:>4} 语速{rec['speed_cpm']} 开场钩子{rec['hook_open']}") - - if not results: - print("\n⚠️ 一个字幕都没抓到。把上面的 ✗ 报错贴给我,我改接口。") - return - - (DATA / "subtitle_features.json").write_text( - json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8") - - # 高低组对比 - print("\n" + "=" * 60) - print("高 vs 低 播放组 · 字幕特征对比") - print("=" * 60) - for g in ("高", "低"): - rows = [r for r in results if r["group"] == g] - if not rows: - continue - print(f"\n■ {g}播放组 (n={len(rows)})") - print(f" 思辨密度 {mean(r['think_per_1k'] for r in rows):.1f} /千字") - print(f" 实验密度 {mean(r['exp_per_1k'] for r in rows):.1f} /千字") - agg_ratio = mean(r['think_per_1k'] for r in rows) / (mean(r['exp_per_1k'] for r in rows) + 0.1) - print(f" 思辨/实验比(组合计,抗异常值) {agg_ratio:.2f}") - print(f" 语速 {mean(r['speed_cpm'] for r in rows):.0f} 字/分") - print(f" 开场钩子词 {mean(r['hook_open'] for r in rows):.1f} 个") - print("\n✅ 已写 data/subtitle_features.json — 把上面整段贴给我解读") - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/tests/test_engine.py b/tests/test_engine.py new file mode 100644 index 0000000..1d34180 --- /dev/null +++ b/tests/test_engine.py @@ -0,0 +1,101 @@ +"""scan_signals.spearman, build_report formatters, runtime paths, +and fetch_bilibili (parse_length / record / errors / pagination) — no network.""" +import asyncio + +import pytest + +import build_report as br +import fetch_bilibili as fb +import runtime +from scan_signals import spearman + + +# ----------------------------- spearman ----------------------------- +def test_spearman_perfect_positive(): + r = spearman([1, 2, 3, 4, 5, 6], [10, 20, 30, 40, 50, 60]) + assert r is not None and r > 0.99 + + +def test_spearman_perfect_negative(): + r = spearman([1, 2, 3, 4, 5, 6], [60, 50, 40, 30, 20, 10]) + assert r is not None and r < -0.99 + + +def test_spearman_small_sample_is_none(): + assert spearman([1, 2, 3], [3, 2, 1]) is None # n < 6 → 信号太弱,返回 None + + +# ----------------------------- build_report formatters ----------------------------- +def test_fmt_play(): + assert br.fmt_play(814_0000) == "814.0万" + assert br.fmt_play(150_000_000) == "1.50亿" + assert br.fmt_play(None) == "-" + + +def test_fmt_dur(): + assert br.fmt_dur(125) == "2:05" + assert br.fmt_dur(None) == "-" + + +def test_video_url(): + assert "bilibili.com/video/BV1xx" in br.video_url({"platform": "bilibili", "bvid": "BV1xx"}) + assert "youtube.com/watch?v=abc" in br.video_url({"platform": "youtube", "vid": "abc"}) + assert br.video_url({"platform": "bilibili"}) == "" # 没 id + + +# ----------------------------- runtime paths (source mode) ----------------------------- +def test_runtime_source_paths(): + assert runtime.FROZEN is False + assert runtime.DATA.name == "data" + assert runtime.REPORTS.name == "reports" + + +# ----------------------------- fetch_bilibili ----------------------------- +def test_parse_length(): + assert fb.parse_length("12:34") == 12 * 60 + 34 + assert fb.parse_length("1:02:03") == 3723 + assert fb.parse_length("") == 0 + + +def test_to_record_shape(): + rec = fb._to_record({"name": "N", "alias": "a", "zone": "z"}, + {"bvid": "BV1", "title": "t", "play": 100, "created": 0, "length": "1:00"}) + assert rec["platform"] == "bilibili" and rec["bvid"] == "BV1" and rec["vid"] == "BV1" + assert rec["duration_sec"] == 60 and rec["play"] == 100 + + +def test_fetch_creator_missing_sessdata_raises(): + with pytest.raises(fb.BilibiliError): + asyncio.run(fb.fetch_creator({"name": "N", "alias": "a", "zone": "z", "uid": 1}, "")) + + +def test_fetch_creator_missing_uid_raises(): + with pytest.raises(fb.BilibiliError): + asyncio.run(fb.fetch_creator({"name": "N", "alias": "a", "zone": "z"}, "fake-sess")) + + +class _FakeUser: + """A user.User stand-in whose get_videos returns canned pages (no network).""" + + def __init__(self, total): + self.total = total + + async def get_videos(self, ps, pn): + already = (pn - 1) * fb.PAGE_MAX + n = max(0, min(ps, self.total - already)) + vlist = [{"bvid": f"BV{pn}_{i}", "title": f"v{i}", "play": i, "created": 0} for i in range(n)] + return {"list": {"vlist": vlist}, "page": {"count": self.total}} + + +def test_fetch_creator_paginates_beyond_one_page(monkeypatch): + # 创作者有 120 条;num=120 必须翻 3 页(50+50+20)——旧版只取第一页会漏 70 条 + monkeypatch.setattr(fb.user, "User", lambda uid, credential=None: _FakeUser(120)) + vids = asyncio.run(fb.fetch_creator({"name": "N", "alias": "a", "zone": "z", "uid": 1}, "sess", num=120)) + assert len(vids) == 120 + + +def test_fetch_creator_stops_when_no_more(monkeypatch): + # 创作者只有 30 条,却要 100 → 只拿到 30,不死循环 + monkeypatch.setattr(fb.user, "User", lambda uid, credential=None: _FakeUser(30)) + vids = asyncio.run(fb.fetch_creator({"name": "N", "alias": "a", "zone": "z", "uid": 1}, "sess", num=100)) + assert len(vids) == 30 diff --git a/tests/test_features.py b/tests/test_features.py new file mode 100644 index 0000000..ff5fbb6 --- /dev/null +++ b/tests/test_features.py @@ -0,0 +1,45 @@ +"""features.extract / off_tag — pure metric+feature logic, no network.""" +import time + +from features import extract, off_tag + + +def _vid(**kw): + base = { + "title": "为什么天是蓝的", "description": "", + "created_ts": time.time() - 86400 * 10, + "play": 1_000_000, "duration_sec": 600, "comment": 2000, + "danmaku": 5000, "platform": "bilibili", + } + base.update(kw) + return base + + +def test_extract_basic_metrics(): + f = extract(_vid(), time.time()) + assert f["play"] == 1_000_000 + assert f["play_per_day"] > 0 + assert f["comment_per_10k"] > 0 + assert f["dur_bucket"] == "中 5-12min" + assert f["has_question"] is False + assert f["has_curiosity"] is True # 标题含「为什么」 + + +def test_off_format_bilibili_scans_description(): + # B 站扫 title + description;描述里的商单词应判偏题 + f = extract(_vid(title="正常标题", description="本视频与某品牌合作"), time.time()) + assert f["off_format"] is True + + +def test_off_format_youtube_title_only(): + # YouTube 简介塞满 hashtag,只扫 title;描述里的 sponsored 不算 + f = extract(_vid(platform="youtube", title="My normal video", description="sponsored by X"), time.time()) + assert f["off_format"] is False + f2 = extract(_vid(platform="youtube", title="sponsored haul", description=""), time.time()) + assert f2["off_format"] is True + + +def test_off_tag_labels(): + assert off_tag("和某品牌合作", "") == "商单" + assert off_tag("我的 vlog 日常", "") == "vlog" + assert off_tag("正经硬核科普", "") == "" diff --git a/tests/test_markers_validation.py b/tests/test_markers_validation.py new file mode 100644 index 0000000..8383f6f --- /dev/null +++ b/tests/test_markers_validation.py @@ -0,0 +1,51 @@ +"""shared_markers single-source + compare_form.off_tag + creators.validate_creators.""" +import compare_form +from creators import validate_creators +from shared_markers import OFF_MARKERS_COMPARE, OFF_MARKERS_FULL + + +def test_markers_intentional_divergence(): + # FULL(特征工程)含 livestream / 联名;COMPARE(对比报告)故意更保守、去掉它们。 + assert "livestream" in OFF_MARKERS_FULL["EN"] + assert "live stream" in OFF_MARKERS_FULL["EN"] + assert "livestream" not in OFF_MARKERS_COMPARE["EN"] + assert "联名" in OFF_MARKERS_FULL["商单"] + assert "联名" not in OFF_MARKERS_COMPARE["商单"] + + +def test_compare_off_tag_youtube_scans_title_only(): + yt = {"platform": "youtube", "title": "My video", "description": "interview podcast sponsored"} + assert compare_form.off_tag(yt) == "" # 只扫 title,描述里的词不算 + yt2 = {"platform": "youtube", "title": "sponsored haul", "description": ""} + assert compare_form.off_tag(yt2) == "EN" # title 命中 sponsored(EN 类) + + +def test_compare_off_tag_does_not_flag_livestream(): + v = {"platform": "youtube", "title": "secretly in a livestream challenge", "description": ""} + assert compare_form.off_tag(v) == "" # COMPARE 口径不把 livestream 当偏题 + + +def test_validate_creators_ok(): + good = [ + {"name": "A", "alias": "a", "zone": "知识", "platform": "bilibili", "uid": 1}, + {"name": "B", "alias": "b", "zone": "Ent", "platform": "youtube", "channel": "@b"}, + ] + assert validate_creators(good) == [] + + +def test_validate_creators_collects_all_problems(): + bad = [ + {"name": "X", "zone": "知识", "platform": "bilibili"}, # 缺 alias + {"name": "Y", "alias": "y", "zon": "typo", "platform": "bilibili"}, # 拼错 zon + 缺 zone + {"name": "Z", "alias": "y", "zone": "z", "platform": "youtube"}, # alias 重复 + youtube 缺 channel + ] + blob = " ".join(validate_creators(bad)) + assert "缺必填字段 'alias'" in blob + assert "未知字段 'zon'" in blob + assert "缺必填字段 'zone'" in blob + assert "重复" in blob + assert "channel" in blob + + +def test_validate_creators_empty(): + assert validate_creators([]) == ["CREATORS 必须是非空列表"]