Fix Bilibili pagination window drift, isolate chart failures, consolidate shared helpers

Bugs:
- fetch_bilibili.py: keep ps constant across pages. Shrinking ps on the last
  page shifts Bilibili's [(pn-1)*ps, pn*ps) window: num=120 re-fetched items
  41-60 and never fetched 101-120. Overshoot is trimmed by the existing
  [:num] slice. The pagination test's mock computed offsets from constant
  PAGE_MAX, which masked exactly this -- it now mirrors the real window
  semantics and asserts all 120 vids are distinct (old code fails this test).
- charts.py: run the three charts independently. cross_creator_meme.json is
  produced only by the opt-in compare_meme.py, so the standard --report run
  always died on chart 3 with FileNotFoundError and the whole charts step
  reported as failed. Missing input now skips just that chart with a note;
  empty cross_creator_form.json no longer crashes chart 1.

Hardening:
- compare_form.py: imported data with missing/null play or title no longer
  raises (KeyError/TypeError); empty creator files are skipped; bot5_med == 0
  no longer writes JSON-invalid Infinity into cross_creator_form.json.
- creator_profile.py: fmt_play crashed on play=None (n >= 1e8 TypeError);
  now uses the shared formatter with "-" fallback.
- diagnose.py: private-dims failures were swallowed by a blanket
  except-pass after the user uploaded their CSV; now logged to stderr.
- fetch_multi.py: the printed time span was taken from the play-sorted list
  (date of lowest-play video ~ date of highest-play video); now sorted by date.

Consolidation (anti-drift, same outputs -- formatter outputs are test-pinned):
- video_url was copy-pasted in app/build_report/export_data, fmt_play had
  four drifting variants (one with the None crash above); both now live in
  schema.py next to the record contract, charts keeps its short axis format
  via fmt_play(n, yi=1, wan=0).
- off_tag matching loop unified in features.py (markers parameter);
  compare_form keeps only its YouTube-title-only policy wrapper.
- Dead code: analyze_video._url (unused 4th URL builder), unused pathlib
  imports in 10 files, a branch in dim_title whose if/elif arms were
  identical, hand-rolled median replaced with statistics.median.

Left alone deliberately: diagnose zh_num/en_num (bilingual pair, local by
design), the two _worst() variants (different "na" semantics), and the
mixed naive/UTC created_iso between the two fetchers (changing it would
shift dates on already-fetched data; needs its own change).

Verified: mypy clean (25 files), 23/23 tests pass, compileall clean, all
touched modules import.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: HarryXin0919

scripts/analyze_video.py

@@ -66,14 +66,6 @@ def _safe(s):
     return re.sub(r"[^0-9A-Za-z_.-]", "_", str(s))[:80]
 
 
-def _url(v):
-    plat = (v.get("platform") or "").lower()
-    if plat == "youtube":
-        return f"https://www.youtube.com/watch?v={v.get('vid')}"
-    bv = v.get("bvid") or v.get("vid")
-    return f"https://www.bilibili.com/video/{bv}"
-
-
 # ————————————————————————— 下载(yt-dlp,只下开头) —————————————————————————
 def _sessdata():
     """B 站 cookie。只在内存里用,绝不打印/落盘到 git。"""

scripts/app.py

@@ -25,6 +25,7 @@
 from urllib.parse import urlparse, parse_qs
 
 import runtime                        # 收口「源码跑 vs 打包成 app 跑」的路径/子进程差异
+from schema import video_url          # 单一数据源:链接拼法只在 schema.py 维护
 
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")
@@ -107,17 +108,6 @@ def data_stats():
     return len(files), count
 
 
-def video_url(v):
-    """按平台拼出可点开的视频链接(给结果卡片用)。"""
-    p = (v.get("platform") or "bilibili").lower()
-    vid = v.get("bvid") or v.get("vid") or ""
-    if not vid:
-        return ""
-    if p == "youtube":
-        return f"https://www.youtube.com/watch?v={vid}"
-    return f"https://www.bilibili.com/video/{vid}"
-
-
 # ——————————————————————— 诊断页:选择器数据 + 单视频诊断 ———————————————————————
 def diag_list():
     """给诊断页的选择器:每个创作者 + 其视频(按播放降序)。"""

