"""douyin.meta 的离线测试：标题/标签拆分、记录追加与幂等。不联网。""" from __future__ import annotations # 移到 scripts/ tests/ 后，直接运行时脚本目录成了 sys.path[0]， # 把项目根加回来，import douyin / pipeline 才找得到。 import sys as _sys from pathlib import Path as _Path _sys.path.insert(0, str(_Path(__file__).resolve().parent.parent)) import json import tempfile from pathlib import Path from douyin.meta import RecordLog, split_desc def te(desc, *names): """按真实位置构造 text_extra —— 偏移量算出来，别手数。""" out = [] for n in names: i = desc.index("#" + n) out.append({"type": 1, "hashtag_name": n, "hashtag_id": "1", "start": i, "end": i + len(n) + 1}) return out def te_broken(*names): """偏移量是垃圾（越界）的 text_extra。""" return [{"type": 1, "hashtag_name": n, "hashtag_id": "1", "start": 999, "end": 1200} for n in names] def test_split_basic(): desc = "是谁教他这样平账的？？？ #沙雕动画 #原创动画" title, tags = split_desc(desc, te(desc, "沙雕动画", "原创动画")) assert title == "是谁教他这样平账的？？？", repr(title) assert tags == ["沙雕动画", "原创动画"], tags def test_split_adjacent_no_space(): # 实测抖音有 "#四川话#方言" 这种连着写的 title, tags = split_desc("老板说话#四川话#方言", None) assert title == "老板说话", repr(title) assert tags == ["四川话", "方言"], tags def test_split_tag_missing_from_text_extra(): """text_extra 会漏标签（实测 12 条里有 2 条），tags 必须以 desc 为准。""" desc = "太搞笑了 #搞笑配音 #专治不开心 #非洲搞笑视频" title, tags = split_desc(desc, te(desc, "搞笑配音", "专治不开心")) assert tags == ["搞笑配音", "专治不开心", "非洲搞笑视频"], tags assert title == "太搞笑了", repr(title) def test_split_tag_only_in_text_extra(): """反过来，desc 里没写但 text_extra 有的也不能丢。""" _, tags = split_desc("光秃秃的标题", te_broken("隐藏标签")) assert tags == ["隐藏标签"], tags def test_split_tag_in_middle(): title, tags = split_desc("前面 #中间 后面", None) assert title == "前面 后面", repr(title) assert tags == ["中间"], tags def test_split_no_tags(): title, tags = split_desc(" 纯标题没有标签 ", None) assert title == "纯标题没有标签", repr(title) assert tags == [], tags def test_split_empty(): assert split_desc("", None) == ("", []) assert split_desc(None, None) == ("", []) def test_split_dedupes_tags(): _, tags = split_desc("标题 #方言 #方言 #四川话", None) assert tags == ["方言", "四川话"], tags def test_split_bad_offsets_dont_corrupt(): """偏移量对不上时必须退回正则，绝不能把标题切坏。""" title, tags = split_desc("标题正文 #标签", te_broken("标签")) assert title == "标题正文", repr(title) assert tags == ["标签"], tags def test_split_keeps_hash_inside_word(): """偏移量可信时不该误伤正文里的 #（如 C# 教程）——这正是用偏移的理由。""" desc = "C# 入门教程 #编程" title, tags = split_desc(desc, te(desc, "编程")) assert title == "C# 入门教程", repr(title) assert tags == ["编程"], tags def test_split_lone_hash_is_not_a_tag(): """# 后面必须跟非空白才算标签，所以 "C# 教程" 里的 # 不会被当标签。""" title, tags = split_desc("C# 入门教程 #编程", None) assert tags == ["编程"], tags assert title == "C# 入门教程", repr(title) def test_split_repeated_tagname_in_body_is_lossy(): """没有可信偏移、且标签名以 #名字 的形式也出现在正文里时，兜底会连正文 一起删。这是正则兜底的已知代价，钉住它免得以后误以为无损。""" title, tags = split_desc("聊聊#教程 这件事 #教程", None) assert tags == ["教程"], tags assert title == "聊聊 这件事", repr(title) # 正文里那个 #教程 也没了 # 有可信偏移时就不会误伤——只挖服务端标注的那一处 desc = "聊聊#教程 这件事 #教程" i = desc.rindex("#教程") ok = [{"type": 1, "hashtag_name": "教程", "start": i, "end": i + 3}] title2, _ = split_desc(desc, ok) assert title2 == "聊聊#教程 这件事", repr(title2) def test_split_case_insensitive_dedupe(): """抖音把 text_extra 里的标签归一化成小写，desc 里却是原始大小写。 并集必须按大小写不敏感去重，否则同一个标签会存两份（实测踩到）。""" desc = "教学 #一站式AI创作工作台 #国产AI" te = [{"type": 1, "hashtag_name": "一站式ai创作工作台", "start": -1, "end": -1}, {"type": 1, "hashtag_name": "国产ai", "start": -1, "end": -1}] title, tags = split_desc(desc, te) # 保留 desc 里的原始大小写，不是 text_extra 的小写版 assert tags == ["一站式AI创作工作台", "国产AI"], tags assert title == "教学", repr(title) def test_split_drops_at_mention_pseudo_tag(): """正文里手打的 # 会被正则捕到，但带 @ 的明显是提及不是标签。 实测：desc 里的 "#(提示词出自@青木"，抖音自己也没把它当标签。""" desc = "示例#正经标签 #(提示词出自@青木 文本#另一个" title, tags = split_desc(desc, None) assert tags == ["正经标签", "另一个"], tags # 它不进 tags，但仍然从标题里剥掉——作者本来就是当标签打的 assert title == "示例 文本", repr(title) def test_split_strips_edge_punctuation(): title, tags = split_desc("标题 #（括号标签） #正常", None) assert tags == ["括号标签", "正常"], tags # 标题里那对括号也得跟着标签一起剥掉——剥标题用的是 desc 原文， # 不是清洗过的标签名，否则会匹配不上而留在标题里 assert title == "标题", repr(title) def test_recordlog_roundtrip_and_resume(): with tempfile.TemporaryDirectory() as d: p = Path(d) / "records.jsonl" log = RecordLog(p) assert len(log) == 0 assert log.write({"aweme_id": "111", "title": "甲", "raw": {"x": 1}}) is True assert log.write({"aweme_id": "222", "title": "乙", "raw": {"x": 2}}) is True assert "111" in log and "222" in log and "333" not in log # 重开：应该扫出已有的两条 log2 = RecordLog(p) assert len(log2) == 2, len(log2) assert "111" in log2 and "222" in log2 # 重复写入被幂等挡掉 assert log2.write({"aweme_id": "111", "title": "改过的"}) is False assert len(p.read_text(encoding="utf-8").strip().splitlines()) == 2 # aweme_id 必须是每行第一个字段（续跑扫描的契约） for line in p.read_text(encoding="utf-8").splitlines(): assert line.startswith('{"aweme_id":'), line[:40] rows = [json.loads(x) for x in p.read_text(encoding="utf-8").splitlines()] assert rows[0]["title"] == "甲" and rows[0]["raw"] == {"x": 1} def test_recordlog_survives_broken_tail(): """进程被 Ctrl-C 打断可能留下半行，扫描不能因此崩掉。""" with tempfile.TemporaryDirectory() as d: p = Path(d) / "records.jsonl" p.write_text('{"aweme_id":"111","title":"完整"}\n{"aweme_id":"222","tit', encoding="utf-8") log = RecordLog(p) assert "111" in log and "222" in log # 半行的 id 也认，照样跳过 log.write({"aweme_id": "333", "title": "新的"}) lines = p.read_text(encoding="utf-8").splitlines() # 补写前必须先断行，不能跟半行黏成一行 assert lines[-1].startswith('{"aweme_id":"333"'), lines[-1][:40] assert json.loads(lines[-1])["title"] == "新的" def test_recordlog_no_ascii_escaping(): """中文必须原样存，不能转成 \\uXXXX——否则文件体积翻几倍且不可读。""" with tempfile.TemporaryDirectory() as d: p = Path(d) / "r.jsonl" RecordLog(p).write({"aweme_id": "1", "title": "四川方言"}) assert "四川方言" in p.read_text(encoding="utf-8") def main() -> int: tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] failed = 0 for t in tests: try: t() print(f" ok {t.__name__}") except AssertionError as exc: failed += 1 print(f" FAIL {t.__name__}: {exc}") print(f"\n{len(tests) - failed}/{len(tests)} 通过") return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())