tests/test_agency_roster_harvest.py
Rama master · réplica de lectura del repositorio original.
"""事务所名册采集：编号要核对到家，名册要翻到底，归属只补空着的那一位。"""
import json
import sqlite3
import tempfile
import types
import unittest
from peach.entities import normalize_entity_name
from support.ledger import fresh_ledger
from scripts.harvest_agency_rosters import (
ABSENT, AMBIGUOUS, INSTALLABLE, KNOWN, MISSING, MOVED, NEW, PRODUCTION, PROVIDER, REJECTED,
apply_rows, fetch_roster, find_production, plan, roster_key, roster_url, search_url,
shown_keys,
)
STAMP = "2026-09-05T00:00:00Z"
def roster_html(names: list[tuple[str, str]], total: int, following: str = "") -> str:
"""按名册页的形状造一页：人在 JSON-LD 里，下一页在 `<link rel="next">` 里。"""
listing = {"@type": "CollectionPage", "mainEntity": {
"@type": "ItemList", "numberOfItems": total,
"itemListElement": [
{"item": {"name": name,
"url": f"https://www.minnano-av.com/actress{found}.html"}}
for name, found in names]}}
head = f'<link rel="next" href="{following}">' if following else ""
return ('<html><head>' + head
+ '<script type="application/ld+json">'
+ json.dumps(listing, ensure_ascii=False)
+ '</script></head><body></body></html>')
def profile_html(shown: str, production: str) -> str:
return ('<table><tr><td><span>所属事務所</span><p>'
f'<a href="actress_list.php?production={production}">{shown}</a>'
'</p></td></tr></table>')
class Site:
"""按地址发页的取数器替身；没登记的地址返回空正文，等同取不回来。"""
def __init__(self, pages: dict[str, tuple[str, str]]):
self.pages = pages
self.asked: list[str] = []
def __call__(self, url: str) -> tuple[str, str]:
self.asked.append(url)
return self.pages.get(url, (url, ""))
class ShownNameTests(unittest.TestCase):
"""站上那一格写的名字，哪几段可以拿去和账本对。"""
def test_the_reading_in_brackets_is_matched_as_an_alias(self):
"""账本里是 `KRONE` 加别名 `クローネ`，站上写作 `KRONE(クローネ)`。
整串拿去对，一个都对不上；拆开才落在这家名下，名册也才走得下去。
"""
self.assertEqual(shown_keys("KRONE(クローネ)"),
{roster_key("KRONE"), roster_key("クローネ")})
def test_a_former_name_in_brackets_does_not_claim_the_agency_that_still_exists(self):
"""`GG(旧・Prime Agency)` 里的旧名不作数。
认下它，GG 的名册就整份挂到 Prime Agency 头上，而 Prime Agency 在账本里
仍是独立一家。
"""
keys = shown_keys("GG(旧・Prime Agency)")
self.assertEqual(keys, {roster_key("GG")})
def test_a_former_name_after_the_bracket_is_dropped_too(self):
keys = shown_keys("SO MODEL AGENT(ソウ モデルエージェント)旧・Eightman Production")
self.assertIn(roster_key("SO MODEL AGENT"), keys)
self.assertNotIn(roster_key("Eightman Production"), keys)
def test_typographic_noise_never_decides_a_match(self):
"""站上写 `明日花 キララ`，账本别名写 `明日花キララ`：空格不是名字的一部分。"""
self.assertEqual(shown_keys("明日花 キララ"), shown_keys("明日花キララ"))
class ProductionLookupTests(unittest.TestCase):
"""编号只能从人身上取，取到还要核对到家。"""
def setUp(self):
self.agency = {"id": 30, "name": "KRONE", "aliases": ["クローネ"], "members": [11]}
self.names = {11: "凉森玲梦"}
self.aliases = {11: ["涼森れむ"]}
def probe_site(self, shown: str, production: str) -> Site:
return Site({search_url("涼森れむ"):
("https://www.minnano-av.com/actress231277.html",
profile_html(shown, production))})
def test_the_number_is_taken_from_a_known_member_and_checked_against_the_name(self):
found, note = find_production(self.probe_site("KRONE(クローネ)", "573"),
self.agency, self.names, self.aliases.get, 3)
self.assertEqual(found, "573")
self.assertIn("涼森れむ", note)
def test_a_member_pointing_at_another_agency_is_refused(self):
"""采错的成员会把别家的编号装到这家头上，而复核件上看不出来。"""
found, note = find_production(self.probe_site("T-POWERS", "110"),
self.agency, self.names, self.aliases.get, 3)
self.assertEqual(found, "")
self.assertIn("不是这家", note)
def test_an_agency_without_known_members_says_so(self):
found, note = find_production(Site({}), {**self.agency, "members": []},
self.names, self.aliases.get, 3)
self.assertEqual(found, "")
self.assertIn("没有已知成员", note)
class RosterWalkTests(unittest.TestCase):
"""站点在最后一页之后照旧给 `rel=next`，超范围的页原样重复最后一页。"""
def pages(self, mapping: dict[str, str]) -> Site:
return Site({url: (url, html) for url, html in mapping.items()})
def test_the_walk_ends_when_a_page_brings_nobody_new(self):
first, second = roster_url("573"), roster_url("573", 2)
site = self.pages({
first: roster_html([("篠田ゆう", "1"), ("鈴北梨乃", "2")], 2, second),
second: roster_html([("篠田ゆう", "1"), ("鈴北梨乃", "2")], 2,
roster_url("573", 3)),
})
people, total, note = fetch_roster(site, "573", 20)
self.assertEqual([name for name, _ in people], ["篠田ゆう", "鈴北梨乃"])
self.assertEqual(total, 2)
self.assertEqual(len(site.asked), 2)
self.assertNotIn("没采全", note)
def test_hitting_the_page_cap_is_reported_as_our_own_shortfall(self):
"""两种差额要分得开：这一种再翻还有人，另一种是站点自己的名册有重号。"""
first, second = roster_url("110"), roster_url("110", 2)
site = self.pages({
first: roster_html([("A", "1")], 100, second),
second: roster_html([("B", "2")], 100, roster_url("110", 3)),
})
_, _, note = fetch_roster(site, "110", 2)
self.assertIn("页数封顶没采全", note)
def test_a_page_that_cannot_be_fetched_stops_the_walk_and_says_which(self):
_, _, note = fetch_roster(Site({}), "573", 20)
self.assertIn("第 1 页取不回来", note)
class LedgerFixture(unittest.TestCase):
"""按真实形状建一份小账本：两家事务所、五位女优，一份五个人的名册。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.db = fresh_ledger(self.tmp.name)
self.con = sqlite3.connect(self.db)
self.con.row_factory = sqlite3.Row
self.addCleanup(self.con.close)
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,?,?,?,'t','t')",
[(11, "performer", "凉森玲梦", normalize_entity_name("凉森玲梦")),
(12, "performer", "筱田优", normalize_entity_name("筱田优")),
(13, "performer", "神雪", normalize_entity_name("神雪")),
(14, "performer", "明日花绮罗罗", normalize_entity_name("明日花绮罗罗")),
(15, "performer", "明日花キララ", normalize_entity_name("明日花キララ")),
(30, "agency", "KRONE", normalize_entity_name("KRONE")),
(31, "agency", "T-POWERS", normalize_entity_name("T-POWERS"))])
self.con.executemany(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source)"
" VALUES(?,?,?,'test')",
[(11, "涼森れむ", normalize_entity_name("涼森れむ")),
(12, "篠田ゆう", normalize_entity_name("篠田ゆう")),
(13, "神ユキ", normalize_entity_name("神ユキ")),
(14, "明日花キララ", normalize_entity_name("明日花キララ")),
(30, "クローネ", normalize_entity_name("クローネ"))])
self.con.executemany(
"INSERT INTO entity_membership(member_id,agency_id,source,confidence,checked_at)"
" VALUES(?,?,'minnano-av:所属事務所',1.0,?)",
[(11, 30, STAMP), (13, 31, STAMP)])
self.con.execute(
"INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id,"
"metadata_json,last_synced_at) VALUES(?,?,?,'573','{}',?)",
(30, PROVIDER, PRODUCTION, STAMP))
self.con.commit()
self.site = Site({roster_url("573"): (roster_url("573"), roster_html(
[("涼森れむ", "231277"), ("篠田ゆう", "3145"), ("神ユキ", "9001"),
("明日花キララ", "3146"), ("誰も知らない", "9002")], 5))})
def args(self, **overrides):
base = {"only": ["KRONE"], "probe": 3, "max_pages": 20, "all_names": False}
return types.SimpleNamespace(**{**base, **overrides})
def verdicts(self, rows) -> dict[str, str]:
return {str(row["roster_name"]): str(row["verdict"]) for row in rows}
class RosterPlanTests(LedgerFixture):
"""名册上的每个名字落到哪个判词。"""
def test_the_known_number_spares_the_lookup_entirely(self):
"""编号记在账本里的那一家，不该再为它去搜一位成员。"""
plan(self.site, self.con, self.args())
self.assertEqual(self.site.asked, [roster_url("573")])
def test_each_roster_name_lands_on_exactly_one_verdict(self):
rows = plan(self.site, self.con, self.args())
found = self.verdicts(rows)
self.assertEqual(found["涼森れむ"], KNOWN)
self.assertEqual(found["篠田ゆう"], NEW)
self.assertEqual(found["神ユキ"], MOVED)
self.assertEqual(found["明日花キララ"], AMBIGUOUS)
def test_a_rejected_agency_is_not_offered_back_to_her(self):
"""她驳回过这一家，名册再收她也只记一行，`--apply` 不装。"""
self.con.execute("UPDATE entity SET metadata_json=? WHERE id=12", (json.dumps(
{"agency_rejected": [{"name": "KRONE(クローネ)"}]}, ensure_ascii=False),))
found = self.verdicts(plan(self.site, self.con, self.args()))
self.assertEqual(found["篠田ゆう"], REJECTED)
self.assertNotIn(REJECTED, INSTALLABLE)
def test_people_the_ledger_does_not_have_collapse_into_one_counted_row(self):
"""名册的绝大多数是库里没有的人，而这些行没有任何可执行的下一步。"""
rows = [row for row in plan(self.site, self.con, self.args())
if row["verdict"] == ABSENT]
self.assertEqual(len(rows), 1)
self.assertIn("誰も知らない", str(rows[0]["evidence"]))
self.assertEqual(rows[0]["roster_name"], "")
def test_all_names_writes_them_out_one_by_one(self):
rows = plan(self.site, self.con, self.args(all_names=True))
self.assertEqual(self.verdicts(rows)["誰も知らない"], ABSENT)
def test_an_agency_whose_number_cannot_be_obtained_is_reported_as_missing(self):
"""一家取不到编号不能让整轮空转，它自己占一行「未取得」。"""
rows = plan(Site({}), self.con, self.args(only=["T-POWERS"]))
self.assertEqual([row["verdict"] for row in rows], [MISSING])
def test_only_the_named_agencies_are_walked(self):
rows = plan(self.site, self.con, self.args())
self.assertEqual({row["agency"] for row in rows}, {"KRONE"})
class ApplyTests(LedgerFixture):
"""写入只补空着的那一位，别的判词一律不动账本。"""
def test_a_new_membership_is_written_and_a_conflicting_one_is_left_alone(self):
rows = plan(self.site, self.con, self.args())
done = apply_rows(self.con, rows, STAMP)
self.con.commit()
held = dict(self.con.execute(
"SELECT member_id, agency_id FROM entity_membership").fetchall())
self.assertEqual(done["归属"], 1)
self.assertEqual(held[12], 30)
# 神雪 在账本里属于 T-POWERS，名册说她属于 KRONE：留给人判，不改。
self.assertEqual(held[13], 31)
self.assertEqual(self.con.execute(
"SELECT source FROM entity_membership WHERE member_id=12").fetchone()[0],
"minnano-av:事務所名簿")
def test_an_ambiguous_name_is_never_installed(self):
rows = plan(self.site, self.con, self.args())
apply_rows(self.con, rows, STAMP)
self.con.commit()
self.assertEqual(self.con.execute(
"SELECT count(*) FROM entity_membership WHERE member_id IN (14,15)"
).fetchone()[0], 0)
def test_the_number_is_kept_so_the_next_run_need_not_ask_again(self):
rows = plan(self.site, self.con, self.args())
apply_rows(self.con, rows, STAMP)
self.con.commit()
self.assertEqual(self.con.execute(
"SELECT external_id FROM entity_external_ref WHERE entity_id=30"
" AND provider=? AND external_kind=?", (PROVIDER, PRODUCTION)).fetchone()[0],
"573")
if __name__ == "__main__":
unittest.main()