scripts/build_report.py

@@ -15,6 +15,7 @@
 import sys
 import time
 from runtime import DATA, REPORTS, IMG
+from schema import fmt_play, video_url    # 单一数据源:链接拼法/播放量格式只在 schema.py 维护
 
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")
@@ -36,19 +37,6 @@ def esc(s):
     return html.escape(str(s if s is not None else ""))
 
 
-def fmt_play(n):
-    """中文友好的播放量:1.2亿 / 814万 / 9532。"""
-    try:
-        n = float(n)
-    except (TypeError, ValueError):
-        return "-"
-    if n >= 1e8:
-        return f"{n/1e8:.2f}亿"
-    if n >= 1e4:
-        return f"{n/1e4:.1f}万"
-    return f"{n:,.0f}"
-
-
 def fmt_dur(sec):
     try:
         sec = int(sec)
@@ -58,16 +46,6 @@ def fmt_dur(sec):
     return f"{m}:{s:02d}"
 
 
-def video_url(v):
-    plat = (v.get("platform") or "bilibili").lower()
-    vid = v.get("bvid") or v.get("vid") or ""
-    if not vid:
-        return ""
-    if plat == "youtube":
-        return f"https://www.youtube.com/watch?v={vid}"
-    return f"https://www.bilibili.com/video/{vid}"
-
-
 def img_data_uri(path):
     try:
         b = path.read_bytes()

scripts/charts.py

@@ -8,7 +8,6 @@
 """
 import json
 import sys
-from pathlib import Path
 
 import matplotlib
 matplotlib.use("Agg")
@@ -18,6 +17,7 @@
     sys.stdout.reconfigure(encoding="utf-8")
 
 from creators import CREATORS
+from schema import fmt_play     # 播放量格式化收口在 schema.py;轴标签传 yi=1, wan=0 用短格式
 
 # 中文字体:按 Windows → macOS → Linux 顺序取第一个装了的,都没有才显示 □□□
 plt.rcParams["font.sans-serif"] = [
@@ -58,16 +58,11 @@ def anon(name):
     return ANON.get(name, "创作者")
 
 
-def fmt_play(n):
-    if n >= 1e8:
-        return f"{n/1e8:.1f}亿"
-    if n >= 1e4:
-        return f"{n/1e4:.0f}万"
-    return f"{n:,.0f}"
-
-
 def chart_form_spread():
     rows = json.loads((DATA / "cross_creator_form.json").read_text(encoding="utf-8"))
+    if not rows:
+        print("  ⏭ cross_creator_form.json 是空的(还没有创作者数据)—— 跳过这张图")
+        return
     rows.sort(key=lambda r: r["top5_med"])   # 从下往上递增,天花板最高的在顶部
     zone_of = {c["name"]: c.get("zone", "?") for c in CREATORS}
 
@@ -79,7 +74,7 @@ def chart_form_spread():
         ax.scatter(hi, i, color=HIT, s=110, zorder=3, edgecolors="white", linewidths=1.2)
         ax.text(hi * 1.3, i, f"{r['ratio']:.0f}×" if r["ratio"] >= 10 else f"{r['ratio']:.1f}×",
                 va="center", ha="left", fontsize=11, color="#333", fontweight="bold")
-        ax.text(lo * 0.72, i, fmt_play(lo), va="center", ha="right", fontsize=8.5, color="#888")
+        ax.text(lo * 0.72, i, fmt_play(lo, yi=1, wan=0), va="center", ha="right", fontsize=8.5, color="#888")
 
     ax.set_xscale("log")
     ax.set_yticks(range(len(rows)))
@@ -225,10 +220,22 @@ def chart_2nd_person_falsified():
 
 
 def main():
-    chart_form_spread()
-    chart_2nd_person_falsified()
-    chart_meme_falsified()
-    print("\n✅ 图已生成到 reports/img/")
+    # 三张图各自独立:缺哪个输入就跳过哪张,别让一张图(尤其是依赖可选步骤的)拖垮整个 charts 步骤
+    jobs = [
+        (chart_form_spread, "cross_creator_form.json(compare_form.py 生成)"),
+        (chart_2nd_person_falsified, "signal_scan.json(scan_signals.py 生成)"),
+        (chart_meme_falsified, "cross_creator_meme.json(可选步骤 compare_meme.py 才会生成)"),
+    ]
+    n_ok = 0
+    for fn, src in jobs:
+        try:
+            fn()
+            n_ok += 1
+        except FileNotFoundError:
+            print(f"  ⏭ 缺 {src} —— 跳过这张图")
+        except Exception as e:
+            print(f"  ⚠ 这张图没画成({type(e).__name__}: {e})—— 跳过")
+    print(f"\n✅ {n_ok} 张图已生成到 reports/img/")
 
 
 if __name__ == "__main__":

scripts/compare_form.py

@@ -8,14 +8,14 @@
 """
 import json
 import sys
