feat(monitor): 作品标签 —— 词表在设置里维护，作品上贴
和「作品备注」刻意并存，不是一回事：备注是一句话的自由文字，标签是**从一份固定
词表里选的分类**。分类必须封闭，否则「重点 / 重要 / 优先」各写各的，筛就没法用了。
* `monitor_tag`：词表，全局一套（不按平台分）—— 「重点」是给人自己用的心智，不该在
小红书和抖音各定义一遍。颜色存的是**调色板里的名字**，不是色值：Tailwind 的类名是
静态提取的，拼出来的 `text-${color}` 它看不见，线上会静默变无色。
* `monitor_note_tag`：作品↔标签，**多对多**。一条作品可以既是重点又是竞品；只能贴一个
的话人就会跑去备注里写自由文字，这份词表就白建了。全部替换式提交，不是逐个增删。
* 删除标签时**显式删关联**，不靠外键级联 —— 测试跑 SQLite，它默认不开外键约束。
* 组头上的筛选做成平铺开关片，多选是「或」。
顺带：/monitor/notes 现在支持按 tag_id 筛。
This commit is contained in:
@@ -0,0 +1,237 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""作品标签 —— 词表在设置里维护，作品上贴。
|
||||
|
||||
和「作品备注」是两件事，刻意并存：备注是一句话说明（自由文字），标签是从**一份固定
|
||||
词表**里选的分类。分类得是一套封闭的词，否则「重点 / 重要 / 优先」会各写各的，筛选
|
||||
就没法用了。
|
||||
"""
|
||||
|
||||
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, MonitorNote, MonitorTask
|
||||
|
||||
|
||||
@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 = "任务", note_ids=("note-a",)) -> int:
|
||||
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()
|
||||
for note_id in note_ids:
|
||||
session.add(
|
||||
MonitorNote(
|
||||
task_id=task.id, note_id=note_id, title=f"作品{note_id}",
|
||||
note_url="", cover="", creator_hash="hash-a",
|
||||
creator_name="博主", source_kind="", published_at=1,
|
||||
first_seen_run_id=1, first_seen_at=1,
|
||||
last_seen_run_id=1, last_seen_at=1,
|
||||
)
|
||||
)
|
||||
return task.id
|
||||
|
||||
|
||||
async def _make_tag(client, name: str, color: str = "") -> int:
|
||||
response = await client.post("/api/monitor/tags", json={"name": name, "color": color})
|
||||
assert response.status_code == 200, response.text
|
||||
return response.json()["id"]
|
||||
|
||||
|
||||
async def _tags(client):
|
||||
return (await client.get("/api/monitor/tags")).json()["tags"]
|
||||
|
||||
|
||||
async def _notes(client, **params):
|
||||
return (await client.get("/api/monitor/notes", params=params)).json()["notes"]
|
||||
|
||||
|
||||
async def _attach(client, note_id: str, tag_ids, platform: str = "xhs"):
|
||||
return await client.put(
|
||||
f"/api/monitor/notes/{note_id}/tags",
|
||||
params={"platform": platform},
|
||||
json={"tag_ids": tag_ids},
|
||||
)
|
||||
|
||||
|
||||
class TestTagVocabulary:
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_new_tag_shows_up_in_the_list(self, client):
|
||||
await _make_tag(client, "重点", "pink")
|
||||
|
||||
tags = await _tags(client)
|
||||
|
||||
assert [(tag["name"], tag["color"]) for tag in tags] == [("重点", "pink")]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_duplicate_names_are_rejected(self, client):
|
||||
"""两个「重点」在筛选列表里是灾难 —— 选哪个都像对的。"""
|
||||
await _make_tag(client, "重点")
|
||||
|
||||
response = await client.post("/api/monitor/tags", json={"name": "重点"})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "已经存在" in response.json()["detail"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_unknown_color_falls_back_instead_of_breaking(self, client):
|
||||
await _make_tag(client, "重点", color="chartreuse")
|
||||
|
||||
assert (await _tags(client))[0]["color"] == "cyan"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_renaming_works(self, client):
|
||||
tag_id = await _make_tag(client, "重点")
|
||||
|
||||
await client.put(f"/api/monitor/tags/{tag_id}", json={"name": "非常重点"})
|
||||
|
||||
assert (await _tags(client))[0]["name"] == "非常重点"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_renaming_onto_an_existing_name_is_rejected(self, client):
|
||||
await _make_tag(client, "重点")
|
||||
tag_id = await _make_tag(client, "次要")
|
||||
|
||||
response = await client.put(
|
||||
f"/api/monitor/tags/{tag_id}", json={"name": "重点"}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_count_says_how_many_works_wear_it(self, client):
|
||||
"""删之前要能看清代价：用在 47 条作品上和 0 条，是两个决定。"""
|
||||
await _seed(note_ids=("note-a", "note-b"))
|
||||
tag_id = await _make_tag(client, "重点")
|
||||
|
||||
await _attach(client, "note-a", [tag_id])
|
||||
await _attach(client, "note-b", [tag_id])
|
||||
|
||||
assert (await _tags(client))[0]["note_count"] == 2
|
||||
|
||||
|
||||
class TestTaggingWorks:
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_work_can_wear_several_tags(self, client):
|
||||
"""**多对多是这个功能的重点**：一条作品可以既是「重点」又是「竞品」。只能贴一个
|
||||
的话，人就会跑去备注里写自由文字，那这份封闭词表就白建了。"""
|
||||
await _seed()
|
||||
first = await _make_tag(client, "重点")
|
||||
second = await _make_tag(client, "竞品")
|
||||
|
||||
await _attach(client, "note-a", [first, second])
|
||||
|
||||
names = [tag["name"] for tag in (await _notes(client))[0]["tags"]]
|
||||
assert names == ["重点", "竞品"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_attaching_replaces_the_whole_set(self, client):
|
||||
await _seed()
|
||||
first = await _make_tag(client, "重点")
|
||||
second = await _make_tag(client, "竞品")
|
||||
await _attach(client, "note-a", [first, second])
|
||||
|
||||
await _attach(client, "note-a", [second])
|
||||
|
||||
assert [t["name"] for t in (await _notes(client))[0]["tags"]] == ["竞品"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_empty_list_clears_them(self, client):
|
||||
await _seed()
|
||||
tag_id = await _make_tag(client, "重点")
|
||||
await _attach(client, "note-a", [tag_id])
|
||||
|
||||
await _attach(client, "note-a", [])
|
||||
|
||||
assert (await _notes(client))[0]["tags"] == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unknown_tag_ids_are_ignored_not_stored(self, client):
|
||||
"""前端传了脏 id（比如标签刚被删掉）时，别在关联表里留下孤儿。"""
|
||||
await _seed()
|
||||
tag_id = await _make_tag(client, "重点")
|
||||
|
||||
await _attach(client, "note-a", [tag_id, 99999])
|
||||
|
||||
assert [t["id"] for t in (await _notes(client))[0]["tags"]] == [tag_id]
|
||||
assert (await _tags(client))[0]["note_count"] == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_works_start_without_tags(self, client):
|
||||
await _seed()
|
||||
|
||||
assert (await _notes(client))[0]["tags"] == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tags_do_not_leak_to_another_platform(self, client):
|
||||
"""作品的 id 分平台编号体系 —— 抖音的 123 和小红书的 123 是两条作品。"""
|
||||
await _seed(platform="xhs")
|
||||
await _seed(platform="dy")
|
||||
tag_id = await _make_tag(client, "重点")
|
||||
|
||||
await _attach(client, "note-a", [tag_id], platform="xhs")
|
||||
|
||||
assert (await _notes(client, platform="xhs"))[0]["tags"] != []
|
||||
assert (await _notes(client, platform="dy"))[0]["tags"] == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_deleting_a_tag_unsticks_it_everywhere(self, client):
|
||||
await _seed(note_ids=("note-a", "note-b"))
|
||||
tag_id = await _make_tag(client, "重点")
|
||||
await _attach(client, "note-a", [tag_id])
|
||||
await _attach(client, "note-b", [tag_id])
|
||||
|
||||
response = await client.delete(f"/api/monitor/tags/{tag_id}")
|
||||
|
||||
assert response.json()["removed"] == 2
|
||||
assert (await _tags(client)) == []
|
||||
for note in await _notes(client):
|
||||
assert note["tags"] == []
|
||||
|
||||
|
||||
class TestFilteringByTag:
|
||||
@pytest.mark.asyncio
|
||||
async def test_only_works_wearing_the_tag_come_back(self, client):
|
||||
await _seed(note_ids=("note-a", "note-b"))
|
||||
tag_id = await _make_tag(client, "重点")
|
||||
await _attach(client, "note-a", [tag_id])
|
||||
|
||||
notes = await _notes(client, tag_id=tag_id)
|
||||
|
||||
assert [note["note_id"] for note in notes] == ["note-a"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_several_tags_means_any_of_them(self, client):
|
||||
"""多选是「或」不是「且」—— 想同时满足两个标签的做法是再点一个标签，而不是
|
||||
指望筛选去求交集。"""
|
||||
await _seed(note_ids=("note-a", "note-b", "note-c"))
|
||||
first = await _make_tag(client, "重点")
|
||||
second = await _make_tag(client, "竞品")
|
||||
await _attach(client, "note-a", [first])
|
||||
await _attach(client, "note-b", [second])
|
||||
|
||||
notes = await _notes(client, tag_id=[first, second])
|
||||
|
||||
assert sorted(note["note_id"] for note in notes) == ["note-a", "note-b"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_filter_returns_everything(self, client):
|
||||
await _seed(note_ids=("note-a", "note-b"))
|
||||
|
||||
assert len(await _notes(client)) == 2
|
||||
Reference in New Issue
Block a user