tests/test_performer_agency_resync.py
Rama master · réplica de lectura del repositorio original.
"""重问所属事务所：选人是并集，判词按站上写的那一串，写入只碰移籍的那几行。"""
import contextlib
import io
import json
import sqlite3
import tempfile
import types
import unittest
from pathlib import Path
from peach.entities import normalize_entity_name
from peach.migrations import upgrade
from scripts.resync_performer_agency import (
GONE, MISSING, MOVED, REJECTED, SAME, apply_rows, ask, plan, search_url, targets,
)
ROOT = Path(__file__).resolve().parents[1]
MIGRATIONS = ROOT / "migrations"
STAMP = "2026-09-05T00:00:00Z"
def profile_html(shown: str) -> str:
return ('<table><tr><td><span>所属事務所</span><p>'
f'<a href="actress_list.php?production=1251">{shown}</a>'
'</p></td></tr></table>')
class Response:
def __init__(self, status: int, url: str, body: str = ""):
self.status = status
self.url = url
self.body = body.encode("utf-8")
class Http:
"""按检索地址发页的取数器替身。没登记的名字停在检索页，等同未唯一命中。"""
def __init__(self, pages: dict[str, Response]):
self.pages = pages
self.asked: list[str] = []
def __call__(self, request, timeout, limit):
self.asked.append(request.url)
return self.pages.get(
request.url, Response(200, "https://www.minnano-av.com/search_result.php"))
def close(self):
pass
class AskTests(unittest.TestCase):
"""站点那一问只有三种结果：命中、没这一格、取不到。"""
def page(self, shown: str) -> Http:
return Http({search_url("松本一香"): Response(
200, "https://www.minnano-av.com/actress21661.html", profile_html(shown))})
def test_the_shown_name_comes_back_whole(self):
"""括号里那一段是拆名字那一步要用的，采集这一步不动它。"""
shown, found, note = ask(self.page("ELTRA(エルトラ)旧・LIGHT"), "松本一香", 1.0)
self.assertEqual(shown, "ELTRA(エルトラ)旧・LIGHT")
self.assertEqual(found, "21661")
self.assertIn("actress21661", note)
def test_a_search_page_that_did_not_redirect_is_not_read(self):
"""检索页正文里那一堆女优是「相关女优」，信号只在最终地址里。"""
shown, found, note = ask(self.page("ELTRA"), "查无此人", 1.0)
self.assertEqual((shown, found), ("", ""))
self.assertIn("未取得", note)
def test_a_transport_failure_is_a_row_not_a_crash(self):
class Broken:
def __call__(self, *args):
raise TimeoutError("connect")
shown, _, note = ask(Broken(), "松本一香", 1.0, tries=1)
self.assertEqual(shown, "")
self.assertIn("未取得", note)
def test_a_blip_on_the_first_try_does_not_become_a_verdict(self):
"""一次 SSL EOF 写出来的行，和「站点查不到这个人」长得一模一样。"""
page = self.page("EST(エスト)旧・LIGHT")
class Flaky:
def __init__(self):
self.calls = 0
def __call__(self, request, timeout, limit):
self.calls += 1
if self.calls == 1:
raise ConnectionError("SSL: UNEXPECTED_EOF_WHILE_READING")
return page(request, timeout, limit)
flaky = Flaky()
shown, _, note = ask(flaky, "松本一香", 1.0, pause=0.0)
self.assertEqual(shown, "EST(エスト)旧・LIGHT")
self.assertEqual(flaky.calls, 2)
def test_a_search_page_that_never_redirects_is_not_retried(self):
"""再问一遍还是同一个答案，重试只是多敲别人的门。"""
http = self.page("EST")
ask(http, "查无此人", 1.0, pause=0.0)
self.assertEqual(len(http.asked), 1)
class LedgerFixture(unittest.TestCase):
"""一家事务所、三位成员，外加一位没有归属的人。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.db = Path(self.tmp.name).resolve() / "ledger.db"
upgrade(self.db, MIGRATIONS)
self.con = sqlite3.connect(self.db)
self.con.row_factory = sqlite3.Row
self.addCleanup(self.con.close)
held = json.dumps({"agency": {"name": "LIGHT", "source": "t", "checked_at": STAMP}},
ensure_ascii=False)
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,metadata_json,"
"created_at,updated_at) VALUES(?,?,?,?,?,'t','t')",
[(11, "performer", "松本一香", normalize_entity_name("松本一香"), held),
(12, "performer", "宫西光", normalize_entity_name("宫西光"), held),
(13, "performer", "小泉日向", normalize_entity_name("小泉日向"), held),
(14, "performer", "北野未奈", normalize_entity_name("北野未奈"), "{}"),
(30, "agency", "LIGHT", normalize_entity_name("LIGHT"), "{}")])
self.con.executemany(
"INSERT INTO entity_membership(member_id,agency_id,source,confidence,checked_at)"
" VALUES(?,30,'minnano-av:所属事務所',1.0,?)",
[(11, STAMP), (12, STAMP), (13, STAMP)])
self.con.commit()
def agency_of(self, entity_id: int) -> dict:
row = self.con.execute(
"SELECT metadata_json FROM entity WHERE id=?", (entity_id,)).fetchone()
return json.loads(row[0] or "{}").get("agency") or {}
class TargetTests(LedgerFixture):
def test_a_whole_agency_can_be_asked_again(self):
found = targets(self.con, ["LIGHT"], [])
self.assertEqual([row["performer"] for row in found],
["松本一香", "宫西光", "小泉日向"])
self.assertEqual(found[0]["current_agency"], "LIGHT")
def test_a_named_outsider_joins_the_batch(self):
"""移籍走了的人不在这一家名下，两个条件是并集才问得到她。"""
found = targets(self.con, ["LIGHT"], ["北野未奈"])
self.assertEqual(len(found), 4)
self.assertEqual(found[-1]["current_agency"], "")
def test_the_same_person_named_twice_is_asked_once(self):
self.assertEqual(len(targets(self.con, ["LIGHT"], ["松本一香"])), 3)
def test_the_japanese_alias_leads_the_search_chain(self):
"""站点只认日文写法，规范名是简体中文；不带别名去问就查不到这个人。"""
self.con.execute(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source)"
" VALUES(11,'松本いちか',?,'test')", (normalize_entity_name("松本いちか"),))
found = targets(self.con, [], ["松本一香"])
self.assertEqual(found[0]["chain"], ["松本いちか", "松本一香"])
class PlanTests(LedgerFixture):
def args(self, **overrides):
base = {"agency": ["LIGHT"], "only": [], "interval": 0.0, "timeout": 1.0}
return types.SimpleNamespace(**{**base, **overrides})
def http(self, answers: dict[str, str]) -> Http:
return Http({search_url(name): Response(
200, f"https://www.minnano-av.com/actress{2000 + index}.html",
profile_html(shown))
for index, (name, shown) in enumerate(answers.items())})
def run_plan(self, answers: dict[str, str]):
"""进度是打给人看的，测试把它接走：控制台是 GBK，中点写不进去。"""
with contextlib.redirect_stdout(io.StringIO()):
return plan(self.con, self.http(answers), self.args())
def test_each_person_lands_on_exactly_one_verdict(self):
rows = self.run_plan({"松本一香": "ELTRA(エルトラ)旧・LIGHT", "宫西光": "LIGHT",
"小泉日向": ""})
self.assertEqual({str(row["performer"]): str(row["verdict"]) for row in rows},
{"松本一香": MOVED, "宫西光": SAME, "小泉日向": GONE})
def test_the_agency_she_rejected_is_not_a_move(self):
"""站上那一格写的是她驳回过的那家（个人经纪公司），不当移籍写回元数据。"""
self.con.execute("UPDATE entity SET metadata_json=? WHERE id=11", (json.dumps(
{"agency_rejected": [{"name": "株式会社Miss"}]}, ensure_ascii=False),))
rows = self.run_plan({"松本一香": "株式会社Miss", "宫西光": "LIGHT", "小泉日向": ""})
self.assertEqual(str(rows[0]["verdict"]), REJECTED)
apply_rows(self.con, rows, "later")
self.assertEqual(self.agency_of(11), {})
def test_a_page_that_never_arrived_is_not_a_departure(self):
"""取不到和「这一格是空的」必须分得开，否则重跑一次就把人清空了。"""
rows = self.run_plan({})
self.assertEqual({str(row["verdict"]) for row in rows}, {MISSING})
def with_alias(self):
self.con.execute(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source)"
" VALUES(11,'松本いちか',?,'test')", (normalize_entity_name("松本いちか"),))
def test_the_chain_stops_at_the_first_name_that_lands(self):
"""日文名排在链首，它命中了就不该再为简体名多敲一次门。"""
self.with_alias()
http = self.http({"松本いちか": "EST(エスト)旧・LIGHT"})
with contextlib.redirect_stdout(io.StringIO()):
rows = plan(self.con, http, self.args(agency=[], only=["松本一香"]))
self.assertEqual(str(rows[0]["verdict"]), MOVED)
self.assertEqual(http.asked, [search_url("松本いちか")])
def test_a_miss_on_the_first_name_tries_the_next(self):
self.with_alias()
http = self.http({"松本一香": "EST(エスト)旧・LIGHT"})
with contextlib.redirect_stdout(io.StringIO()):
rows = plan(self.con, http, self.args(agency=[], only=["松本一香"]))
self.assertEqual(str(rows[0]["verdict"]), MOVED)
self.assertEqual(len(http.asked), 2)
class ApplyTests(LedgerFixture):
def rows(self, **overrides):
base = {"entity_id": 11, "performer": "松本一香", "current_agency": "LIGHT",
"site_agency": "ELTRA(エルトラ)旧・LIGHT", "verdict": MOVED,
"evidence": "t"}
return [{**base, **overrides}]
def test_a_move_overwrites_the_name_with_a_fresh_stamp(self):
apply_rows(self.con, self.rows(), "2026-09-06T00:00:00Z")
found = self.agency_of(11)
self.assertEqual(found["name"], "ELTRA(エルトラ)旧・LIGHT")
self.assertEqual(found["checked_at"], "2026-09-06T00:00:00Z")
def test_an_unchanged_row_is_left_alone(self):
apply_rows(self.con, self.rows(site_agency="LIGHT", verdict=SAME), "later")
self.assertEqual(self.agency_of(11)["checked_at"], STAMP)
def test_an_empty_cell_never_clears_the_ledger(self):
"""站点某天不显示这一格，和这个人真的解约了，在页面上长得一模一样。"""
apply_rows(self.con, self.rows(site_agency="", verdict=GONE), "later")
self.assertEqual(self.agency_of(11)["name"], "LIGHT")
def test_membership_is_left_to_the_installer(self):
"""归属表主键在成员一侧，移籍要走 REPLACE，那是 install_agencies 的事。"""
apply_rows(self.con, self.rows(), STAMP)
self.assertEqual(self.con.execute(
"SELECT agency_id FROM entity_membership WHERE member_id=11").fetchone()[0], 30)
def test_other_metadata_survives_the_rewrite(self):
self.con.execute(
"UPDATE entity SET metadata_json=? WHERE id=11",
(json.dumps({"agency": {"name": "LIGHT"}, "note": "留着"}, ensure_ascii=False),))
apply_rows(self.con, self.rows(), STAMP)
row = self.con.execute("SELECT metadata_json FROM entity WHERE id=11").fetchone()
self.assertEqual(json.loads(row[0])["note"], "留着")
if __name__ == "__main__":
unittest.main()