-from pathlib import Path
 from statistics import median
 
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")
 
 from runtime import DATA           # 源码=仓库/data,打包成 app 时=用户数据目录
 from creators import CREATORS
+import features
 
 # 偏离核心形式的关键词 —— 单一数据源在 shared_markers.py(对比报告用保守口径版)。
 from shared_markers import OFF_MARKERS_COMPARE as OFF_MARKERS
@@ -24,14 +24,9 @@
 def off_tag(v):
     # YouTube description 几乎必含 brand list / hashtag(podcast/interview 等),全军 false positive;
     # 只扫 title。B 站 description 短而精,继续扫 title + description。
-    if (v.get("platform") or "bilibili") == "youtube":
-        s = (v.get("title", "") or "").lower()
-    else:
-        s = (v.get("title", "") + " " + v.get("description", "")).lower()
-    for label, ms in OFF_MARKERS.items():
-        if any(m.lower() in s for m in ms):
-            return label
-    return ""
+    # 匹配循环本身复用 features.off_tag,只是换成保守口径的标记表。
+    desc = "" if (v.get("platform") or "bilibili") == "youtube" else (v.get("description") or "")
+    return features.off_tag(v.get("title") or "", desc, OFF_MARKERS)
 
 
 def fmt(n):
@@ -46,32 +41,40 @@ def main():
             print(f"  ✗ 缺 {p.name}")
             continue
         vids = json.loads(p.read_text(encoding="utf-8"))
+        if not vids:
+            print(f"  ✗ {p.name} 是空的,跳过")
+            continue
         for v in vids:
             v["off"] = off_tag(v)
-        vids.sort(key=lambda x: -(x["play"] or 0))
+        # 导入的数据可能缺 play/title 字段或为 null —— 全部走 .get + 兜底,别在分析里崩
+        vids.sort(key=lambda x: -(x.get("play") or 0))
         top5, bot5 = vids[:5], vids[-5:]
-        top5_med = median([v["play"] for v in top5])
-        bot5_med = median([v["play"] for v in bot5])
-        ratio = top5_med / bot5_med if bot5_med else float("inf")
+        top5_med = median([v.get("play") or 0 for v in top5])
+        bot5_med = median([v.get("play") or 0 for v in bot5])
+        if not bot5_med:
+            # 尾部播放全是 0/缺失:头尾倍数没意义,而且 Infinity 进不了 JSON
+            print(f"  ✗ {c['name']} 尾部播放全是 0/缺失,头尾对比无意义,跳过")
+            continue
+        ratio = top5_med / bot5_med
 
         print("\n" + "=" * 74)
         print(f"■ {c['name']}   top5中位 {fmt(top5_med)} / bot5中位 {fmt(bot5_med)}  = {ratio:.0f}×")
         print("  🔴 TOP5(爆款):")
         for v in top5:
             tag = f"[{v['off']}]" if v["off"] else ""
