tests/test_creator_identity_followup.py
Rama master · réplica de lectura del repositorio original.
"""创作者身份查找后继：资料站作品一览直接落 observed，没有页与没取到都不写，写下的能整批撤回。
全程临时账本、临时缓存；取页走真实的 `WikiSitePages`，传输换成按地址回页的替身，不联网。
"""
import contextlib
import importlib.util
import io
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
from urllib.parse import quote
from peach import creator_identity_followup as followup
from peach import entity_identity_research as research
from peach import performer_alias_followup as alias
from peach.entities import normalize_entity_name
from peach.entity_classification import write_claim
from peach.followups import Attempts, attempts_root
from peach.http import HttpResponse
from peach.repository import LedgerDatabase
from peach.scraping_access import paused_until
from peach.sources.seesaa import AV_NEME, SEESAA
from support.ledger import fresh_ledger
ROOT = Path(__file__).resolve().parents[1]
STAMP = "2026-10-09T00:00:00.000Z"
CAST = ("八ッ橋さい子", "本多由奈", "北川ゆず")
def page_url(wiki: str, name: str) -> str:
return research.WIKI_ROOTS[wiki] + "d/" + quote(name.encode("euc_jp"))
def work_table(*cast: str) -> bytes:
rows = "".join(f'<tr><td>COSH-00{index}</td><td>こすっち00{index}</td>'
f'<td><a href="{page_url("sougouwiki", name)}">{name}</a></td></tr>'
for index, name in enumerate(cast, 1))
return ('<meta charset="utf-8"><div id="page-body"><div class="user-area"><table>'
'<tr><th>NO</th><th>TITLE</th><th>ACTRESS</th></tr>' + rows + '</table></div></div>').encode()
def load_revert():
spec = importlib.util.spec_from_file_location(
"revert_auto_landing_under_test", ROOT / "scripts" / "revert_auto_landing.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class Transport:
"""按地址回页面的传输替身；没登记的回 404。"""
def __init__(self, pages: dict):
self.pages, self.calls = pages, []
def __call__(self, request, _timeout, _limit):
self.calls.append(request.url)
status, body = self.pages.get(request.url, (404, b""))
return HttpResponse(status, {}, body, request.url)
def close(self):
pass
class NoWait:
def wait(self, _url):
pass
class Case(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name).resolve()
self.db = fresh_ledger(self.root)
self.database = LedgerDatabase(self.db)
self.generated = self.root / "generated"
self.cooldown = self.root / "secrets"
self.busted = []
self.contract = SimpleNamespace(
database=self.database, candidate_root=self.generated,
follow_secrets_root=self.cooldown, cache_bust=lambda: self.busted.append(1))
self.transport = Transport({})
self.next_asset = 1
def entity(self, name: str, kind: str = "creator") -> int:
with self.database.write_transaction(notify=False) as connection:
cursor = connection.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,?,?,?,?)", (kind, name, normalize_entity_name(name), STAMP, STAMP))
return int(cursor.lastrowid)
def work(self, creator: int, filename: str) -> None:
asset_id, self.next_asset = self.next_asset, self.next_asset + 1
with self.database.write_transaction(notify=False) as connection:
connection.execute("INSERT INTO asset(id,location,path,name,medium) VALUES(?,'local',?,?,'video')",
(asset_id, f"R:\\media\\{asset_id}.mp4", filename))
connection.execute("INSERT INTO asset_entity(asset_id,entity_id,role,source)"
" VALUES(?,?,'creator','test')", (asset_id, creator))
def sites(self) -> dict:
cache = self.generated / "provider-cache" / "seesaa-pages"
return {config.name: alias.WikiSitePages(cache, self.cooldown, self.transport, config=config)
for config in (SEESAA, AV_NEME)}
def run_followup(self, entity_id: int, run_id: int = 7) -> dict:
handle = SimpleNamespace(run_id=run_id, progress=lambda **_kwargs: None)
with mock.patch.object(followup, "open_sites", side_effect=lambda _contract: self.sites()), \
mock.patch.object(alias, "_LIMITER", NoWait()):
return followup.run(self.contract, followup.followup_key(entity_id), handle)
def claims(self, entity_id: int) -> set[tuple[str, str, str]]:
with self.database.read_connection() as connection:
return {(row[0], row[1], row[2]) for row in connection.execute(
"SELECT value,status,source FROM entity_classification WHERE entity_id=? AND facet='identity'",
(entity_id,))}
def series_creator(self) -> int:
"""名字查不到、文件名前缀查得到的创作者：三部三位不同出演者。"""
for name in CAST:
self.entity(name, "performer")
creator = self.entity("Scotch Amateur")
for index, name in enumerate(CAST, 1):
self.work(creator, f"こすっち00{index} {name}")
return creator
class LandingTests(Case):
def test_a_work_list_under_the_filename_prefix_lands_as_observed_release(self):
creator = self.series_creator()
self.transport.pages[page_url("sougouwiki", "こすっち")] = (200, work_table(*CAST))
summary = self.run_followup(creator)
self.assertEqual(summary["outcome"], "是厂牌或系列")
self.assertEqual(self.claims(creator), {("release", "observed", f"{followup.SOURCE}@7")})
self.assertEqual(self.busted, [1])
self.assertIn(page_url("sougouwiki", "Scotch Amateur"), self.transport.calls)
# 有了可信身份，存量补派与再跑都不再问站。
asked = len(self.transport.calls)
self.assertEqual(self.run_followup(creator)["outcome"], "已有身份")
self.assertEqual(len(self.transport.calls), asked)
with self.database.read_connection() as connection:
self.assertEqual(followup.stock(connection, Attempts(self.root / "attempts"), limit=5), [])
def test_missing_pages_write_nothing_and_are_not_asked_again(self):
creator = self.entity("lucky")
summary = self.run_followup(creator)
self.assertEqual(summary["outcome"], "两站都没有作品一览")
self.assertIn("sougouwiki「lucky」（页不存在）", summary["asked"])
self.assertEqual(self.claims(creator), set())
asked = len(self.transport.calls)
self.assertEqual(asked, 2)
self.run_followup(creator)
self.assertEqual(len(self.transport.calls), asked)
with self.database.read_connection() as connection:
self.assertEqual(followup.stock(connection, Attempts(attempts_root(self.generated)), limit=5), [])
def test_a_refused_site_is_unfetched_cools_down_and_retries_later(self):
creator = self.entity("lucky")
self.transport.pages[page_url("sougouwiki", "lucky")] = (403, b"")
self.assertEqual(self.run_followup(creator)["outcome"], "未取得")
self.assertEqual(self.claims(creator), set())
self.assertTrue(paused_until(self.cooldown, SEESAA.name))
self.assertFalse(paused_until(self.cooldown, AV_NEME.name))
attempts = Attempts(attempts_root(self.generated), clock=lambda: 10 ** 10)
with self.database.read_connection() as connection:
self.assertEqual([item.key for item in followup.stock(connection, attempts, limit=5)],
[followup.followup_key(creator)])
def test_reviewed_identities_and_structural_names_are_never_queried(self):
reviewed = self.entity("Known Person")
with self.database.write_transaction(notify=False) as connection:
write_claim(connection, entity_id=reviewed, facet="identity", value="person",
source="user:identity-review", evidence="用户复核", status="approved")
self.entity("1080p")
fresh = self.entity("lucky")
with self.database.read_connection() as connection:
self.assertEqual([item.key for item in followup.plan(connection, since_entity_id=0)],
[followup.followup_key(fresh)])
self.assertEqual(self.run_followup(reviewed)["outcome"], "已有身份")
self.assertEqual(self.transport.calls, [])
self.assertEqual(self.run_followup(999), {"outcome": "档案已不存在"})
class RevertTests(Case):
def test_a_batch_of_claims_is_reverted_without_touching_other_sources(self):
creator = self.series_creator()
self.transport.pages[page_url("sougouwiki", "こすっち")] = (200, work_table(*CAST))
self.run_followup(creator)
with self.database.write_transaction(notify=False) as connection:
write_claim(connection, entity_id=creator, facet="identity", value="unknown",
source=research.SOURCE, evidence="公开身份来源未取得")
revert = load_revert()
base = ["--db", str(self.db), "--source", followup.SOURCE]
with contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(revert.main([*base, "--apply", "--backup", str(self.root / "backup.db")]), 0)
self.assertEqual(self.claims(creator), {("unknown", "candidate", research.SOURCE)})
class AllocationTests(Case):
def test_new_and_stock_creators_get_identity_followups(self):
from peach import library_processing, performer_profile_followup, task_runs
old = [self.entity(f"account{index:02d}") for index in range(5)]
for entity_id in old:
self.work(entity_id, f"{entity_id}.mp4")
new = self.entity("newcomer")
config = SimpleNamespace(directory=lambda _name: self.generated)
# 一轮 8 条，补别名、补资料各留 1 条，创作者身份的存量取自己那 2 条。
with mock.patch.object(task_runs, "MAX_FOLLOWUPS", 8), \
mock.patch.object(alias, "STOCK_SHARE", 1), \
mock.patch.object(performer_profile_followup, "STOCK_SHARE", 1), \
mock.patch.object(followup, "STOCK_SHARE", 2):
found = library_processing._entity_followups(self.database, config, old[-1])
keys = [item["key"] for item in found if item["task_key"] == followup.TASK_KEY]
self.assertEqual(keys[0], followup.followup_key(new))
self.assertEqual(len(keys), 3)
if __name__ == "__main__":
unittest.main()