# -*- coding: utf-8 -*- """作品栏里的**博主**：备注（他到底是谁）与账号级指标（他现在多大）。 按 creator_hash 分组、显示 creator_name，两样都认不出人：一个是哈希，一个是平台昵称。 备注是人自己起的名字。账号级指标则是作品列表给不了的东西 —— 作品说的是"这条涨了多少赞"， 粉丝数说的是"这个人整个账号在涨还是在掉"。 """ import httpx import pytest import pytest_asyncio from api.main import app from api.monitor import db as monitor_db from api.monitor.models import ( MODE_CREATOR, MonitorCreatorStat, MonitorNote, MonitorTask, ) CREATOR_HASH = "hash-a" NICKNAME = "张三" @pytest_asyncio.fixture async def client(tmp_path): monitor_db.set_sqlite_path(tmp_path / "monitor.db") await monitor_db.init_db() transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client: yield http_client await monitor_db.dispose_engine() async def _seed(platform: str = "xhs", task_name: str = "任务") -> int: """一个任务 + 一条作品，博主固定用 CREATOR_HASH / NICKNAME。""" async with monitor_db.get_session() as session: task = MonitorTask( name=task_name, platform=platform, mode=MODE_CREATOR, enabled=True, interval_minutes=60, max_notes_count=20, enable_comments=False, max_comments_count=50, run_timeout_seconds=3600, notify_enabled=False, created_at=0, updated_at=0, ) session.add(task) await session.flush() session.add( MonitorNote( task_id=task.id, note_id=f"{platform}-n1", title="作品", note_url="", cover="", creator_hash=CREATOR_HASH, creator_name=NICKNAME, source_kind="", published_at=None, first_seen_run_id=1, first_seen_at=0, last_seen_run_id=1, last_seen_at=0, ) ) return task.id async def _notes(client, platform: str = "xhs"): return (await client.get("/api/monitor/notes", params={"platform": platform})).json()["notes"] async def _set_alias(client, alias: str, platform: str = "xhs", creator_hash=CREATOR_HASH): return await client.put( f"/api/monitor/creators/{creator_hash}", params={"platform": platform}, json={"alias": alias}, ) class TestCreatorAlias: @pytest.mark.asyncio async def test_notes_start_without_an_alias(self, client): await _seed() assert (await _notes(client))[0]["creator_alias"] == "" @pytest.mark.asyncio async def test_an_alias_comes_back_with_the_notes(self, client): await _seed() response = await _set_alias(client, "竞品A") assert response.status_code == 200 note = (await _notes(client))[0] assert note["creator_alias"] == "竞品A" # 备注是**叠加**在昵称之上的，不是替换 —— 昵称仍然是有用的对照。 assert note["creator_name"] == NICKNAME @pytest.mark.asyncio async def test_an_alias_is_shared_across_tasks(self, client): """同一个博主出现在两个任务里，备注只该填一次。 creator_hash 对同一个 uid 是稳定的，所以键取 (platform, creator_hash) 而不是 按任务存 —— 否则每加一个任务都要重新认一遍人。 """ await _seed(task_name="任务甲") await _seed(task_name="任务乙") await _set_alias(client, "竞品A") for note in await _notes(client): assert note["creator_alias"] == "竞品A" @pytest.mark.asyncio async def test_the_alias_does_not_leak_to_another_platform(self, client): """同一个哈希在另一个平台上是另一个（或同一个）人 —— 别串味。""" await _seed(platform="xhs") await _seed(platform="dy") await _set_alias(client, "小红书那边的", platform="xhs") assert (await _notes(client, "xhs"))[0]["creator_alias"] == "小红书那边的" assert (await _notes(client, "dy"))[0]["creator_alias"] == "" @pytest.mark.asyncio async def test_an_empty_alias_clears_it(self, client): await _seed() await _set_alias(client, "竞品A") await _set_alias(client, "") assert (await _notes(client))[0]["creator_alias"] == "" @pytest.mark.asyncio async def test_an_alias_is_trimmed(self, client): await _seed() await _set_alias(client, " 竞品A ") assert (await _notes(client))[0]["creator_alias"] == "竞品A" async def _add_stat( task_id: int, run_id: int, fans: int | None, *, total_favorited: int | None = 83000, works: int | None = 42, creator_hash: str = CREATOR_HASH, captured_at: int = 1, ) -> None: async with monitor_db.get_session() as session: session.add( MonitorCreatorStat( task_id=task_id, run_id=run_id, creator_hash=creator_hash, nickname=NICKNAME, fans=fans, total_favorited=total_favorited, works_count=works, following=7, captured_at=captured_at, ) ) class TestCreatorStats: """账号级指标跟着作品一起返回 —— 界面上是按博主归组的，为了一个组头再发一轮请求 没道理。""" @pytest.mark.asyncio async def test_without_snapshots_the_fields_are_null(self, client): """null 而不是 0：0 会显示成「粉丝 0」，而事实是"还没采到"。""" await _seed() note = (await _notes(client))[0] assert note["creator_fans"] is None assert note["creator_total_favorited"] is None assert note["creator_works"] is None @pytest.mark.asyncio async def test_a_snapshot_shows_up_on_every_work_of_that_creator(self, client): task_id = await _seed() await _add_stat(task_id, run_id=1, fans=12000) for note in await _notes(client): assert note["creator_fans"] == 12000 assert note["creator_total_favorited"] == 83000 assert note["creator_works"] == 42 @pytest.mark.asyncio async def test_the_latest_run_wins(self, client): """一轮一条，所以总会有好几条 —— 给界面的必须是最近那条。""" task_id = await _seed() await _add_stat(task_id, run_id=1, fans=12000) await _add_stat(task_id, run_id=2, fans=12300) assert (await _notes(client))[0]["creator_fans"] == 12300 @pytest.mark.asyncio async def test_another_platforms_snapshot_does_not_leak(self, client): """两个平台上恰好同名同哈希的博主是两个人 —— 快照挂在任务上，不该串。""" await _seed(platform="xhs") dy_task_id = await _seed(platform="dy") await _add_stat(dy_task_id, run_id=1, fans=999) assert (await _notes(client, "xhs"))[0]["creator_fans"] is None assert (await _notes(client, "dy"))[0]["creator_fans"] == 999 @pytest.mark.asyncio async def test_an_unparsed_count_stays_null(self, client): """快照在，但某一项没解析出来 —— 那一项必须是 null，不能变成 0。 和上一条的区别：那条是"根本没有快照"，这条是"有快照、其中一项平台没给"。 界面上两种都该是「—」。 """ task_id = await _seed() await _add_stat( task_id, run_id=1, fans=None, total_favorited=None, works=None ) note = (await _notes(client))[0] assert note["creator_fans"] is None assert note["creator_works"] is None # 快照本身是有的（有采集时间），只是值不知道 —— 前端要能分开这两件事。 assert note["creator_stats_at"] is not None