-            print(f"     {fmt(v['play']):>12}  {tag}{v['title'][:34]}")
+            print(f"     {fmt(v.get('play')):>12}  {tag}{(v.get('title') or '')[:34]}")
         print("  🔵 BOTTOM5(翻车):")
         for v in bot5:
             tag = f"[{v['off']}]" if v["off"] else ""
-            print(f"     {fmt(v['play']):>12}  {tag}{v['title'][:34]}")
+            print(f"     {fmt(v.get('play')):>12}  {tag}{(v.get('title') or '')[:34]}")
 
         rows.append({
             "creator": c["name"],
             "top5_med": top5_med, "bot5_med": bot5_med, "ratio": round(ratio, 1),
             "off_in_top5": sum(1 for v in top5 if v["off"]),
             "off_in_bot5": sum(1 for v in bot5 if v["off"]),
-            "top5": [{"play": v["play"], "off": v["off"], "title": v["title"]} for v in top5],
-            "bot5": [{"play": v["play"], "off": v["off"], "title": v["title"]} for v in bot5],
+            "top5": [{"play": v.get("play"), "off": v["off"], "title": v.get("title") or ""} for v in top5],
+            "bot5": [{"play": v.get("play"), "off": v["off"], "title": v.get("title") or ""} for v in bot5],
         })
 
     (DATA / "cross_creator_form.json").write_text(

scripts/compare_meme.py

@@ -17,7 +17,6 @@
 import sys
 import re
 import time
-from pathlib import Path
 from collections import Counter
 
 from bilibili_api import comment, Credential

scripts/creator_profile.py

@@ -21,7 +21,6 @@
 import json
 import sys
 import time
-from pathlib import Path
 from statistics import median
 
 if hasattr(sys.stdout, "reconfigure"):
@@ -31,6 +30,7 @@
 from creators import CREATORS
 from features import extract
 from scan_signals import spearman      # 复用秩相关,避免重复实现
+from schema import fmt_play            # 播放量格式化收口在 schema.py(旧本地版遇 None 会崩)
 
 NOW = time.time()
 MATURE_DAYS = 30        # 发布满 30 天才算"播放稳定",用于趋势
@@ -111,14 +111,6 @@ def trend(vids):
             "gap_early": gap_early, "gap_recent": gap_recent, "cadence": cad}
 
 
-def fmt_play(n):
-    if n >= 1e8:
-        return f"{n/1e8:.2f}亿"
-    if n >= 1e4:
-        return f"{n/1e4:.1f}万"
-    return str(n)
-
-
 def fmt_dur(sec):
     return f"{sec//60}分{sec%60:02d}秒"
 

scripts/diagnose.py

@@ -18,15 +18,16 @@
 import json
 import re
 import sys
-from pathlib import Path
+from statistics import median
+from typing import Any
 
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")
 
 from runtime import DATA           # 源码=仓库/data,打包成 app 时=用户数据目录
 
 # —— 懒加载 + 缓存:同一进程里只读一次盘 ——
-_CACHE = {}
+_CACHE: dict[str, Any] = {}
 
 
 def _load(name):
@@ -100,12 +101,9 @@ def _subtitle_of(vid):
 
 # ————————————————————————— 小工具 —————————————————————————
 def _median(xs):
-    xs = sorted(x for x in xs if x is not None)
-    n = len(xs)
-    if not n:
-        return None
-    m = n // 2
-    return xs[m] if n % 2 else (xs[m - 1] + xs[m]) / 2
+    """statistics.median + 本项目约定:先滤掉 None,空列表返回 None 而不是抛异常。"""
+    xs = [x for x in xs if x is not None]
+    return median(xs) if xs else None
 
 
 def zh_num(n):
