# -*- coding: utf-8 -*- """抖音采集编排 —— 不碰网络，把 douyin_api 整个换掉。 验的是编排本身：产物落在正确的目录、文件名是 store 那套、以及**作品列表被挡时的退化** （用已知 aweme_id 逐条刷新）—— 那条退化路径决定了今天这个功能是「完全没用」还是 「已知作品还能看」。 """ import json import pytest from api.monitor import douyin_api, douyin_fetch def _video(aweme_id: str, likes: str = "1") -> dict: return { "aweme_id": aweme_id, "title": f"title-{aweme_id}", "aweme_url": f"https://www.douyin.com/video/{aweme_id}", "cover_url": "", "aweme_type": "0", "create_time": 1790574515, "creator_hash": "hash", "nickname": "博主", "liked_count": likes, "comment_count": "0", "collected_count": "0", "share_count": "0", } def _comment(aweme_id: str, index: int) -> dict: return { "comment_id": f"c{index}", "aweme_id": aweme_id, "content": f"评论{index}", "nickname": "路人", "creator_hash": "h2", "create_time": 1790574600, "like_count": "0", "sub_comment_count": "0", "parent_comment_id": "0", } class _Target: def __init__(self, external_id: str) -> None: self.external_id = external_id def _read(path): return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line] async def _collect(tmp_path, **overrides): kwargs = dict( platform="dy", mode="creator", limit=20, want_comments=True, comment_limit=20, targets=[_Target("MS4w-sec")], cookie="sessionid=x", ) kwargs.update(overrides) return await douyin_fetch.collect(tmp_path, **kwargs) class TestHappyPath: @pytest.mark.asyncio async def test_writes_the_layout_ingest_expects(self, monkeypatch, tmp_path): async def fake_videos(sec_user_id, count=20, *, cookie=""): return [_video("111"), _video("222")] async def fake_comments(aweme_id, count=20, *, cookie=""): return [_comment(aweme_id, 1)] monkeypatch.setattr(douyin_api, "author_videos", fake_videos) monkeypatch.setattr(douyin_api, "video_comments", fake_comments) result = await _collect(tmp_path) assert result["notes"] == 2 assert result["comments"] == 2 assert result["errors"] == [] # 目录名必须是 douyin（不是平台 id dy）—— ingest 找文件用的是同一个来源。 jsonl_dir = tmp_path / "douyin" / "jsonl" assert jsonl_dir.is_dir() contents = list(jsonl_dir.glob("*_contents_*.jsonl")) comments = list(jsonl_dir.glob("*_comments_*.jsonl")) assert len(contents) == 1 assert len(comments) == 1 notes = _read(contents[0]) assert [n["aweme_id"] for n in notes] == ["111", "222"] # 键名照抄 store/douyin —— ingest 靠这个读出来。 assert notes[0]["aweme_url"] == "https://www.douyin.com/video/111" # 秒，不是毫秒；换算交给 adapters。 assert notes[0]["create_time"] == 1790574515 assert _read(comments[0])[0]["aweme_id"] == "111" @pytest.mark.asyncio async def test_the_comment_file_exists_even_without_comments(self, monkeypatch, tmp_path): """评论文件必须建出来。 ingest 靠「文件在不在」区分「这一轮没评论」和「这一轮什么都没抓到」—— 两种情况的含义完全不同。 """ async def fake_videos(sec_user_id, count=20, *, cookie=""): return [_video("111")] monkeypatch.setattr(douyin_api, "author_videos", fake_videos) await _collect(tmp_path, want_comments=False) assert len(list((tmp_path / "douyin" / "jsonl").glob("*_comments_*.jsonl"))) == 1 @pytest.mark.asyncio async def test_duplicate_works_are_written_once(self, monkeypatch, tmp_path): async def fake_videos(sec_user_id, count=20, *, cookie=""): return [_video("111"), _video("111")] monkeypatch.setattr(douyin_api, "author_videos", fake_videos) result = await _collect(tmp_path, want_comments=False) assert result["notes"] == 1 class TestNoteMode: @pytest.mark.asyncio async def test_a_work_target_is_fetched_by_detail_not_by_creator_list( self, monkeypatch, tmp_path ): """作品模式的目标**本身就是作品 id**，不能拿它当博主的 sec_uid 去查列表。 走错了会必然失败，而且失败原因很难看懂（接口说你没登录/不是浏览器）—— 「粘贴作品链接的监控」今天本来是能用的，别让它因为这一处走错而废掉。 """ async def must_not_be_called(*args, **kwargs): raise AssertionError("作品模式不该去拉博主的作品列表") async def fake_detail(aweme_id, *, cookie=""): return _video(aweme_id) monkeypatch.setattr(douyin_api, "author_videos", must_not_be_called) monkeypatch.setattr(douyin_api, "video_detail", fake_detail) result = await _collect( tmp_path, mode="note", want_comments=False, targets=[_Target("111"), _Target("222")], ) assert result["notes"] == 2 assert result["errors"] == [] notes = _read(list((tmp_path / "douyin" / "jsonl").glob("*_contents_*.jsonl"))[0]) assert [n["aweme_id"] for n in notes] == ["111", "222"] @pytest.mark.asyncio async def test_a_broken_work_does_not_lose_the_others(self, monkeypatch, tmp_path): async def flaky(aweme_id, *, cookie=""): if aweme_id == "222": raise douyin_api.DouyinApiError("作品已被删除") return _video(aweme_id) monkeypatch.setattr(douyin_api, "video_detail", flaky) result = await _collect( tmp_path, mode="note", want_comments=False, targets=[_Target("111"), _Target("222")], ) assert result["notes"] == 1 assert any("222" in error for error in result["errors"]) class TestDegradation: """作品列表被挡时的行为 —— 决定了这个功能今天有没有用。""" @pytest.mark.asyncio async def test_falls_back_to_refreshing_known_works(self, monkeypatch, tmp_path): async def blocked(sec_user_id, count=20, *, cookie=""): raise douyin_api.DouyinApiError("接口返回了空内容") refreshed = [] async def fake_detail(aweme_id, *, cookie=""): refreshed.append(aweme_id) return _video(aweme_id, likes="9") monkeypatch.setattr(douyin_api, "author_videos", blocked) monkeypatch.setattr(douyin_api, "video_detail", fake_detail) result = await _collect( tmp_path, want_comments=False, known_aweme_ids=["999", "888"] ) assert refreshed == ["999", "888"] assert result["notes"] == 2 # 但错误照样报出来 —— 这一轮是「部分可用」，不是「一切正常」，别粉饰。 assert any("作品列表失败" in error for error in result["errors"]) @pytest.mark.asyncio async def test_known_works_are_not_refetched_for_every_target(self, monkeypatch, tmp_path): """退化路径不能每个目标都把同一批已知作品再刷一遍。 任务有多个目标时那会让同一件作品在一轮里出现两次，而指标快照的唯一键是 (task_id, note_id, run_id) —— 第二次插入直接撞键，整个 run 崩掉（踩过： Duplicate entry for key 'uq_note_metric'）。 """ async def blocked(sec_user_id, count=20, *, cookie=""): raise douyin_api.DouyinApiError("接口返回了空内容") calls = [] async def fake_detail(aweme_id, *, cookie=""): calls.append(aweme_id) return _video(aweme_id) monkeypatch.setattr(douyin_api, "author_videos", blocked) monkeypatch.setattr(douyin_api, "video_detail", fake_detail) result = await _collect( tmp_path, want_comments=False, targets=[_Target("sec-a"), _Target("sec-b")], known_aweme_ids=["999"], ) assert calls == ["999"], "同一件已知作品只该刷一次，而不是每个目标一次" assert result["notes"] == 1 @pytest.mark.asyncio async def test_nothing_at_all_still_reports_the_reason(self, monkeypatch, tmp_path): async def blocked(sec_user_id, count=20, *, cookie=""): raise douyin_api.DouyinApiError("接口返回了空内容") monkeypatch.setattr(douyin_api, "author_videos", blocked) result = await _collect(tmp_path, want_comments=False) assert result["notes"] == 0 assert result["errors"] # 产物仍然写出来（空的），让调用方去判断这是失败而不是「这个博主没作品」。 assert (tmp_path / "douyin" / "jsonl").is_dir() @pytest.mark.asyncio async def test_a_failing_comment_fetch_does_not_lose_the_work(self, monkeypatch, tmp_path): async def fake_videos(sec_user_id, count=20, *, cookie=""): return [_video("111")] async def broken_comments(aweme_id, count=20, *, cookie=""): raise douyin_api.DouyinApiError("评论接口抽风") monkeypatch.setattr(douyin_api, "author_videos", fake_videos) monkeypatch.setattr(douyin_api, "video_comments", broken_comments) result = await _collect(tmp_path) # 评论拿不到是小事，作品不能跟着丢。 assert result["notes"] == 1 assert any("评论失败" in error for error in result["errors"])