# -*- coding: utf-8 -*- """抖音采集编排 —— 不碰网络，把 douyin_api 整个换掉。 验的是编排本身：产物落在正确的目录、文件名是 store 那套、以及**作品列表被挡时的退化** （用已知 aweme_id 逐条刷新）—— 那条退化路径决定了今天这个功能是「完全没用」还是 「已知作品还能看」。 """ import json import pytest from api.monitor import douyin_api, douyin_fetch def _profile(**overrides) -> dict: profile = { "creator_hash": "hash", "nickname": "博主", "unique_id": "abc", "fans": 12000, "total_favorited": 83000, "works": 42, "following": 7, } profile.update(overrides) return profile @pytest.fixture(autouse=True) def fake_profile(monkeypatch): """每个用例都挡住「问博主资料」这一跳。 它是附加信息，不在任何一条编排路径上，但真发出去就会去连 9222 那个浏览器 —— 于是所有 creator 用例都会多出一次连接失败、并把 ``errors`` 弄脏。想验它自己的 用例再单独覆盖这个 fixture。 """ async def _profile_call(sec_user_id, *, cookie=""): return _profile() monkeypatch.setattr(douyin_api, "author_profile", _profile_call) 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"]) class TestCreatorProfile: """博主的**账号级**指标 —— 粉丝 / 总获赞 / 作品数。 作品列表给不了这个东西：它说的是一件作品涨了多少赞，不是这个人整个账号的粉丝 在涨还是在掉。单独问一次资料接口。 """ @staticmethod def _profiles(tmp_path): files = list((tmp_path / "douyin" / "jsonl").glob("*_profile_*.jsonl")) assert len(files) == 1 return _read(files[0]) @pytest.mark.asyncio async def test_the_profile_lands_in_the_run_dir(self, monkeypatch, tmp_path): 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 self._profiles(tmp_path) == [_profile()] @pytest.mark.asyncio async def test_the_profile_is_keyed_by_the_same_hash_as_the_works( self, monkeypatch, tmp_path ): """**这条是关键。** 快照表的唯一键是 (任务, creator_hash, 轮次)，而界面上是按 作品的 creator_hash 归组去查它的。两边只要差一个字符，粉丝数就永远查不出来 —— 而且是静默的：表里有数据，界面上什么都没有。 """ async def fake_videos(sec_user_id, count=20, *, cookie=""): return [_video("111")] # 作品带的哈希是 "hash" # 资料接口自己算出来的是另一个值（比如它那边 uid 缺字段、只能拿 sec_uid 算）。 async def off_hash_profile(sec_user_id, *, cookie=""): return _profile(creator_hash="另一个哈希") monkeypatch.setattr(douyin_api, "author_videos", fake_videos) monkeypatch.setattr(douyin_api, "author_profile", off_hash_profile) await _collect(tmp_path, want_comments=False) # 以作品为准：作品才是界面上的行，快照必须挂在能查到它的那个键上。 assert self._profiles(tmp_path)[0]["creator_hash"] == "hash" @pytest.mark.asyncio async def test_a_profile_without_an_identity_is_dropped(self, monkeypatch, tmp_path): """哈希都算不出来的快照，落下去只会是一条谁也查不到的垃圾。""" async def fake_videos(sec_user_id, count=20, *, cookie=""): return [] async def anonymous_profile(sec_user_id, *, cookie=""): return _profile(creator_hash="") monkeypatch.setattr(douyin_api, "author_videos", fake_videos) monkeypatch.setattr(douyin_api, "author_profile", anonymous_profile) result = await _collect(tmp_path, want_comments=False) assert self._profiles(tmp_path) == [] assert any("身份标识" in error for error in result["errors"]) @pytest.mark.asyncio async def test_a_failing_profile_does_not_lose_the_works(self, monkeypatch, tmp_path): """附加信息拿不到，这一轮采到的作品不能跟着判成失败。""" async def fake_videos(sec_user_id, count=20, *, cookie=""): return [_video("111")] async def broken_profile(sec_user_id, *, cookie=""): raise douyin_api.DouyinApiError("资料接口抽风") monkeypatch.setattr(douyin_api, "author_videos", fake_videos) monkeypatch.setattr(douyin_api, "author_profile", broken_profile) result = await _collect(tmp_path, want_comments=False) assert result["notes"] == 1 assert self._profiles(tmp_path) == [] assert any("资料失败" in error for error in result["errors"]) @pytest.mark.asyncio async def test_a_work_target_never_asks_for_a_profile(self, monkeypatch, tmp_path): """作品模式的目标是一件作品，没有「这个博主是谁」可问 —— 不该白发一个请求。""" asked = [] async def fake_detail(aweme_id, *, cookie=""): return _video(aweme_id) async def recording_profile(sec_user_id, *, cookie=""): asked.append(sec_user_id) return _profile() monkeypatch.setattr(douyin_api, "video_detail", fake_detail) monkeypatch.setattr(douyin_api, "author_profile", recording_profile) await _collect(tmp_path, mode="note", want_comments=False) assert asked == [] # 文件仍然建出来（空的）：事后翻 run 目录能看出「这次根本没问过」。 assert self._profiles(tmp_path) == []