tests/test_follow_store.py
Rama master · réplica de lectura del repositorio original.
"""追更订阅与候选落地的隔离测试。全部使用临时数据库，绝不碰真实 ledger。"""
import json
import sqlite3
import tempfile
import unittest
from dataclasses import replace
from datetime import datetime, timedelta, timezone
from pathlib import Path
from peach import follow_providers
from peach.follow import FollowSourceError
from peach.follow_sources import FollowCandidate, IwaraConnector, SourceFetch
from peach.follow_store import (
FollowStore, ReleaseGroup, author_display_text, f95_author_handles,
f95_author_name, normalized_author_name,
)
from support.ledger import fresh_ledger
ROOT = Path(__file__).resolve().parents[1]
MOMENT = datetime(2026, 8, 25, 9, 0, tzinfo=timezone.utc)
def _candidate(external_id, title, **kwargs):
kwargs.setdefault("provider", "rule34video")
return FollowCandidate(external_id=external_id, title=title, **kwargs)
def _fetch(candidates, *, provider="rule34video", ref="lazyprocrastinator",
semantics="work", raw=b"<html/>", **kwargs):
return SourceFetch(
provider=provider, ref=ref,
request_url=f"https://{provider}.test/{ref}", semantics=semantics,
candidates=tuple(candidates), raw_body=raw, **kwargs)
class _StoreCase(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.root = Path(self.temporary.name).resolve()
self.connection = sqlite3.connect(fresh_ledger(self.root))
self.connection.row_factory = sqlite3.Row
self.addCleanup(self.connection.close)
self.store = FollowStore(lambda: self.connection,
sources_root=self.root / "sources")
def _entity(self, name="LazyProcrast", aliases=("LazyProcrastinator",)):
cursor = self.connection.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES('creator',?,?,?,?)",
(name, name.lower(), "2026-08-25T00:00:00Z", "2026-08-25T00:00:00Z"))
entity_id = int(cursor.lastrowid)
for alias in aliases:
self.connection.execute(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source)"
" VALUES(?,?,?,'test')", (entity_id, alias, alias.lower()))
return entity_id
def _source(self, provider="rule34video", ref="lazyprocrastinator", **kwargs):
kwargs.setdefault("label", "LazyProcrastinator")
kwargs.setdefault("url", f"https://{provider}.test/{ref}")
return self.store.register(provider=provider, ref=ref, moment=MOMENT, **kwargs)
class RegistrationTests(_StoreCase):
def test_register_is_idempotent_and_updates_the_label(self):
first = self._source()
second = self._source(label="Lazy P")
self.assertEqual(first, second)
rows = self.store.sources()
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["label"], "Lazy P")
def test_merging_metadata_keeps_the_keys_it_does_not_mention(self):
"""名片是登记之后才解析出来的，补它不能抹掉登记时写下的 author_key。"""
source_id = self._source(metadata={"author_key": "mrstrauz"})
self.store.merge_source_metadata(
source_id, {"official_links": [{"service": "twitter",
"handle": "Mr_Strauz"}]},
moment=MOMENT)
stored = json.loads(self.store.sources()[0]["metadata_json"])
self.assertEqual(stored["author_key"], "mrstrauz")
self.assertEqual([link["handle"] for link in stored["official_links"]],
["Mr_Strauz"])
def test_an_empty_profile_list_is_recorded_as_an_answer(self):
# 「问过了，他没留主页」要留得下来，否则每次检查都要再问一遍。
source_id = self._source()
self.store.merge_source_metadata(source_id, {"official_links": []},
moment=MOMENT)
stored = json.loads(self.store.sources()[0]["metadata_json"])
self.assertEqual(stored["official_links"], [])
def test_rule34_case_variants_are_the_same_source(self):
first = self._source(provider="rule34xxx", ref="LazyProcrastinator")
second = self._source(provider="rule34xxx", ref="lazyprocrastinator")
self.assertEqual(first, second)
self.assertEqual(len(self.store.sources()), 1)
self.assertEqual(self.store.sources()[0]["ref"], "lazyprocrastinator")
def test_register_keeps_an_existing_entity_when_the_update_omits_it(self):
entity_id = self._entity()
self._source(entity_id=entity_id)
self._source(label="again")
self.assertEqual(self.store.sources()[0]["entity_id"], entity_id)
def test_enabled_only_filters_disabled_sources(self):
source_id = self._source()
self._source(provider="kemono", ref="fanbox/30917150")
self.store.set_enabled(source_id, False, moment=MOMENT)
self.assertEqual([row["provider"] for row in self.store.sources(enabled_only=True)],
["kemono"])
def test_bad_semantics_is_rejected(self):
with self.assertRaises(FollowSourceError):
self._source(semantics="whatever")
def test_creator_aliases_include_canonical_name_and_aliases(self):
entity_id = self._entity()
self.assertEqual(self.store.creator_aliases(entity_id),
("LazyProcrast", "LazyProcrastinator"))
self.assertEqual(self.store.creator_aliases(None), ())
class RecordTests(_StoreCase):
def test_iwara_general_tags_enter_the_inventory_without_guessing_other_types(self):
from peach.web_follow import _item_tags, online_tag_inventory
candidate = IwaraConnector()._candidate({
"id": "Video123", "title": "Test animation", "tags": [
{"id": "dance", "type": "general"},
{"id": "blender", "type": "category"},
{"id": "series_title", "type": "source"},
{"id": "pose", "type": "unknown"},
{"id": "animation"}]}, {"name": "Test Creator"}, "creator")
source_id = self._source(provider="iwara", ref="creator")
self.store.record(source_id, _fetch([candidate], provider="iwara", ref="creator"),
moment=MOMENT)
rows = online_tag_inventory(self.connection)
self.assertEqual([(row["k"], row["cat"]) for row in rows], [("dance", "general")])
item = self.store.items()[0]
self.assertEqual(_item_tags(item), ["dance"])
self.assertEqual(item.metadata["source_tag_types"], {
"dance": "general", "blender": "category", "series_title": "source", "pose": "unknown"})
self.assertEqual(item.metadata["tags"], ["dance", "blender", "series_title", "pose", "animation"])
def test_booru_source_only_becomes_a_profile_when_the_handle_matches(self):
source_id = self._source(
provider="rule34xxx", ref="auxtasy", label="Auxtasy",
metadata={"author_key": "auxtasy", "official_links": [
{"service": "twitter", "handle": "OldWrongName",
"url": "https://x.com/OldWrongName"},
]},
)
self.store.record(source_id, _fetch([
_candidate("1", "one", provider="rule34xxx",
extra={"source": "https://x.com/Auxtasy/status/1"}),
_candidate("2", "two", provider="rule34xxx",
extra={"source": "https://www.patreon.com/futavr/posts/2"}),
], provider="rule34xxx", ref="auxtasy"), moment=MOMENT)
metadata = json.loads(self.store.sources()[0]["metadata_json"])
self.assertEqual(
[(link["service"], link["handle"]) for link in metadata["official_links"]],
[("twitter", "Auxtasy")],
)
def test_booru_source_accepts_a_confirmed_alias_but_not_a_collaborator(self):
self.store.upsert_author_alias(
"LazyProcrastinator", "lazyprocrast", source="manual", moment=MOMENT)
source_id = self._source(
provider="rule34xxx", ref="lazyprocrastinator",
metadata={"author_key": "lazyprocrastinator"},
)
self.store.record(source_id, _fetch([
_candidate("1", "one", provider="rule34xxx",
extra={"source": "https://lazyprocrast.fanbox.cc/posts/1"}),
_candidate("2", "two", provider="rule34xxx",
extra={"source": "https://x.com/unrelated/status/2"}),
], provider="rule34xxx", ref="lazyprocrastinator"), moment=MOMENT)
metadata = json.loads(self.store.sources()[0]["metadata_json"])
self.assertEqual([link["handle"] for link in metadata["official_links"]],
["lazyprocrast"])
def test_backfill_pages_add_to_the_profile_links_the_first_page_found(self):
"""往回抓只并集追加出处：更早那几页看不到近期作品链出去的那些站。"""
source_id = self._source(
provider="rule34xxx", ref="auxtasy", label="Auxtasy",
metadata={"author_key": "auxtasy"},
)
self.store.record(source_id, _fetch([
_candidate("1", "one", provider="rule34xxx",
extra={"source": "https://x.com/Auxtasy/status/1"}),
], provider="rule34xxx", ref="auxtasy"), moment=MOMENT)
self.store.record(source_id, _fetch([
_candidate("9", "nine", provider="rule34xxx",
extra={"source": "https://auxtasy.fanbox.cc/posts/9"}),
], provider="rule34xxx", ref="auxtasy"), moment=MOMENT, page=2)
metadata = json.loads(self.store.sources()[0]["metadata_json"])
self.assertEqual([link["handle"] for link in metadata["official_links"]],
["Auxtasy", "auxtasy"])
def test_first_fetch_adds_items_and_classifies_variants(self):
source_id = self._source(entity_id=self._entity())
outcome = self.store.record(source_id, _fetch([
_candidate("4542713", "Fiona - Paizuri"),
_candidate("4542721", "Fiona - Paizuri (Nude)"),
_candidate("4542899", "Mitsuru [WIP]"),
]), creator_aliases=self.store.creator_aliases(1), moment=MOMENT)
self.assertEqual((outcome.added, outcome.updated, outcome.discovered), (3, 0, 3))
kinds = {item.external_id: item.variant_kind for item in self.store.items()}
self.assertEqual(kinds, {"4542713": "main", "4542721": "alt", "4542899": "wip"})
def test_second_fetch_updates_without_resetting_user_state(self):
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "Fiona - Paizuri")]),
moment=MOMENT)
item = self.store.items()[0]
self.store.set_status(item.id, "ignored")
later = MOMENT + timedelta(hours=1)
outcome = self.store.record(
source_id, _fetch([_candidate("1", "Fiona - Paizuri", url="https://x.test/1")],
raw=b"<html>2</html>"), moment=later)
self.assertEqual((outcome.added, outcome.updated), (0, 1))
refreshed = self.store.items()[0]
self.assertEqual(refreshed.status, "ignored")
self.assertEqual(refreshed.url, "https://x.test/1")
self.assertEqual(refreshed.first_seen_at, item.first_seen_at)
self.assertNotEqual(refreshed.last_seen_at, item.last_seen_at)
def test_a_failed_detail_never_erases_a_known_upload_time(self):
"""paheal 每个候选都要单独打一次详情页，被限流时 `_detail` 返回 {}。
此前 upsert 无条件覆盖，于是上一轮已经取到的上传时间和时长被抹成 NULL，
而 `COALESCE(published_at, first_seen_at)` 会让界面改显示抓取时刻——
看着是条完整记录，时间却是错的。实测 168 条 paheal 里 7 条正是这样。
"""
source_id = self._source()
self.store.record(source_id, _fetch([_candidate(
"1", "A", published_at="2026-08-18T09:03:30Z", duration=44.7)]), moment=MOMENT)
first = self.store.items()[0]
self.assertEqual(first.published_at, "2026-08-18T09:03:30Z")
later = MOMENT + timedelta(hours=1)
self.store.record(source_id, _fetch([_candidate("1", "A")], raw=b"<html>2</html>"),
moment=later)
again = self.store.items()[0]
self.assertEqual(again.published_at, "2026-08-18T09:03:30Z")
self.assertEqual(again.duration, 44.7)
self.assertEqual(again.published_precision, "exact")
def test_a_later_fetch_with_a_real_time_still_wins(self):
"""保守只针对「取不到」；来源真的给了新时间就要覆盖，不能变成只写一次。"""
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "A")]), moment=MOMENT)
self.assertIsNone(self.store.items()[0].published_at)
self.store.record(source_id, _fetch([_candidate(
"1", "A", published_at="2026-08-18T09:03:30Z")], raw=b"<html>2</html>"),
moment=MOMENT + timedelta(hours=1))
item = self.store.items()[0]
self.assertEqual(item.published_at, "2026-08-18T09:03:30Z")
self.assertEqual(item.published_precision, "exact")
def test_not_modified_records_the_check_without_touching_items(self):
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "A")]), moment=MOMENT)
outcome = self.store.record(
source_id, _fetch([], raw=None, not_modified=True), moment=MOMENT)
self.assertTrue(outcome.not_modified)
self.assertEqual(len(self.store.items()), 1)
self.assertEqual(self.store.sources()[0]["last_status"], "not_modified")
def test_conditional_cursor_is_stored_on_the_source_row(self):
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "A")], etag='"v1"',
last_modified="Mon, 25 Aug 2026 09:00:00 GMT"),
moment=MOMENT)
row = self.store.sources()[0]
self.assertEqual(row["etag"], '"v1"')
self.assertEqual(row["last_modified"], "Mon, 25 Aug 2026 09:00:00 GMT")
def test_raw_evidence_is_written_once_and_immutably(self):
source_id = self._source()
outcome = self.store.record(source_id, _fetch([_candidate("1", "A")]),
moment=MOMENT)
# evidence_path 是相对 `sources/` 的路径，分隔符随平台不同——不能写死 `/`。
evidence = self.root / "sources" / Path(outcome.evidence_path)
self.assertTrue(evidence.is_file())
sidecar = json.loads(evidence.with_suffix(".json").read_text(encoding="utf-8"))
self.assertEqual(sidecar["candidates"], 1)
self.assertEqual(sidecar["request_url"], "https://rule34video.test/lazyprocrastinator")
def test_an_unusable_evidence_root_does_not_lose_the_candidates(self):
"""Mac 上 `peach-data/sources` 指向外置盘。盘不在时它是一条断链，
`mkdir(exist_ok=True)` 会抛 `FileExistsError`（链接在、目标不在）。
发现跟归档盘无关，不该让整次检查连同已抓到的候选一起炸掉。
这里用「路径被普通文件占住」构造同一个 `FileExistsError`：断链要 symlink，
而非管理员的 Windows 建不了 symlink，那样测试就只在一个平台成立。
"""
(self.root / "evidence").write_text("not a directory", encoding="utf-8")
store = FollowStore(lambda: self.connection,
sources_root=self.root / "evidence")
source_id = store.register(provider="rule34video", ref="x", label="X",
url="https://x.test/", moment=MOMENT)
outcome = store.record(source_id, _fetch([_candidate("1", "Fiona - Paizuri")]),
moment=MOMENT)
self.assertEqual(outcome.added, 1)
self.assertIsNone(outcome.evidence_path)
self.assertIn("证据未取得", outcome.evidence_error)
item = store.items()[0]
self.assertEqual(item.title, "Fiona - Paizuri")
self.assertIsNone(item.metadata.get("evidence_path"))
def test_a_readable_evidence_root_reports_no_error(self):
source_id = self._source()
outcome = self.store.record(source_id, _fetch([_candidate("1", "A")]),
moment=MOMENT)
self.assertIsNotNone(outcome.evidence_path)
self.assertIsNone(outcome.evidence_error)
def test_identical_evidence_is_reused_without_overwriting_files(self):
source_id = self._source()
fetch = _fetch([_candidate("1", "A")])
first = self.store.record(source_id, fetch, moment=MOMENT)
evidence = self.root / "sources" / Path(first.evidence_path)
sidecar = evidence.with_suffix(".json")
before = [(path.read_bytes(), path.stat().st_mtime_ns) for path in (evidence, sidecar)]
again = self.store.record(source_id, fetch, moment=MOMENT)
self.assertEqual(again.evidence_path, first.evidence_path)
self.assertEqual(again.added, 0)
self.assertEqual([(path.read_bytes(), path.stat().st_mtime_ns)
for path in (evidence, sidecar)], before)
def test_shared_response_body_keeps_distinct_snapshot_metadata(self):
for provider, spec in follow_providers.PROVIDERS.items():
if not spec.source_url:
continue
with self.subTest(provider=provider):
source_id = self._source(provider=provider, semantics=spec.semantics)
fetch = _fetch([_candidate("1", "A", provider=provider)],
provider=provider, semantics=spec.semantics)
variants = (fetch, replace(fetch, request_url=fetch.request_url + "?firstId=1"),
replace(fetch, candidates=()), replace(fetch, raw_body=b"different body"))
paths = []
for offset, variant in enumerate(variants):
# 时刻各不相同：保留规则按时间留最近几份，同一时刻的先后没有定义。
outcome = self.store.record(source_id, variant, moment=MOMENT + timedelta(seconds=offset))
self.assertIsNone(outcome.evidence_error)
paths.append(self.root / "sources" / Path(outcome.evidence_path))
self.assertEqual(len(set(paths)), 4)
for path, variant in zip(paths, variants):
self.assertEqual(path.read_bytes(), variant.raw_body)
sidecar = json.loads(path.with_suffix(".json").read_text(encoding="utf-8"))
self.assertEqual(sidecar["provider"], provider)
self.assertEqual(sidecar["request_url"], variant.request_url)
self.assertEqual(sidecar["candidates"], len(variant.candidates))
def test_conflicting_evidence_still_rejects_overwriting(self):
source_id = self._source()
fetch = _fetch([_candidate("1", "A")])
outcome = self.store.record(source_id, fetch, moment=MOMENT)
evidence = self.root / "sources" / Path(outcome.evidence_path)
sidecar = evidence.with_suffix(".json")
original = sidecar.read_bytes()
sidecar.write_bytes(b"conflicting evidence")
with self.assertRaisesRegex(FollowSourceError, "immutable follow snapshot collision"):
self.store.record(source_id, fetch, moment=MOMENT)
self.assertEqual(sidecar.read_bytes(), b"conflicting evidence")
self.assertEqual(evidence.read_bytes(), fetch.raw_body)
sidecar.write_bytes(original)
def test_approximate_precision_is_carried_through_from_the_connector(self):
source_id = self._source()
self.store.record(source_id, _fetch([
_candidate("1", "A", published_at="2026-08-18T00:00:00Z",
extra={"published_precision": "approximate"}),
_candidate("2", "B"),
]), moment=MOMENT)
precision = {i.external_id: i.published_precision for i in self.store.items()}
self.assertEqual(precision, {"1": "approximate", "2": "unknown"})
def test_record_error_keeps_the_message_and_status(self):
source_id = self._source()
self.store.record_error(source_id, "HTTP 403", moment=MOMENT,
status="unauthorized")
row = self.store.sources()[0]
self.assertEqual(row["last_status"], "unauthorized")
self.assertEqual(row["last_error"], "HTTP 403")
def test_items_can_be_filtered_by_status(self):
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "A"), _candidate("2", "B")]),
moment=MOMENT)
self.store.set_status(self.store.items()[0].id, "seen")
self.assertEqual(len(self.store.items(statuses=("new",))), 1)
def test_an_ignored_item_can_be_restored_to_new(self):
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "A")]), moment=MOMENT)
item = self.store.items()[0]
self.store.set_status(item.id, "ignored")
self.store.set_status(item.id, "new")
self.assertEqual(self.store.items()[0].status, "new")
def test_saved_status_cannot_be_set_directly(self):
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "A")]), moment=MOMENT)
with self.assertRaises(FollowSourceError):
self.store.set_status(self.store.items()[0].id, "saved")
with self.assertRaises(FollowSourceError):
self.store.set_status(self.store.items()[0].id, "nonsense")
def test_a_tag_the_site_later_marks_as_a_character_follows_on_older_items(self):
"""站方事后把标签改成角色：新条目记上之后，同站旧条目里的同名标签跟着改。
跨作者也改；别的站不动；只往 character 改，新条目记成 general 不降级旧行。
"""
def typed(external_id, kind, **kwargs):
return _candidate(external_id, external_id, extra={"tag_types": {
"d.mon (overwatch)": kind, "sex": "general"}}, **kwargs)
old = self._source(ref="vileclipse")
other_site = self._source(provider="rule34xxx", ref="vileclipse")
self.store.record(old, _fetch([typed("1", "general")], ref="vileclipse"),
moment=MOMENT)
self.store.record(other_site, _fetch([typed("9", "general", provider="rule34xxx")],
provider="rule34xxx", ref="vileclipse"),
moment=MOMENT)
newer = self._source(ref="someone-else")
# 探详情失败的候选是 partial，它带的类型不作数。
self.store.record(newer, _fetch([typed("2", "character", partial=True)],
ref="someone-else"), moment=MOMENT)
def kinds():
return {item.external_id: item.metadata["tag_types"]["d.mon (overwatch)"]
for item in self.store.items()}
self.assertEqual(kinds()["1"], "general")
self.store.record(newer, _fetch([typed("3", "character")], ref="someone-else"),
moment=MOMENT)
self.assertEqual(kinds(), {"1": "character", "2": "character", "3": "character",
"9": "general"})
old_item = next(item for item in self.store.items() if item.external_id == "1")
self.assertEqual(old_item.metadata["tag_types"]["sex"], "general")
self.store.record(old, _fetch([typed("4", "general")], ref="vileclipse"),
moment=MOMENT)
self.assertEqual(kinds()["1"], "character")
class GroupingTests(_StoreCase):
def test_repeated_collection_of_the_main_upload_preserves_its_variants(self):
for ref in ("hydrafxx", "zmsfm"):
source = self._source(ref=ref)
self.store.record(source, _fetch([
_candidate("main", "Evening Movie"),
_candidate("alt", "Evening Movie (4K60fps)"),
], ref=ref), moment=MOMENT)
self.assertEqual(len(self.store.group(self.store.items())), 1)
for source in self.store.sources():
self.assertEqual(len(self.store.group(self.store.items(source_id=source["id"]))), 1)
def test_stored_combined_video_specs_group_without_rewriting_rows(self):
source = self._source()
self.store.record(source, _fetch([
_candidate("one", "Evening Movie (4K60fps)(NO WM)"),
_candidate("two", "Evening Movie (60fps)(NO WM)"),
]), moment=MOMENT)
self.connection.execute("UPDATE follow_item SET release_key=? WHERE external_id='one'",
("evening movie 4k60fps",))
self.connection.commit()
groups = self.store.group(self.store.items())
self.assertEqual(len(groups), 1)
self.assertEqual({row.external_id for row in (groups[0].primary, *groups[0].variants)},
{"one", "two"})
stored = self.connection.execute("SELECT release_key FROM follow_item WHERE external_id='one'").fetchone()[0]
self.assertEqual(stored, "evening movie 4k60fps")
def _populate(self):
video = self._source()
booru = self._source(provider="rule34xxx", ref="lazyprocrastinator")
self.store.record(video, _fetch([
_candidate("4542713", "Fiona - Paizuri"),
_candidate("4542721", "Fiona - Paizuri (Nude)"),
_candidate("4542705", "Fiona - Missionary"),
]), moment=MOMENT)
self.store.record(booru, _fetch([
FollowCandidate(provider="rule34xxx", external_id="9988770",
title="fiona paizuri",
group_hint="rule34xxx:post:9988770"),
FollowCandidate(provider="rule34xxx", external_id="9988776",
title="totally different filename",
title_is_name=False,
group_hint="rule34xxx:post:9988770"),
], provider="rule34xxx"), moment=MOMENT)
def test_alt_folds_under_its_main_and_other_works_stay_apart(self):
self._populate()
groups = {g.primary.title: g for g in self.store.group(self.store.items())}
self.assertIn("Fiona - Paizuri", groups)
self.assertIn("Fiona - Missionary", groups)
paizuri = groups["Fiona - Paizuri"]
self.assertIn("Fiona - Paizuri (Nude)", [v.title for v in paizuri.variants])
self.assertEqual(groups["Fiona - Missionary"].variants, ())
def test_cross_site_duplicate_is_reported_separately_from_same_site_variants(self):
self._populate()
paizuri = next(g for g in self.store.group(self.store.items())
if g.primary.title == "Fiona - Paizuri")
self.assertEqual(sorted(paizuri.providers), ["rule34video", "rule34xxx"])
self.assertTrue(all(d.provider == "rule34xxx" for d in paizuri.duplicates))
def test_group_hint_beats_an_unrelated_title(self):
# booru 子帖的「标题」是标签拼的，和父帖毫无关系，只有来源声明的键能连起来。
self._populate()
groups = self.store.group(self.store.items())
stray = [g for g in groups if g.primary.title == "totally different filename"]
self.assertEqual(stray, [])
def test_a_shared_origin_key_merges_two_different_sites(self):
# 这是跨站去重真正靠得住的那条路：rule34.xxx 从 source 归一出的键，
# 和 kemono 上同一帖子的键完全相同，标题再不一样也能精确合并。
booru = self._source(provider="rule34xxx", ref="lazyprocrastinator")
kemono = self._source(provider="kemono", ref="fanbox/30917150")
self.store.record(booru, _fetch([
FollowCandidate(provider="rule34xxx", external_id="18534395",
title="fiona · blush", title_is_name=False,
group_hint="fanbox:12304831"),
], provider="rule34xxx", ref="lazyprocrastinator"), moment=MOMENT)
self.store.record(kemono, _fetch([
FollowCandidate(provider="kemono", external_id="12304831",
title="Fiona - Paizuri", group_hint="fanbox:12304831"),
], provider="kemono", ref="fanbox/30917150"), moment=MOMENT)
groups = self.store.group(self.store.items())
self.assertEqual(len(groups), 1)
self.assertEqual(sorted(groups[0].providers), ["kemono", "rule34xxx"])
def test_paheal_and_existing_sources_deduplicate_on_the_subscribestar_post(self):
paheal = self._source(provider="rule34paheal", ref="initiala")
mirror = self._source(provider="rule34xxx", ref="initiala")
self.store.record(paheal, _fetch([
FollowCandidate(provider="rule34paheal", external_id="7428820",
title="Amina · Tifa", title_is_name=False,
group_hint="subscribestar:2639932"),
], provider="rule34paheal", ref="initiala"), moment=MOMENT)
self.store.record(mirror, _fetch([
FollowCandidate(provider="rule34xxx", external_id="18540000",
title="Different tag label", title_is_name=False,
group_hint="subscribestar:2639932"),
], provider="rule34xxx", ref="initiala"), moment=MOMENT)
groups = self.store.group(self.store.items())
self.assertEqual(len(groups), 1)
self.assertEqual(sorted(groups[0].providers), ["rule34paheal", "rule34xxx"])
def test_a_tag_derived_title_never_merges_two_works_on_its_own(self):
# 标签拼出来的「标题」不是名字：同一作者标签相似的两条不能因此被并掉。
source_id = self._source(provider="rule34xxx", ref="lazyprocrastinator")
self.store.record(source_id, _fetch([
FollowCandidate(provider="rule34xxx", external_id="1",
title="fiona · blush", title_is_name=False),
FollowCandidate(provider="rule34xxx", external_id="2",
title="fiona · blush", title_is_name=False),
], provider="rule34xxx", ref="lazyprocrastinator"), moment=MOMENT)
self.assertEqual(len(self.store.group(self.store.items())), 2)
def _booru_post(self, external_id, minutes, *, character="angel_(kof)",
general=("cowgirl_position", "blush", "nude", "pov"), **extra):
tag_types = {"king_of_fighters": "copyright", character: "character",
"you": "character", "lazyprocrastinator": "artist",
**{tag: "general" for tag in general}}
return FollowCandidate(
provider="rule34xxx", external_id=external_id,
title=f"{character} · {general[0]}", title_is_name=False,
published_at=(MOMENT + timedelta(minutes=minutes)).isoformat(),
group_hint=f"rule34xxx:post:{external_id}",
extra={"title_from": "tags", "tag_types": tag_types, **extra})
def _record_booru(self, *posts):
source_id = self._source(provider="rule34xxx", ref="lazyprocrastinator")
self.store.record(source_id, _fetch(posts, provider="rule34xxx"), moment=MOMENT)
return self.store.group(self.store.items())
def test_posts_uploaded_in_one_burst_fold_into_one_work(self):
# 同一段动画按横屏、竖屏和机位拆成几帖连着发：角色相同、标签几乎一样。
groups = self._record_booru(
self._booru_post("1", 0), self._booru_post("2", 1),
self._booru_post("3", 2, general=("cowgirl_position", "blush", "nude", "boobjob")),
self._booru_post("4", 150))
self.assertEqual(len(groups), 1)
self.assertEqual(groups[0].primary.external_id, "1")
self.assertEqual(len(groups[0].variants), 3)
def test_a_burst_needs_the_same_characters(self):
groups = self._record_booru(
self._booru_post("1", 0), self._booru_post("2", 1, character="lavinia_voda"))
self.assertEqual(len(groups), 2)
def test_a_long_pause_starts_a_new_work(self):
groups = self._record_booru(self._booru_post("1", 0), self._booru_post("2", 60 * 4))
self.assertEqual(len(groups), 2)
def test_posts_with_little_tag_overlap_stay_apart(self):
groups = self._record_booru(
self._booru_post("1", 0),
self._booru_post("2", 1, general=("standing_sex", "against_wall", "hat", "nude")))
self.assertEqual(len(groups), 2)
def test_a_post_without_an_origin_joins_the_declared_work_it_was_uploaded_with(self):
# LazyProcrastinator 的 Aerith 一批 6 帖，5 帖写了出处，夹在中间的一帖没写。
origin = "https://www.fanbox.cc/@lazy/posts/12304831"
groups = self._record_booru(
self._booru_post("15101361", 0, source=origin),
self._booru_post("15101385", 3),
self._booru_post("15101390", 4, source=origin))
self.assertEqual(len(groups), 1)
self.assertEqual(len(groups[0].variants), 2)
def test_two_declared_works_are_not_joined_through_a_post_between_them(self):
# 来源把两帖分成了两个作品，中间那帖挂哪一组都是猜，三帖各自成卡。
groups = self._record_booru(
self._booru_post("1", 0, source="https://www.fanbox.cc/@lazy/posts/12304831"),
self._booru_post("2", 1),
self._booru_post("3", 2, source="https://www.fanbox.cc/@lazy/posts/12304832"))
self.assertEqual(len(groups), 3)
def test_posts_retagged_together_are_not_a_burst(self):
# Rekin3D 的两帖分别在 05-01 与 05-20 上传，09-14 一起被改了标签，落库时间都是
# 改标签那一分钟。帖子 id 相差近 20 万，不是连着传的。
groups = self._record_booru(
self._booru_post("17361475", 1), self._booru_post("17559518", 0))
self.assertEqual(len(groups), 2)
def test_an_untagged_booru_post_joins_its_batch_by_the_whole_tag_set(self):
# rule34.paheal 不给标签分类。InitialA 一口气传的 Lily 动画里，8 帖写着同一个
# subscribestar 出处，最后一帖没写；整组标签相同，归进那一组。
source_id = self._source(provider="rule34paheal", ref="initiala", label="initiala")
tags = {tag: "general" for tag in
("InitialA", "Lily_Artemis_II", "Stellar_Blade", "animated", "blender")}
def post(external_id, seconds, **extra):
return FollowCandidate(
provider="rule34paheal", external_id=external_id,
title="lily artemis ii · stellar blade", title_is_name=False,
published_at=(MOMENT + timedelta(seconds=seconds)).isoformat(),
group_hint="subscribestar:2685233" if "source" in extra
else f"rule34paheal:post:{external_id}",
extra={"title_from": "tags", "tag_types": tags, **extra})
origin = "https://subscribestar.adult/posts/2685233"
self.store.record(source_id, _fetch([
post("7452299", 0, source=origin), post("7452300", 1, source=origin),
post("7452301", 25),
post("7452302", 30, tag_types={**tags, "Eve": "general"}),
], provider="rule34paheal", ref="initiala"), moment=MOMENT)
groups = sorted(sorted(m.external_id for m in (g.primary, *g.variants, *g.duplicates))
for g in self.store.group(self.store.items()))
self.assertEqual(groups, [["7452299", "7452300", "7452301"], ["7452302"]])
def test_a_stored_hint_is_recomputed_from_the_recorded_source(self):
# 落库时缺协议头的出处没认出来，键退回了帖子自己的 id；读时按出处重算后并组。
stale = self._booru_post("2", 1, character="ahri",
source="x.com/vileclipse/status/2101310844021928089")
groups = self._record_booru(
self._booru_post("1", 0, source="https://x.com/vileclipse/status/2101310844021928089"),
replace(stale, group_hint="rule34xxx:post:2"))
self.assertEqual(len(groups), 1)
def _upload(self, external_id, hours, *, duration=60.054, character="grace_ashcroft",
general=("athletic_female", "blender", "cowgirl_position", "nude"), **extra):
return replace(self._booru_post(external_id, hours * 60, character=character,
general=general, **extra), duration=duration)
def _members(self, groups):
return sorted(sorted(m.external_id for m in (g.primary, *g.variants, *g.duplicates))
for g in groups)
def test_the_same_clip_uploaded_again_days_later_joins_the_first_upload(self):
# Memz 的 Grace 动画：07-18 一帖带出处，07-23 又传两帖。重新编码过，只有时长逐毫秒
# 相同；后两帖多挂一个作品标签，一般标签也不完全一样。
later = {"resident_evil_9:_requiem": "copyright", "grace_ashcroft": "character",
"memz": "artist", "athletic_female": "general", "blender": "general",
"cowgirl_position": "general", "nude": "general", "3d_animation": "general"}
groups = self._record_booru(
self._upload("18148361", 0, source="https://danbooru.donmai.us/posts/11817280"),
self._upload("18193355", 104, tag_types=later),
self._upload("18193376", 104, tag_types=later))
self.assertEqual(self._members(groups), [["18148361", "18193355", "18193376"]])
def test_an_export_length_shared_by_other_characters_is_not_evidence(self):
# LazyProcrastinator 的 20.02 秒用在上百种角色组合上：同一角色隔几天的两帖是两部。
groups = self._record_booru(
self._upload("1", 0, duration=20.02, character="reika_(doa)"),
self._upload("2", 24 * 100, duration=20.02, character="reika_(doa)"),
self._upload("3", 24 * 50, duration=20.02, character="jote_(ffxvi)"))
self.assertEqual(len(groups), 3)
def test_a_filtered_page_still_knows_an_export_length(self):
# 只看未读时，别的角色那几帖可能都不在这一页；认固定长度要按筛选前的全部条目。
self._record_booru(
self._upload("1", 0, duration=20.02, character="reika_(doa)"),
self._upload("2", 24 * 100, duration=20.02, character="reika_(doa)"),
self._upload("3", 24 * 50, duration=20.02, character="jote_(ffxvi)"))
everything = self.store.items()
page = tuple(item for item in everything if item.external_id != "3")
self.assertEqual(len(self.store.group(page, population=everything)), 2)
def test_uploads_within_one_session_are_left_to_the_burst_rule(self):
# 同一小时里时长相同的几帖是同一场景的不同体位，连发判据按相邻帖的标签切开了：
# 1 与 3 标签相近，中间隔着体位不同的 2，这里不拿时长把 1 和 3 重新串起来。
handjob = ("beach", "handjob", "duo", "pov")
groups = self._record_booru(
self._upload("1", 0, general=handjob),
self._upload("2", 0.25, general=("beach", "ass_focus", "doggy", "wall")),
self._upload("3", 0.5, general=(*handjob[:3], "kneeling")))
self.assertEqual(len(groups), 3)
def test_reuploads_need_overlapping_tags_and_a_millisecond_length(self):
groups = self._record_booru(
self._upload("1", 0), self._upload("2", 48, general=("standing", "hat", "wall", "nude")),
self._upload("3", 0, duration=45.0, character="nyotengu"),
self._upload("4", 48, duration=45.0, character="nyotengu"))
self.assertEqual(len(groups), 4)
def test_two_declared_releases_are_not_joined_as_a_reupload(self):
# Pantsushi3D 发过两条推：来源自己说是两次发布，缩略图再像也不并。
groups = self._record_booru(
self._upload("4704518", 0, source="https://twitter.com/Pantsushi3D/status/1"),
self._upload("6178800", 24 * 392, source="https://twitter.com/Pantsushi3D/status/2"))
self.assertEqual(len(groups), 2)
def _clip(self, video_id, title, *, duration=20.0, character="angel (kof)",
work="King of Fighters", **extra_tags):
tag_types = {"beach": "general", character: "general", work: "copyright",
"LazyProcrastinator": "artist", **extra_tags}
return FollowCandidate(provider="rule34video", external_id=str(video_id),
title=title, duration=duration,
extra={"tag_types": tag_types})
def _record_clips(self, *clips):
source_id = self._source(provider="rule34video", ref="lazyprocrastinator")
self.store.record(source_id, _fetch(clips), moment=MOMENT)
return sorted(sorted(m.title for m in (g.primary, *g.variants, *g.duplicates))
for g in self.store.group(self.store.items()))
def test_a_batch_of_clips_folds_per_character(self):
# 同一段 KOF 动画导出成几条 20 秒短片连着传；Angel 与 Mai 是两个角色，分成两组。
mai = "mai shiranui (dead or alive)"
groups = self._record_clips(
self._clip(4608755, "Angel-riding hugging"), self._clip(4608759, "Angel-missionary"),
self._clip(4608779, "Angel-handjob"), self._clip(4608787, "Mai-blowjob", character=mai),
self._clip(4608795, "Mai-doggy", character=mai))
self.assertEqual(groups, [["Angel-handjob", "Angel-missionary", "Angel-riding hugging"],
["Mai-blowjob", "Mai-doggy"]])
def test_a_clip_a_few_dozen_uploads_earlier_joins_its_pack(self):
# Barney's Mom 一包 8 条 20 秒短片，Riding 比下一条早 64 个 id，中间夹着别人的上传。
freyja = {"freyja (final fantasy)": "general"}
groups = self._record_clips(
self._clip(4608531, "Barney's Mom - Riding", work="Final Fantasy", **freyja),
self._clip(4608595, "Barney's Mom - Doggy 3", work="Final Fantasy", **freyja),
self._clip(4608599, "Barney's Mom - Sitting Blowjob", work="Final Fantasy",
**freyja))
self.assertEqual(len(groups), 1)
def test_the_same_character_uploaded_the_next_day_stays_apart(self):
# LazyProcrastinator 的 Lavinia 两条隔天上传，id 相差 300，是两部。
groups = self._record_clips(
self._clip(4600000, "Lavinia-cowgirl", character="lavinia (zzz)"),
self._clip(4600300, "Lavinia-doggystyle", character="lavinia (zzz)"))
self.assertEqual(len(groups), 2)
def test_clips_far_apart_or_of_another_length_stay_apart(self):
groups = self._record_clips(
self._clip(4608755, "Angel-riding hugging"), self._clip(4609755, "Angel-missionary"),
self._clip(4609759, "Angel-handjob", duration=45.0))
self.assertEqual(len(groups), 3)
def test_a_series_title_does_not_fold_different_characters(self):
# 标题开头是系列名，角色在后半截；带括号的角色标签对不上就拆开。配音演员
# `Chloeangelva (VA)` 两边都有，但它归 artist，不算角色。
groups = self._record_clips(
self._clip(4454645, "Summer Fantasy – Cindy & Lunafreya", duration=97.0,
character="cindy aurum (final fantasy)", work="Final Fantasy",
**{"Chloeangelva (VA)": "artist"}),
self._clip(4454649, "Summer Fantasy – Tifa & Aerith", duration=97.0,
character="tifa lockhart (final fantasy)", work="Final Fantasy",
**{"Chloeangelva (VA)": "artist"}))
self.assertEqual(len(groups), 2)
def test_wip_is_surfaced_on_the_group(self):
source_id = self._source()
self.store.record(source_id, _fetch([
_candidate("1", "Mitsuru School Movie"),
_candidate("2", "Mitsuru School Movie [WIP]"),
]), moment=MOMENT)
group = self.store.group(self.store.items())[0]
self.assertTrue(group.has_wip)
self.assertEqual(group.primary.variant_kind, "main")
def test_versions_posted_together_fold_by_title_family(self):
# 同一作品按清晰度分帖发：尾巴上的 `Ultra HD` 不在变体词表里，靠「键接一小段
# 尾巴、自己带变体标记、同期发布」归进同一组。
source_id = self._source()
self.store.record(source_id, _fetch([
_candidate("1", "2B Love at Sunset - 720p", published_at="2026-08-31T23:57:42Z"),
_candidate("2", "2B Love at Sunset - 1080p", published_at="2026-08-31T23:59:24Z"),
_candidate("3", "2B Love at Sunset - 4K Ultra HD Unwatermarked",
published_at="2026-09-01T00:19:38Z"),
# 续作编号、隔了几个月的相似标题、尾巴上没有变体标记的，各自成组。
_candidate("4", "Sayuri - Cowgirl [4K]", published_at="2026-09-01T00:00:00Z"),
_candidate("5", "Sayuri - Cowgirl Part 2 [4K]", published_at="2026-09-01T00:00:00Z"),
_candidate("6", "Tifa Beach [4K]", published_at="2026-05-01T00:00:00Z"),
_candidate("7", "Tifa Beach Night [4K]", published_at="2026-09-01T00:00:00Z"),
_candidate("8", "Aerith Garden", published_at="2026-09-01T00:00:00Z"),
_candidate("9", "Aerith Garden Party", published_at="2026-09-01T00:00:00Z"),
]), moment=MOMENT)
groups = self.store.group(self.store.items())
sunset = [group for group in groups
if group.primary.title.startswith("2B Love at Sunset")]
self.assertEqual(len(sunset), 1)
self.assertEqual(sorted(item.external_id for item in sunset[0].variants), ["2", "3"])
self.assertEqual(len(groups), 7)
def test_the_same_author_on_other_sites_folds_once_names_are_stripped(self):
# 来源没绑实体时，标题里夹着的作者名入库剥不掉。给了作者映射，分组前剥掉，
# 相似标题的版本也跨这位作者的来源去对；一个词的标题底下不接别的作品。
fanbox = self._source(provider="kemono", ref="fanbox/30917150", label="Pantsushi · fanbox")
video = self._source(ref="pantsushi", label="pantsushi")
self.store.record(fanbox, _fetch([
_candidate("f1", "2B Love at Sunset - 1080p", provider="kemono",
published_at="2026-08-31T23:59:24Z"),
_candidate("f2", "Kasumi Training - Alt Black 4K Ultra HD", provider="kemono",
published_at="2026-06-24T00:00:00Z"),
_candidate("f3", "Tifa Lifeguard [4K]", provider="kemono",
published_at="2026-06-20T00:00:00Z"),
], provider="kemono", ref="fanbox/30917150"), moment=MOMENT)
self.store.record(video, _fetch([
_candidate("v1", "2B Love at Sunset [pantsushi] 4K", published_at="2026-09-02T00:00:00Z"),
_candidate("v2", "Kasumi Training - Alt Black 4K", published_at="2026-06-25T00:00:00Z"),
_candidate("v3", "Tifa [pantsushi]", published_at="2026-06-23T00:00:00Z"),
# 剥完等于别的作者的标题，而这位作者自己没有这部：不借剥名去撞别人的键。
_candidate("v4", "Kyrie Canaan [pantsushi]", published_at="2026-06-01T00:00:00Z"),
], ref="pantsushi"), moment=MOMENT)
other = self._source(ref="billyhhyb", label="billyhhyb")
self.store.record(other, _fetch([
_candidate("o1", "Kyrie Canaan", published_at="2026-06-01T00:00:00Z"),
], ref="billyhhyb"), moment=MOMENT)
items = self.store.items()
pantsushi = ("name:pantsushi", frozenset({"pantsushi"}))
authors = {fanbox: pantsushi, video: pantsushi,
other: ("name:billyhhyb", frozenset({"billyhhyb"}))}
folded = {frozenset(item.external_id for item in (g.primary, *g.variants, *g.duplicates))
for g in self.store.group(items, authors)}
self.assertEqual(folded, {frozenset({"f1", "v1"}), frozenset({"f2", "v2"}),
frozenset({"f3"}), frozenset({"v3"}),
frozenset({"v4"}), frozenset({"o1"})})
self.assertEqual(len(self.store.group(items)), 8)
def test_groups_are_ordered_newest_first(self):
source_id = self._source()
self.store.record(source_id, _fetch([
_candidate("1", "Older", published_at="2026-08-01T00:00:00Z"),
_candidate("2", "Newer", published_at="2026-08-20T00:00:00Z"),
]), moment=MOMENT)
self.assertEqual([g.primary.title for g in self.store.group(self.store.items())],
["Newer", "Older"])
def test_empty_input_groups_to_nothing(self):
self.assertEqual(self.store.group(()), ())
def test_same_provider_title_collision_is_not_merged_under_work_semantics(self):
# 实测踩到的：kemono 上「February Poll Animations」和「February Poll + Animations」
# 归一化后完全相同，却是两个帖子。同站两个 main 撞车只说明标题判据到头了。
source_id = self._source(provider="kemono", ref="fanbox/30917150")
self.store.record(source_id, _fetch([
FollowCandidate(provider="kemono", external_id="10",
title="February Poll + Animations",
published_at="2026-02-15T21:27:50Z"),
FollowCandidate(provider="kemono", external_id="19",
title="February Poll Animations",
published_at="2026-01-31T22:10:13Z"),
], provider="kemono", ref="fanbox/30917150"), moment=MOMENT)
groups = self.store.group(self.store.items())
self.assertEqual(len(groups), 2)
self.assertEqual([g.variants for g in groups], [(), ()])
def test_an_alt_still_folds_when_its_provider_has_only_one_main(self):
source_id = self._source()
self.store.record(source_id, _fetch([
_candidate("1", "Fiona - Paizuri"),
_candidate("2", "Fiona - Paizuri (Nude)"),
]), moment=MOMENT)
groups = self.store.group(self.store.items())
self.assertEqual(len(groups), 1)
self.assertEqual(len(groups[0].variants), 1)
def test_f95_resource_replies_are_independent_release_groups(self):
source_id = self._source(provider="f95zone", ref="50685", semantics="release")
self.store.record(source_id, _fetch([
FollowCandidate(provider="f95zone", external_id="21383374",
title="Lazy Procrastinator Collection [2026-06-28]",
published_at="2026-08-21T04:14:09Z"),
FollowCandidate(provider="f95zone", external_id="21394555",
title="Lazy Procrastinator Collection [2026-06-28]",
published_at="2026-08-22T18:09:23Z"),
], provider="f95zone", ref="50685", semantics="release"), moment=MOMENT)
groups = self.store.group(self.store.items())
self.assertEqual(len(groups), 2)
self.assertTrue(all(group.is_release for group in groups))
self.assertTrue(all(group.variants == () for group in groups))
# 最新楼层排在前面，但不会把较早的资源楼层吞成「动态」。
self.assertEqual(groups[0].primary.external_id, "21394555")
# 标题里的日期是批次标签，不是版本：`[2026-06-28]` 不进版本位。
self.assertIsNone(groups[0].primary.version)
class SaveAssetTests(_StoreCase):
def _one_item(self, **candidate_kwargs):
entity_id = self._entity()
source_id = self._source(entity_id=entity_id)
candidate_kwargs.setdefault("url", "https://rule34video.com/video/4542713/x/")
candidate_kwargs.setdefault("duration", 20.0)
self.store.record(
source_id,
_fetch([_candidate("4542713", "Fiona - Paizuri", **candidate_kwargs)]),
moment=MOMENT)
return entity_id, self.store.items()[0]
def test_saving_requires_explicit_confirmation(self):
_, item = self._one_item()
with self.assertRaises(FollowSourceError):
self.store.save_asset(item.id)
self.assertEqual(self.connection.execute(
"SELECT count(*) FROM asset").fetchone()[0], 0)
def test_confirmed_save_creates_one_online_asset_and_links_the_creator(self):
entity_id, item = self._one_item(published_at="2026-08-18T06:23:33Z")
asset_id = self.store.save_asset(item.id, confirm=True, moment=MOMENT)
row = self.connection.execute(
"SELECT location,path,name,medium,creator,duration,release_date FROM asset"
" WHERE id=?", (asset_id,)).fetchone()
self.assertEqual(row["location"], "online")
self.assertEqual(row["path"], "https://rule34video.com/video/4542713/x/")
self.assertEqual(row["medium"], "video")
self.assertEqual(row["creator"], "LazyProcrast")
self.assertEqual(row["release_date"], "2026-08-18")
relation = self.connection.execute(
"SELECT role,source FROM asset_entity WHERE asset_id=? AND entity_id=?",
(asset_id, entity_id)).fetchone()
self.assertEqual((relation["role"], relation["source"]),
("creator", "follow:rule34video"))
self.assertEqual(self.store.items()[0].status, "saved")
def test_saving_twice_is_idempotent(self):
_, item = self._one_item()
first = self.store.save_asset(item.id, confirm=True, moment=MOMENT)
second = self.store.save_asset(item.id, confirm=True, moment=MOMENT)
self.assertEqual(first, second)
self.assertEqual(self.connection.execute(
"SELECT count(*) FROM asset").fetchone()[0], 1)
def test_an_existing_online_asset_is_reused_not_duplicated(self):
_, item = self._one_item()
self.connection.execute(
"INSERT INTO asset(location,path,name) VALUES('online',?,'existing')",
(item.url,))
asset_id = self.store.save_asset(item.id, confirm=True, moment=MOMENT)
self.assertEqual(self.connection.execute(
"SELECT name FROM asset WHERE id=?", (asset_id,)).fetchone()[0], "existing")
def test_kemono_posts_without_duration_are_saved_as_illustrations(self):
source_id = self._source(provider="kemono", ref="fanbox/30917150")
self.store.record(source_id, _fetch(
[FollowCandidate(provider="kemono", external_id="11406814", title="VVD 7",
url="https://kemono.cr/fanbox/user/30917150/post/11406814")],
provider="kemono", ref="fanbox/30917150"), moment=MOMENT)
asset_id = self.store.save_asset(self.store.items()[0].id, confirm=True,
moment=MOMENT)
self.assertEqual(self.connection.execute(
"SELECT medium FROM asset WHERE id=?", (asset_id,)).fetchone()[0],
"illustration")
def test_a_candidate_without_a_page_url_cannot_be_saved(self):
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "No URL")]), moment=MOMENT)
with self.assertRaises(FollowSourceError):
self.store.save_asset(self.store.items()[0].id, confirm=True)
def test_missing_item_is_an_error(self):
with self.assertRaises(FollowSourceError):
self.store.save_asset(9999, confirm=True)
class MediaReparseTests(_StoreCase):
"""`source_needs_media_reparse` 的判定口径。
这个判定交给 SQLite 的 `json_extract`，不把整个来源的 `metadata_json` 取回
Python 再逐条 `json.loads`：回填过的来源单源上千行，为一个布尔值全解一遍。
下推到 SQL 要守住同样的容忍度。
"""
def _write_metadata(self, source_id: int, external_id: str, raw) -> None:
self.connection.execute(
"INSERT INTO follow_item(source_id,external_id,release_key,title,url,"
"metadata_json,first_seen_at,last_seen_at) VALUES(?,?,?,?,?,?,?,?)",
(source_id, external_id, external_id, external_id,
f"https://x.test/{external_id}",
raw, "2026-08-25T00:00:00Z", "2026-08-25T00:00:00Z"))
def test_no_item_needs_a_reparse_on_an_empty_source(self):
self.assertFalse(self.store.source_needs_media_reparse(self._source()))
def test_one_flagged_item_is_enough(self):
source_id = self._source()
self._write_metadata(source_id, "a", json.dumps({"markers": []}))
self._write_metadata(source_id, "b",
json.dumps({"media_needs_credential": True}))
self.assertTrue(self.store.source_needs_media_reparse(source_id))
def test_an_explicit_false_does_not_count(self):
source_id = self._source()
self._write_metadata(source_id, "a",
json.dumps({"media_needs_credential": False}))
self.assertFalse(self.store.source_needs_media_reparse(source_id))
def test_unparsable_or_odd_shaped_metadata_is_skipped_not_fatal(self):
"""非法 JSON 只是「这条不知道」，不该让整个检查崩掉。
`json_extract` 遇到非法 JSON 是报错而不是返回 NULL，所以查询里那层
`json_valid` 不能省：它顶的是 `except JSONDecodeError: continue` 那一级容忍。
列上是 `NOT NULL DEFAULT '{}'`，所以「没有元数据」落盘的形状是 `{}` 和
空字串，不是 NULL；空字串并不是合法 JSON，同样得被 `json_valid` 拦住。
"""
source_id = self._source()
self._write_metadata(source_id, "a", "not json at all")
self._write_metadata(source_id, "b", "")
self._write_metadata(source_id, "c", "[1,2,3]")
self._write_metadata(source_id, "d", "{}")
self.assertFalse(self.store.source_needs_media_reparse(source_id))
self._write_metadata(source_id, "e",
json.dumps({"media_needs_credential": True}))
self.assertTrue(self.store.source_needs_media_reparse(source_id))
def test_the_flag_does_not_leak_across_sources(self):
first = self._source()
second = self._source(ref="someone-else")
self._write_metadata(second, "a",
json.dumps({"media_needs_credential": True}))
self.assertFalse(self.store.source_needs_media_reparse(first))
self.assertTrue(self.store.source_needs_media_reparse(second))
class ReleaseGroupTests(unittest.TestCase):
def test_group_is_immutable(self):
group = ReleaseGroup("k", None, (), ())
with self.assertRaises(Exception):
group.release_key = "other"
class AuthorIdentityTests(_StoreCase):
"""作者别名表的写入口径。这些 SQL 归存储层，不写在 Web 处理函数里。"""
def test_preferred_name_preserves_all_spellings_and_the_group(self):
current = "Suzutaru 3D (鈴太郎3D)"
self.store.upsert_author_alias(current, "suzutaro3d", source="manual")
self.store.upsert_author_alias(current, "鈴太郎3D", source="manual")
result = self.store.rename_author_group(normalized_author_name(current), "鈴太郎3D")
mapping, groups = self.store.author_aliases()
self.assertEqual(result, {"previous": current, "canonical": "鈴太郎3D"})
self.assertEqual(groups[0]["canonical_name"], "鈴太郎3D")
self.assertEqual({row["name"] for row in groups[0]["aliases"]}, {current, "suzutaro3d"})
self.assertEqual(set(mapping.values()), {normalized_author_name("鈴太郎3D")})
def test_preferred_name_rejects_other_groups_and_missing_groups(self):
self.store.upsert_author_alias("作者 A", "alias A", source="manual")
self.store.upsert_author_alias("作者 B", "alias B", source="manual")
before = self.store.author_aliases()
for key, name in ((normalized_author_name("作者 A"), "alias B"), ("missing", "名称"),
(normalized_author_name("作者 A"), "")):
with self.subTest(key=key, name=name), self.assertRaises(ValueError):
self.store.rename_author_group(key, name)
self.assertEqual(self.store.author_aliases(), before)
def test_the_container_suffix_is_not_part_of_the_author_name(self):
"""F95 的线程标题说的是一个容器，不是另一个作者。
真实数据是 `Lazy Procrastinator Collection`，而每一条作者来源都是
`LazyProcrastinator`；留着这个通用后缀会凭空多出一个分组。
"""
self.assertEqual(
normalized_author_name("Lazy Procrastinator Collection",
provider="f95zone"),
normalized_author_name("LazyProcrastinator"))
# 只对 F95 成立：别的站上 `Collection` 可能真是名字的一部分。
self.assertNotEqual(
normalized_author_name("Lazy Procrastinator Collection"),
normalized_author_name("LazyProcrastinator"))
def test_the_service_suffix_only_says_where_they_publish(self):
self.assertEqual(normalized_author_name("LazyProcrastinator · fanbox"),
normalized_author_name("lazyprocrastinator"))
self.assertEqual(author_display_text("Billyhhyb · patreon"), "Billyhhyb")
def test_the_f95_author_sits_in_the_last_bracket(self):
"""站点的标题约定是 `作品名 [版本或日期] [作者]`。"""
self.assertEqual(
author_display_text("Strauzek Collection [2026-09-04] [Mr_Strauz]",
provider="f95zone"), "Mr_Strauz")
self.assertEqual(
author_display_text("Some Game [v1.2] [Final] [DevName]",
provider="f95zone"), "DevName")
def test_a_title_without_an_author_bracket_falls_back_to_its_stem(self):
# `[Collection Request]` 是版块标签，`Complete Collection` 是容器措辞，
# 剥掉之后剩下的才是这个人。
self.assertEqual(
author_display_text(
"[Collection Request] Suzutaru 3D - Complete Collection",
provider="f95zone"), "Suzutaru 3D")
self.assertEqual(
author_display_text("Memz3D Models Collection", provider="f95zone"),
"Memz3D")
def test_both_handles_in_the_author_bracket_are_kept(self):
# 一个人常在作者位上写两个手柄，第一个当显示名，另一个够格做别名候选。
self.assertEqual(
f95_author_handles(
"Lazy Procrastinator Collection [2026-06-28]"
" [LazyProcrastinator/LazyProcrast]"),
("LazyProcrastinator", "LazyProcrast"))
self.assertEqual(
f95_author_name(
"Lazy Procrastinator Collection [2026-06-28]"
" [LazyProcrastinator/LazyProcrast]"), "LazyProcrastinator")
def test_the_bracket_rule_is_only_for_f95(self):
# 别的站上方括号可能真是标签的一部分，不能拿 F95 的约定去套。
self.assertEqual(
author_display_text("Strauzek Collection [2026-09-04] [Mr_Strauz]"),
"Strauzek Collection [2026-09-04] [Mr_Strauz]")
def test_the_f95_author_key_matches_the_name_on_the_other_sites(self):
self.assertEqual(
normalized_author_name("Strauzek Collection [2026-09-04] [Mr_Strauz]",
provider="f95zone"),
normalized_author_name("mr_strauz"))
def test_a_manual_alias_maps_both_names_to_one_canonical_key(self):
self.store.upsert_author_alias("Initiala", "ffxivinitiala", source="manual",
moment=MOMENT)
mapping, groups = self.store.author_aliases()
self.assertEqual(mapping["ffxivinitiala"], "initiala")
self.assertEqual(mapping["initiala"], "initiala")
self.assertEqual([group["canonical_name"] for group in groups], ["Initiala"])
self.assertEqual([alias["name"] for alias in groups[0]["aliases"]],
["ffxivinitiala"])
def test_automatic_evidence_never_overwrites_a_decision(self):
"""人工确认可以有意重新分组；自动证据只填此前未知的手柄。"""
self.store.upsert_author_alias("Initiala", "ffxivinitiala", source="manual",
moment=MOMENT)
self.assertIsNone(self.store.upsert_author_alias(
"Someone Else", "ffxivinitiala", source="official:fanbox", moment=MOMENT))
mapping, _groups = self.store.author_aliases()
self.assertEqual(mapping["ffxivinitiala"], "initiala")
def test_two_names_that_normalize_the_same_are_not_an_alias(self):
for canonical, alias in (("Initiala", "initi-ala"), ("", "x"), ("x", "")):
with self.assertRaises(ValueError):
self.store.upsert_author_alias(canonical, alias, source="manual")
def test_the_canonical_name_cannot_be_removed_as_an_alias(self):
"""删规范名会拆散整个组，剩下的别名指向一个不存在的键。"""
self.store.upsert_author_alias("Initiala", "ffxivinitiala", source="manual",
moment=MOMENT)
with self.assertRaises(ValueError):
self.store.remove_author_alias("Initiala")
with self.assertRaises(ValueError):
self.store.remove_author_alias("never-registered")
with self.assertRaises(ValueError):
self.store.remove_author_alias("")
self.store.remove_author_alias("ffxivinitiala")
self.assertEqual(self.store.author_aliases(), ({"initiala": "initiala"}, []))
def test_an_official_handle_is_learned_only_from_one_unambiguous_author(self):
learned = self.store.learn_official_author_alias(
"fanbox", "ffxivinitiala",
(_candidate("1", "a", provider="fanbox", author="Initiala"),))
self.assertEqual(learned["alias"], "ffxivinitiala")
self.assertEqual(learned["source"], "official:fanbox")
# 归档站的 ref 是数字 id，不是名字。
self.assertIsNone(self.store.learn_official_author_alias(
"kemono", "fanbox/30917150",
(_candidate("1", "a", provider="kemono", author="Initiala"),)))
def test_iwara_author_profile_provides_a_trusted_handle_alias(self):
learned = self.store.learn_official_author_alias(
"iwara", "user_creator",
(_candidate("1", "Work", provider="iwara", author="Test Creator"),))
self.assertEqual(learned["alias"], "user_creator")
self.assertEqual(learned["source"], "official:iwara")
class SourceRemovalTests(_StoreCase):
def test_removing_a_source_takes_it_out_of_the_listing(self):
source_id = self._source()
other = self._source(provider="rule34xxx", ref="tag")
self.store.remove_source(source_id)
self.assertEqual([row["id"] for row in self.store.sources()], [other])
def test_removing_a_source_takes_its_items_and_their_playback(self):
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "Fiona")]), moment=MOMENT)
self.store.record_playback(self.store.items()[0].id, MOMENT)
self.store.remove_source(source_id)
for table in ("follow_item", "follow_playback"):
self.assertEqual(self.connection.execute(
f"SELECT count(*) FROM {table}").fetchone()[0], 0, table)
self.assertEqual(self.connection.execute("PRAGMA foreign_key_check").fetchall(), [])
class PlaybackTests(_StoreCase):
def _item(self):
source_id = self._source()
self.store.record(source_id, _fetch([_candidate("1", "Fiona")]),
moment=MOMENT)
return self.store.items()[0].id
def test_playing_a_new_item_marks_it_seen_once(self):
item_id = self._item()
self.assertEqual(self.store.record_playback(item_id, MOMENT), "seen")
# 已经播过的不再改状态：手动标成 ignored 之后再播不该被拉回 seen。
self.store.set_status(item_id, "ignored")
self.assertEqual(self.store.record_playback(item_id, MOMENT), "ignored")
count = self.connection.execute(
"SELECT play_count FROM follow_playback WHERE follow_item_id=?",
(item_id,)).fetchone()[0]
self.assertEqual(count, 2)
def test_activity_accumulates_time_and_keeps_the_furthest_point(self):
item_id = self._item()
first = self.store.record_playback_activity(
item_id, position=30, duration=100, delta=30, moment=MOMENT)
self.assertEqual(first, {"play_seconds": 30.0, "max_reached": 0.3})
# 往回拖再看一遍：时长累加，最远位置不倒退。
second = self.store.record_playback_activity(
item_id, position=10, duration=100, delta=10, moment=MOMENT)
self.assertEqual(second, {"play_seconds": 40.0, "max_reached": 0.3})
ended = self.store.record_playback_activity(
item_id, position=99, duration=100, delta=1, ended=True, moment=MOMENT)
self.assertEqual(ended["max_reached"], 1.0)
def test_an_unknown_item_is_a_request_error_not_a_storage_failure(self):
"""调用方把 ValueError 映射成 400；这里不能变成 500。"""
for call in (lambda: self.store.record_playback(9999),
lambda: self.store.record_playback_activity(9999)):
with self.assertRaises(ValueError):
call()
def test_nonsense_numbers_do_not_reach_the_database(self):
item_id = self._item()
result = self.store.record_playback_activity(
item_id, position=-5, duration=0, delta=-3, moment=MOMENT)
self.assertEqual(result, {"play_seconds": 0.0, "max_reached": 0.0})
class PartialCandidateTests(_StoreCase):
"""`partial=True` 的候选（跳过了第二阶段）落库时不能抹掉已有的细节。
upsert 不整块替换 metadata、media_url、thumb_url 和 group_hint：第二阶段一旦
按「已补齐」跳过，同一条再落一次就会把上一轮取到的细节清成空——而清空之后
「详情本来就没有」和「这次没去问」在数据里长得一模一样。
"""
def _both(self):
source_id = self._source(provider="rule34paheal", ref="initiala")
full = FollowCandidate(
provider="rule34paheal", external_id="7428820", title="Tifa",
url="https://rule34.paheal.net/post/view/7428820",
media_url="https://r34i.paheal-cdn.net/df/fb/video",
thumb_url="https://r34t.paheal.net/df/fb/poster",
published_at="2026-08-26T15:21:00Z", duration=28.4, author="VHSephi",
group_hint="subscribestar:2639932",
extra={"tag": "initiala", "source": "https://subscribestar.adult/posts/2639932",
"tag_types": {"tifa_lockhart": "general"}})
partial = FollowCandidate(
provider="rule34paheal", external_id="7428820", title="Tifa Lockhart",
url="https://rule34.paheal.net/post/view/7428820",
media_url="https://r34i.paheal-cdn.net/df/fb/listing",
thumb_url="https://r34t.paheal.net/df/fb/thumb",
group_hint="rule34paheal:post:7428820", partial=True,
extra={"tag": "initiala", "media_kind": "video",
"tag_types": {"amina": "general"}})
return source_id, full, partial
def _item(self, source_id):
return self.store.items(source_id=source_id)[0]
def _record(self, source_id, candidate):
self.store.record(source_id, _fetch([candidate], provider="rule34paheal",
ref="initiala"), moment=MOMENT)
def test_a_partial_row_keeps_every_detail_only_column(self):
source_id, full, partial = self._both()
self._record(source_id, full)
self._record(source_id, partial)
row = self._item(source_id)
self.assertEqual(row.published_at, "2026-08-26T15:21:00Z")
self.assertEqual(row.duration, 28.4)
self.assertEqual(row.media_url, "https://r34i.paheal-cdn.net/df/fb/video")
self.assertEqual(row.thumb_url, "https://r34t.paheal.net/df/fb/poster")
self.assertEqual(row.group_hint, "subscribestar:2639932")
# 列表本来就权威的那几列照常更新。
self.assertEqual(row.title, "Tifa Lockhart")
def test_a_partial_row_merges_metadata_instead_of_replacing_it(self):
source_id, full, partial = self._both()
self._record(source_id, full)
self._record(source_id, partial)
metadata = self._item(source_id).metadata
self.assertEqual(metadata["source"], "https://subscribestar.adult/posts/2639932",
"详情给出的出处不在这次的补丁里，就该原样留着")
self.assertEqual(metadata["media_kind"], "video")
self.assertEqual(metadata["tag_types"],
{"tifa_lockhart": "general", "amina": "general"})
def test_a_complete_row_still_replaces_what_it_carries(self):
"""第二阶段成功时是完整视图，该覆盖就覆盖——否则改正过的细节写不进去。"""
source_id, full, partial = self._both()
self._record(source_id, partial)
self._record(source_id, full)
row = self._item(source_id)
self.assertEqual(row.thumb_url, "https://r34t.paheal.net/df/fb/poster")
self.assertEqual(row.group_hint, "subscribestar:2639932")
self.assertEqual(row.metadata["tag_types"], {"tifa_lockhart": "general"})
class ImageDimsTests(_StoreCase):
"""图片宽高是给图片墙占位用的比例，三处来源（接口、文件头、界面回写）同权。"""
def _record(self, source_id, candidate):
self.store.record(source_id, _fetch([candidate], provider="kemono", ref="a"),
moment=MOMENT)
return self.store.items(source_id=source_id)[0]
def _image(self, **extra):
return FollowCandidate(
provider="kemono", external_id="1", title="Sketch",
url="https://kemono.cr/fanbox/user/1/post/1",
media_url="https://kemono.cr/data/ab/cd/abcd.jpg",
thumb_url="https://kemono.cr/thumbnail/data/ab/cd/abcd.jpg", extra=extra)
def test_item_dims_fill_only_a_gap_and_say_whether_they_wrote(self):
source_id = self._source(provider="kemono", ref="a")
item = self._record(source_id, self._image())
self.assertTrue(self.store.set_image_dims(item.id, 800, 600))
self.assertFalse(self.store.set_image_dims(item.id, 1024, 768),
"已有尺寸不覆盖：三处给的都是同一张图的比例，没有谁更权威")
metadata = self.store.item(item.id).metadata
self.assertEqual((metadata["width"], metadata["height"]), (800, 600))
self.assertFalse(self.store.set_image_dims(item.id + 99, 800, 600))
with self.assertRaises(FollowSourceError):
self.store.set_image_dims(item.id, 0, 600)
def test_media_dims_land_on_that_one_media_only(self):
source_id = self._source(provider="kemono", ref="a")
item = self._record(source_id, self._image(media_items=[
{"id": "a", "media_kind": "image", "url": "https://kemono.cr/data/a.png"},
{"id": "b", "media_kind": "image", "url": "https://kemono.cr/data/b.png"},
]))
self.assertTrue(self.store.set_image_dims(item.id, 1920, 1080, media_index=1))
self.assertFalse(self.store.set_image_dims(item.id, 1, 1, media_index=5))
media_items = self.store.item(item.id).metadata["media_items"]
self.assertNotIn("width", media_items[0])
self.assertEqual((media_items[1]["width"], media_items[1]["height"]), (1920, 1080))
self.assertNotIn("width", self.store.item(item.id).metadata,
"媒体级尺寸不该顺手写成条目级")
def test_partial_item_and_media_dimensions_accept_a_complete_pair(self):
source_id = self._source(provider="kemono", ref="a")
item = self._record(source_id, self._image(width=800, media_items=[
{"id": "a", "media_kind": "image", "url": "https://kemono.cr/data/a.png", "width": 800},
{"id": "b", "media_kind": "image", "url": "https://kemono.cr/data/b.png", "height": 600},
]))
self.assertTrue(self.store.set_image_dims(item.id, 800, 600))
for index in (0, 1):
self.assertTrue(self.store.set_image_dims(item.id, 800, 600, media_index=index))
self.assertFalse(self.store.set_image_dims(item.id, 1, 1, media_index=index))
metadata = self.store.item(item.id).metadata
self.assertEqual((metadata["width"], metadata["height"]), (800, 600))
self.assertEqual([(media["width"], media["height"]) for media in metadata["media_items"]],
[(800, 600), (800, 600)])
def test_a_full_refetch_keeps_learned_dims_until_the_source_reports_its_own(self):
source_id = self._source(provider="kemono", ref="a")
item = self._record(source_id, self._image(tags="a"))
self.store.set_image_dims(item.id, 800, 600)
# 下一轮检查更新是完整候选，走整块替换 metadata 的那条 SET。
metadata = self._record(source_id, self._image(tags="a b")).metadata
self.assertEqual(metadata["tags"], "a b")
self.assertEqual((metadata["width"], metadata["height"]), (800, 600))
# 来源自己开始报尺寸时以来源为准。
metadata = self._record(source_id, self._image(width=1600, height=1200)).metadata
self.assertEqual((metadata["width"], metadata["height"]), (1600, 1200))
def test_a_full_refetch_keeps_learned_media_dims_by_media_identity(self):
"""清单里每张图学到的尺寸按媒体稳定键对回：作者在前面插了一张，尺寸不串位。"""
source_id = self._source(provider="kemono", ref="a")
item = self._record(source_id, self._image(media_items=[
{"id": "a", "media_kind": "image", "url": "https://kemono.cr/data/a.png"},
{"id": "b", "media_kind": "image", "url": "https://kemono.cr/data/b.png"},
]))
self.store.set_image_dims(item.id, 1920, 1080, media_index=1)
media_items = self._record(source_id, self._image(media_items=[
{"id": "new", "media_kind": "image", "url": "https://kemono.cr/data/new.png"},
{"id": "a", "media_kind": "image", "url": "https://kemono.cr/data/a.png"},
{"id": "b", "media_kind": "image", "url": "https://kemono.cr/data/b.png",
"width": 640, "height": 480},
{"id": "c", "media_kind": "image", "url": "https://kemono.cr/data/c.png"},
])).metadata["media_items"]
self.assertNotIn("width", media_items[0])
self.assertNotIn("width", media_items[1])
self.assertEqual((media_items[2]["width"], media_items[2]["height"]), (640, 480),
"来源自己报了尺寸时以来源为准")
self.assertNotIn("width", media_items[3])
media_items = self._record(source_id, self._image(media_items=[
{"id": "b", "media_kind": "image", "url": "https://kemono.cr/data/b.png"},
])).metadata["media_items"]
self.assertEqual((media_items[0]["width"], media_items[0]["height"]), (640, 480))
def test_a_refetched_media_tuple_keeps_learned_dims(self):
"""f95 的清单是附件与 gofile 两段拼成的 tuple，学到的尺寸照样对回。"""
source_id = self._source(provider="kemono", ref="a")
attachment = {"id": "f95-attachment-1", "media_kind": "image",
"url": "https://attachments.f95zone.to/2026/10/1_a.png"}
item = self._record(source_id, self._image(media_items=[attachment]))
self.store.set_image_dims(item.id, 2000, 1125, media_index=0)
media_items = self._record(source_id, self._image(media_items=(attachment,))).metadata["media_items"]
self.assertEqual((media_items[0]["width"], media_items[0]["height"]), (2000, 1125))
class EnrichedMarkTests(_StoreCase):
"""「这一行不必再打详情页」怎么从 ledger 里读出来。"""
def test_the_paheal_mark_is_a_published_time_the_detail_page_gave(self):
source_id = self._source(provider="rule34paheal", ref="initiala")
self.store.record(source_id, _fetch([
FollowCandidate(provider="rule34paheal", external_id="1", title="a",
published_at="2026-08-26T15:21:00Z"),
FollowCandidate(provider="rule34paheal", external_id="2", title="b",
partial=True),
], provider="rule34paheal", ref="initiala"), moment=MOMENT)
self.assertEqual(self.store.enriched_external_ids(source_id, "published_at"),
frozenset({"1"}))
def test_the_rule34xxx_mark_is_the_taxonomy_only_the_detail_page_gives(self):
source_id = self._source(provider="rule34xxx", ref="tag")
self.store.record(source_id, _fetch([
FollowCandidate(provider="rule34xxx", external_id="1", title="a",
published_at="2026-08-26T15:21:00Z",
extra={"tag_types": {"tifa": "general"}}),
# 上传时间来自列表，详情页这次被挡回来了：这一条还没补齐。
FollowCandidate(provider="rule34xxx", external_id="2", title="b",
published_at="2026-08-26T15:20:00Z", partial=True,
extra={"tag": "tag"}),
], provider="rule34xxx", ref="tag"), moment=MOMENT)
self.assertEqual(self.store.enriched_external_ids(source_id, "tag_types_duration"),
frozenset({"1"}))
def test_a_rule34xxx_video_without_duration_is_not_yet_enriched(self):
"""分类有了、mp4 的时长没读到，下次检查还要进第二阶段；图片不需要时长。"""
source_id = self._source(provider="rule34xxx", ref="tag")
types = {"tag_types": {"tifa": "general"}}
self.store.record(source_id, _fetch([
FollowCandidate(provider="rule34xxx", external_id="clip", title="a",
media_url="https://api-cdn-mp4.rule34.xxx/images/1/a.mp4",
extra=types),
FollowCandidate(provider="rule34xxx", external_id="timed", title="b",
media_url="https://api-cdn-mp4.rule34.xxx/images/1/b.mp4",
duration=12.5, extra=types),
FollowCandidate(provider="rule34xxx", external_id="still", title="c",
media_url="https://api-cdn.rule34.xxx/images/1/c.jpeg",
extra=types),
], provider="rule34xxx", ref="tag"), moment=MOMENT)
self.assertEqual(self.store.enriched_external_ids(source_id, "tag_types_duration"),
frozenset({"timed", "still"}))
def test_a_partial_candidate_fills_an_empty_duration_without_overwriting(self):
"""帖子页被挡回来时候选仍是 partial，文件头读到的时长只补空着的那一格。"""
source_id = self._source(provider="rule34xxx", ref="tag")
self.store.record(source_id, _fetch([
FollowCandidate(provider="rule34xxx", external_id="empty", title="a"),
FollowCandidate(provider="rule34xxx", external_id="known", title="b",
duration=30.0),
], provider="rule34xxx", ref="tag"), moment=MOMENT)
self.store.record(source_id, _fetch([
FollowCandidate(provider="rule34xxx", external_id="empty", title="a",
partial=True, duration=10.0),
FollowCandidate(provider="rule34xxx", external_id="known", title="b",
partial=True, duration=99.0),
], provider="rule34xxx", ref="tag"), moment=MOMENT)
durations = dict(self.connection.execute(
"SELECT external_id, duration FROM follow_item WHERE source_id=?", (source_id,)))
self.assertEqual(durations, {"empty": 10.0, "known": 30.0})
def test_a_rule34video_row_stays_detailed_when_the_list_sees_it_again(self):
"""补齐过的 rule34video 行再被列表看到时不打详情页，详情给的值都要留住。"""
source_id = self._source(provider="rule34video", ref="artist")
self.store.record(source_id, _fetch([
FollowCandidate(provider="rule34video", external_id="1", title="a",
media_url="https://rule34video.com/get_file/1_720p.mp4/",
published_at="2026-08-18T00:00:00Z",
extra={"media_kind": "video", "published_precision": "exact",
"tag_types": {"3D": "metadata"}}),
FollowCandidate(provider="rule34video", external_id="2", title="b",
media_url="https://rule34video.com/get_file/2_preview.mp4/",
partial=True, extra={"media_kind": "preview_clip"}),
], provider="rule34video", ref="artist"), moment=MOMENT)
self.assertEqual(self.store.enriched_external_ids(source_id, "tag_types"),
frozenset({"1"}))
self.store.record(source_id, _fetch([
FollowCandidate(provider="rule34video", external_id="1", title="a",
media_url="https://rule34video.com/get_file/1_preview.mp4/",
published_at="2026-08-10T00:00:00Z", partial=True,
extra={"added_text": "2 months ago"}),
], provider="rule34video", ref="artist"), moment=MOMENT + timedelta(minutes=1))
media_url, published_at, metadata = self.connection.execute(
"SELECT media_url, published_at, metadata_json FROM follow_item"
" WHERE source_id=? AND external_id='1'", (source_id,)).fetchone()
self.assertEqual(media_url, "https://rule34video.com/get_file/1_720p.mp4/")
self.assertEqual(published_at, "2026-08-18T00:00:00Z")
self.assertEqual(json.loads(metadata)["media_kind"], "video")
self.assertEqual(self.store.enriched_external_ids(source_id, "tag_types"),
frozenset({"1"}))
def test_every_kemono_row_in_the_ledger_counts_as_judged(self):
source_id = self._source(provider="kemono", ref="fanbox/1")
self.store.record(source_id, _fetch([
FollowCandidate(provider="kemono", external_id="10", title="a"),
FollowCandidate(provider="kemono", external_id="11", title="b", partial=True),
], provider="kemono", ref="fanbox/1"), moment=MOMENT)
self.assertEqual(self.store.enriched_external_ids(source_id, "kept"),
frozenset({"10", "11"}))
def test_an_unregistered_mark_is_refused_not_silently_matched(self):
source_id = self._source()
with self.assertRaises(ValueError):
self.store.enriched_external_ids(source_id, "whatever")
class StoredCompilationTests(_StoreCase):
"""判据收紧之前入库的跨作者打包，按同一判据清退。"""
def _record(self, external_id, title, models):
source_id = self._source()
self.store.record(source_id, _fetch([
_candidate(external_id, title, extra={"models": list(models),
"model_count": len(models)}),
]), moment=MOMENT)
return self.store.items()[0]
def test_voice_and_audio_credits_keep_a_single_creator_work_out_of_the_list(self):
self._record("1", "Yunara Showing Ahri Some Discipline",
["Iidssm", "Adaline (VA)", "GeminiStarsign1 (VA)",
"Huntress___ (Audio/SFX)", "HentAudio (Audio)"])
self.assertEqual(self.store.collected_compilations(), ())
def test_a_cross_artist_pack_is_listed_with_both_counts(self):
self._record("2", "Fuck Track / Futa PMV", [f"M{n}" for n in range(13)])
listed = self.store.collected_compilations()
self.assertEqual(len(listed), 1)
self.assertEqual((listed[0].credited, listed[0].visual), (13, 13))
self.assertEqual(listed[0].external_id, "2")
def test_an_item_already_saved_as_an_asset_is_left_alone(self):
item = self._record("3", "ON AND ON | HMV / PMV", [f"M{n}" for n in range(17)])
self.connection.execute("UPDATE follow_item SET url=? WHERE id=?",
("https://rule34video.com/video/3/x/", item.id))
self.store.save_asset(item.id, confirm=True, moment=MOMENT)
self.assertEqual(self.store.collected_compilations(), ())
def test_purging_requires_explicit_confirmation(self):
self._record("4", "Resident Evil - The Fallen Saga",
[f"M{n}" for n in range(14)])
listed = self.store.collected_compilations()
with self.assertRaises(FollowSourceError):
self.store.purge_compilations(listed)
self.assertEqual(self.connection.execute(
"SELECT count(*) FROM follow_item").fetchone()[0], 1)
def test_a_confirmed_purge_takes_the_playback_rows_with_it(self):
item = self._record("5", "triss blacked censored", [f"M{n}" for n in range(13)])
self.store.record_playback(item.id)
listed = self.store.collected_compilations()
self.assertEqual(self.store.purge_compilations(listed, confirm=True), 1)
self.assertEqual(self.connection.execute(
"SELECT count(*) FROM follow_item").fetchone()[0], 0)
self.assertEqual(self.connection.execute(
"SELECT count(*) FROM follow_playback").fetchone()[0], 0)
def _old_snapshot(directory: Path, index: int, size: int = 10) -> Path:
"""升级前留下的一份快照：`.raw` 加同名 `.json`，名字按序号排出先后。"""
directory.mkdir(parents=True, exist_ok=True)
raw = directory / f"20260801T{index:06d}.000000Z-{index:064x}.raw"
raw.write_bytes(b"x" * size)
raw.with_suffix(".json").write_bytes(b"{}")
return raw
class EvidenceRetentionTests(_StoreCase):
"""原始快照：相同的不重写，每个来源只留最近 3 份和被条目引用的，存量等用户确认才删。"""
def _evidence(self):
return [row[0] for row in self.connection.execute("SELECT evidence_path FROM follow_item")]
def _raws(self, provider="rule34video"):
return sorted((self.root / "sources" / "follow" / provider).rglob("*.raw"))
def test_an_unchanged_page_reuses_the_latest_snapshot(self):
source_id = self._source()
fetch = _fetch([_candidate("1", "A")])
first = self.store.record(source_id, fetch, moment=MOMENT)
again = self.store.record(source_id, fetch, moment=MOMENT + timedelta(hours=1))
self.assertEqual(again.evidence_path, first.evidence_path)
self.assertEqual(len(self._raws()), 1)
changed = self.store.record(source_id, replace(fetch, raw_body=b"<html>new</html>"),
moment=MOMENT + timedelta(hours=2))
self.assertNotEqual(changed.evidence_path, first.evidence_path)
repeat = self.store.record(source_id, fetch, moment=MOMENT + timedelta(hours=3))
self.assertNotEqual(repeat.evidence_path, first.evidence_path,
"只和最新一份比：页面改回旧样子也记一份新的")
self.assertEqual(len(self._raws()), 3)
self.assertEqual(self._evidence(), [first.evidence_path])
def test_each_source_keeps_three_recent_snapshots_and_the_referenced_ones(self):
other = self.root / "sources" / "follow" / "rule34video" / "another-source"
untouched = [_old_snapshot(other, index) for index in range(5)]
source_id = self._source()
paths = []
for index in range(5):
outcome = self.store.record(
source_id, _fetch([_candidate("1", "A")], raw=f"<html>{index}</html>".encode()),
moment=MOMENT + timedelta(hours=index))
paths.append(self.root / "sources" / Path(outcome.evidence_path))
self.assertEqual(self._evidence(), [str(paths[0].relative_to(self.root / "sources"))])
self.assertEqual([path.exists() for path in paths], [True, False, True, True, True])
self.assertFalse(paths[1].with_suffix(".json").exists())
self.assertTrue(all(path.exists() for path in untouched), "只清这次写入的那个来源目录")
def test_a_new_snapshot_only_offsets_its_own_growth_and_leaves_the_stock(self):
"""升级后第一次检查不能把存量一口气删光：那在写事务里，也绕过了「修复与迁移」的确认。"""
source_id = self._source()
fetch = _fetch([_candidate("1", "A")])
directory = (self.root / "sources" / Path(self.store.record(source_id, fetch, moment=MOMENT)
.evidence_path)).parent
stock = [_old_snapshot(directory, index) for index in range(20)]
self.store.record(source_id, replace(fetch, raw_body=b"<html>new</html>"),
moment=MOMENT + timedelta(hours=1))
self.assertEqual(sum(not raw.exists() for raw in stock), 1)
self.assertEqual(len(list(directory.glob("*.raw"))), 22 - 1)
self.store.record(source_id, replace(fetch, raw_body=b"<html>new</html>"),
moment=MOMENT + timedelta(hours=2))
self.assertEqual(sum(not raw.exists() for raw in stock), 1, "去重命中时没写新的，也不删")
def test_stock_snapshots_wait_for_the_user_and_keep_what_items_reference(self):
from peach import repair_center
from peach.follow_evidence import Reclaim
from peach.repository import LedgerDatabase
source_id = self._source()
directory = self.root / "sources" / "follow" / "rule34video" / "legacy"
raws = [_old_snapshot(directory, index, size=100) for index in range(6)]
# 迁移数据根前留下的绝对路径、反斜杠也认得出是同一个文件。
self.connection.execute(
"INSERT INTO follow_item(source_id,external_id,title,release_key,evidence_path,first_seen_at,"
"last_seen_at) VALUES(?,?,?,'old',?,?,?)", (source_id, "old", "Old", "D:\\peach-data\\sources\\follow\\rule34video"
f"\\legacy\\{raws[1].name}", "2026-08-01T00:00:00Z", "2026-08-01T00:00:00Z"))
self.connection.commit()
from peach.task_runs import TaskRunStore
database = LedgerDatabase(self.root / "ledger.db")
runs = TaskRunStore(database)
reclaim = Reclaim(self.root / "sources" / "follow", database, runs=runs)
def settle():
while reclaim.worker is not None and reclaim.worker.is_alive():
reclaim.worker.join(10)
def item():
with database.read_connection() as connection:
center = repair_center.snapshot(connection, {}, cloud_pending=0,
follow_evidence_status=reclaim.status())
return next((row for row in center["items"] if row["id"] == "follow-evidence"), None), center
self.assertTrue(item()[0]["estimating"])
settle()
waiting, _ = item()
# 留 3、4、5 三份最近的和被引用的 1，删 0、2 两份，每份 100 + 2 字节。
self.assertEqual((waiting["mode"], waiting["state"], waiting["files"], waiting["bytes"]),
("confirm", "waiting", 4, 204))
self.assertTrue(all(raw.exists() for raw in raws), "估算不删文件")
reclaim.start()
settle()
self.assertEqual([raw.exists() for raw in raws], [False, True, False, True, True, True])
done, center = item()
self.assertEqual((done["state"], done["changed"]), ("done", 4))
self.assertEqual([(row["title"], row["changed"]) for row in center["history"]],
[("清理关注抓取的旧快照", 4)])
# 删除这一轮在活动页登记：计数、结算与摘要都在任务中心。
run = runs.query(task_key="follow-evidence-reclaim")[0]
self.assertEqual((run.status, run.progress_current, run.progress_total, run.result_summary),
("succeeded", 4, 4, {"files_removed": 4, "bytes_reclaimed": 204}))
def test_stock_cleanup_pauses_and_cancels_from_the_task_center(self):
"""活动页的暂停停在两份快照之间，继续前不再删；取消结束这一轮，删了多少照实记，剩下的重新等确认。"""
import threading
import time
from unittest.mock import patch
from peach import follow_evidence, task_control
from peach.follow_evidence import Reclaim
from peach.repository import LedgerDatabase
from peach.task_runs import TaskRunStore
root = self.root / "sources" / "follow" / "rule34video"
first = [_old_snapshot(root / "a", index) for index in range(5)]
second = [_old_snapshot(root / "b", index) for index in range(5)]
database = LedgerDatabase(self.root / "ledger.db")
runs = TaskRunStore(database)
reclaim = Reclaim(self.root / "sources" / "follow", database, runs=runs)
reclaim.status()
reclaim.worker.join(10)
# 删完第一个目录后停在闸门前，等测试发出暂停再放行。
reached, release = threading.Event(), threading.Event()
listing = follow_evidence.directories
def gated(path):
for index, directory in enumerate(listing(path)):
if index == 1:
reached.set()
release.wait(10)
yield directory
def wait_for(predicate):
deadline = time.monotonic() + 10
while not predicate():
self.assertLess(time.monotonic(), deadline)
time.sleep(0.02)
with patch.object(follow_evidence, "directories", gated):
reclaim.start()
self.assertTrue(reached.wait(10))
run = runs.query(task_key="follow-evidence-reclaim")[0]
self.assertEqual(task_control.projection(runs, run)["actions"], ["pause", "cancel", "set_speed"])
task_control.request(runs, run.id, "pause")
release.set()
wait_for(lambda: task_control.projection(runs, runs.get(run.id))["state"] == "paused")
self.assertEqual([raw.exists() for raw in first], [False, False, True, True, True])
self.assertTrue(all(raw.exists() for raw in second), "暂停后不再删")
task_control.request(runs, run.id, "cancel")
reclaim.worker.join(10)
run = runs.get(run.id)
self.assertEqual((run.status, run.result_summary["files_removed"]), ("cancelled", 4))
self.assertTrue(all(raw.exists() for raw in second))
wait_for(lambda: reclaim.status()["phase"] == "ready")
self.assertEqual(reclaim.status()["files"], 4, "剩下的重新等确认")
def test_a_library_without_follow_snapshots_does_not_list_the_item(self):
from peach import repair_center
from peach.follow_evidence import Reclaim
reclaim = Reclaim(self.root / "sources" / "follow", None)
self.assertEqual(reclaim.status(), {"phase": "idle"})
center = repair_center.snapshot(self.connection, {}, cloud_pending=0,
follow_evidence_status=reclaim.status())
self.assertNotIn("follow-evidence", [row["id"] for row in center["items"]])
if __name__ == "__main__":
unittest.main()