feat(monitor): 组头上加一个跳到博主主页的按钮
Deploy VitePress site to Pages / build (push) Canceled after 0s
Deploy VitePress site to Pages / Deploy (push) Canceled after 0s

`creator_hash` 是单向的,拼不出主页地址 —— 所以这个 id 必须在采集时单独存下来。
抖音的资料接口本来就返回 sec_uid,只是之前没落库。

* `monitor_creator_stat` 加 `creator_id` 列(老数据默认空,下一次采集补上);
* 链接在 `adapters.creator_url()` 里拼,**空串是有意义的返回值**:平台不认识、
  或者没拿到 id,就返回空,界面据此不画按钮 —— 画一个点开 404 的按钮比不画糟;
* 没有作品的博主照样有链接:他恰恰是你想点进去看的那个人。

已知缺口:小红书那条路(爬虫子进程)根本不落创作者资料,所以那边没有链接。
This commit is contained in:
2026-10-11 08:52:51 +08:00
parent 65fc16ddc9
commit 44277600e4
10 changed files with 172 additions and 0 deletions
+47
View File
@@ -139,6 +139,7 @@ async def _add_stat(
works: int | None = 42,
creator_hash: str = CREATOR_HASH,
captured_at: int = 1,
creator_id: str = "",
) -> None:
async with monitor_db.get_session() as session:
session.add(
@@ -147,6 +148,7 @@ async def _add_stat(
run_id=run_id,
creator_hash=creator_hash,
nickname=NICKNAME,
creator_id=creator_id,
fans=fans,
total_favorited=total_favorited,
works_count=works,
@@ -320,3 +322,48 @@ class TestCreatorList:
assert len(creators) == 2
assert {row["creator_hash"] for row in creators} == {CREATOR_HASH}
assert sum(row["note_count"] for row in creators) == 2
class TestCreatorProfileLink:
"""组头上那个「跳到他主页」的按钮。
它要的是**平台上的主页 id**,不是 ``creator_hash``(哈希单向,拼不出地址)。所以
这个 id 是采集资料时单独存下来的;拿不到就返回空串让界面不画按钮。
"""
@pytest.mark.asyncio
async def test_the_link_is_built_from_the_stored_id(self, client):
task_id = await _seed_task(platform="dy")
await _add_stat(task_id, run_id=1, fans=1, creator_id="MS4wLjABAAAAsomebody")
creators = await _creators(client, "dy")
assert creators[0]["creator_url"] == (
"https://www.douyin.com/user/MS4wLjABAAAAsomebody"
)
@pytest.mark.asyncio
async def test_no_id_means_no_link(self, client):
"""没采到主页 id 就不给链接 —— 画一个点开 404 的按钮比不画糟。"""
task_id = await _seed_task(platform="dy")
await _add_stat(task_id, run_id=1, fans=1, creator_id="")
assert (await _creators(client, "dy"))[0]["creator_url"] == ""
@pytest.mark.asyncio
async def test_a_creator_with_only_works_has_no_link_either(self, client):
"""小红书那条路根本不落创作者资料,所以那边永远是空的 —— 这是已知的、可接受的。"""
await _seed(platform="xhs")
assert (await _creators(client, "xhs"))[0]["creator_url"] == ""
@pytest.mark.asyncio
async def test_a_creator_with_no_works_still_gets_the_link(self, client):
"""**没有作品的博主照样能跳过去** —— 他恰恰是你想点进去看看发的什么的那个人。"""
task_id = await _seed_task(platform="dy")
await _add_stat(task_id, run_id=1, fans=1, creator_id="MS4wLjABAAAAquiet")
creators = await _creators(client, "dy")
assert creators[0]["note_count"] == 0
assert creators[0]["creator_url"].endswith("/user/MS4wLjABAAAAquiet")
+34
View File
@@ -852,6 +852,7 @@ def _profile(creator_hash: str = "hash", **extra) -> Dict[str, Any]:
record: Dict[str, Any] = {
"creator_hash": creator_hash,
"nickname": "博主",
"creator_id": "MS4wLjABAAAAtestsuid",
"unique_id": "abc",
"fans": 12000,
"total_favorited": 83000,
@@ -976,3 +977,36 @@ class TestCreatorStatSnapshots:
)
assert [s.fans for s in stats] == [12000, 12300]
class TestCreatorIdIsStored:
"""主页 id 要单独存下来 —— ``creator_hash`` 是单向的,拼不出主页地址。"""
@pytest.mark.asyncio
async def test_the_profile_id_lands_on_the_snapshot(self, db, tmp_path):
task = await _make_task(db, platform="dy")
run = await _make_run(db, task, started_at=1)
_write_run_dir(
tmp_path, [_dy_note("1")], comments=[], subdir="douyin", profiles=[_profile()]
)
await ingest_run(db, run, task, tmp_path)
stat = await db.scalar(select(MonitorCreatorStat))
assert stat.creator_id == "MS4wLjABAAAAtestsuid"
@pytest.mark.asyncio
async def test_a_missing_id_is_empty_not_a_crash(self, db, tmp_path):
"""老数据/别的平台没有这个字段 —— 空串,界面据此不画按钮。"""
task = await _make_task(db, platform="dy")
run = await _make_run(db, task, started_at=1)
_write_run_dir(
tmp_path,
[_dy_note("1")],
comments=[],
subdir="douyin",
profiles=[_profile(creator_id="")],
)
await ingest_run(db, run, task, tmp_path)
assert (await db.scalar(select(MonitorCreatorStat))).creator_id == ""
+27
View File
@@ -389,3 +389,30 @@ class TestLegacyKeyMigration:
assert "xhs_cookie" not in rows
await monitor_db.dispose_engine()
class TestCreatorUrl:
"""博主主页链接。
注意它要的是**平台上的主页 id**(抖音是 sec_uid),不是 ``creator_hash`` —— 哈希是
单向的,拼不出地址。所以这条链路上少任何一环(采集没存 id、平台上没有这个博主、
平台不认识)都必须返回空串,让界面干脆不画按钮,而不是给一个点开 404 的链接。
"""
def test_douyin(self):
assert (
adapters.creator_url("dy", "MS4wLjABAAAAabc-123")
== "https://www.douyin.com/user/MS4wLjABAAAAabc-123"
)
def test_xiaohongshu(self):
assert (
adapters.creator_url("xhs", "5f8a1b2c3d4e5f6a7b8c9d0e")
== "https://www.xiaohongshu.com/user/profile/5f8a1b2c3d4e5f6a7b8c9d0e"
)
def test_no_id_is_empty_not_a_broken_link(self):
assert adapters.creator_url("dy", "") == ""
def test_an_unknown_platform_is_empty_not_a_crash(self):
assert adapters.creator_url("nope", "abc") == ""