@@ -246,13 +244,8 @@ def dim_title(alias, v):
     hit_len = _median([len(h.get("title") or "") for h in vids[:n_hit]]) or tlen
 
     metrics, levels = [], []
-    # 长度:跟自己爆款比
-    if hit_len and (0.7 * hit_len <= tlen <= 1.4 * hit_len):
-        lvl = "good"
-    elif hit_len and tlen > 1.4 * hit_len:
-        lvl = "ok"
-    else:
-        lvl = "ok"
+    # 长度:跟自己爆款比 —— 在爆款长度 0.7~1.4 倍区间内算 good,出区间只是 ok(不警告)
+    lvl = "good" if hit_len and 0.7 * hit_len <= tlen <= 1.4 * hit_len else "ok"
     levels.append(lvl)
     metrics.append({"name": {"zh": "长度", "en": "Length"},
                     "value": f"{tlen}", "ref": {"zh": f"你爆款约 {hit_len:.0f} 字", "en": f"your hits ~{hit_len:.0f} chars"},
@@ -418,14 +411,15 @@ def diagnose_video(alias, vid):
         dims.append(sub)
     dims = [d for d in dims if d]
 
-    # 私有后台数据(用户上传 CSV 后才有):完播率 / 点击率 —— 有就插进来一起算
+    # 私有后台数据(用户上传 CSV 后才有):完播率 / 点击率 —— 有就插进来一起算。
+    # 失败不挡公开维度,但要在控制台留痕:别让用户刚上传的 CSV 无声消失。
     try:
         import import_private
         priv = import_private.load_private(alias).get(v.get("vid") or v.get("bvid"))
         if priv:
             dims += import_private.private_dims(v, priv.get("metrics", {}))
-    except Exception:
-        pass
+    except Exception as e:
+        print(f"  ⚠ 私有数据维度没加上(alias={alias}):{type(e).__name__}: {e}", file=sys.stderr)
 
     real = [d for d in dims if d["level"] in ("good", "ok", "warn")]
     n_good = sum(1 for d in real if d["level"] == "good")

scripts/export_data.py

@@ -13,29 +13,18 @@
 import csv
 import json
 import sys
-from pathlib import Path
 
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")
 
 from runtime import DATA           # 源码=仓库/data,打包成 app 时=用户数据目录
+from schema import video_url       # 单一数据源:链接拼法只在 schema.py 维护
 
 # CSV 里放哪些列、按什么顺序(挑人看得懂、Excel 排序有用的;长描述留在 JSON 里不塞 CSV)
 CSV_COLS = ["creator", "platform", "zone", "title", "play", "comment", "like",
             "danmaku", "duration_sec", "length", "published", "url", "cover_url"]
 
 
-def video_url(v):
-    """按平台拼出可点开的视频链接。"""
-    p = (v.get("platform") or "bilibili").lower()
-    vid = v.get("bvid") or v.get("vid") or ""
-    if not vid:
-        return ""
-    if p == "youtube":
-        return f"https://www.youtube.com/watch?v={vid}"
-    return f"https://www.bilibili.com/video/{vid}"
-
-
 def hhmmss(sec):
     """秒 → 人看的 时长(M:SS 或 H:MM:SS)。"""
     sec = int(sec or 0)

scripts/features.py

@@ -24,10 +24,11 @@
          "测评", "评测", "实测", "上手", "教你"]                          # 教程/实用
 
 
-def off_tag(title, desc=""):
-    """命中任一偏离标记就返回标签,否则空串。宁紧勿松,靠人眼复核标题。"""
+def off_tag(title, desc="", markers=OFF_MARKERS):
+    """命中任一偏离标记就返回标签,否则空串。宁紧勿松,靠人眼复核标题。
+    markers 可换口径(compare_form.py 传保守版 OFF_MARKERS_COMPARE,匹配循环只写这一处)。"""
     s = (title + " " + desc).lower()
-    for label, ms in OFF_MARKERS.items():
+    for label, ms in markers.items():
         if any(m.lower() in s for m in ms):
             return label
     return ""