whale-fall-code
2bb83f3685
App 端免签名接口（设备注册 + detail + CDN 直链），一体化流水线 下载→抽音→传COS→记录→删本地，AI 作品四档判定，四份文档： README（使用指南）、PROGRESS（批次进度）、RE（逆向全过程）、AI（判定规则）。 Co-Authored-By: Claude Code <noreply@anthropic.com>
218 lines
8.9 KiB
Python
218 lines
8.9 KiB
Python
"""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())
|