tests/test_agency_reject.py
Rama master · réplica de lectura del repositorio original.
"""驳回一家事务所：记在人身上，撤归属、删空了的实体，重跑安装与标签修复都不写回。"""
import json
import sqlite3
import tempfile
import unittest
from peach.entities import normalize_entity_name, rejected_agencies
from support.ledger import fresh_ledger
from scripts.install_agencies import collect as agency_plan
from scripts.reject_agency import apply_rows, plan
from scripts.repair_link_labels import apply_rows as repair_labels
from scripts.repair_link_labels import collect as collect_labels
STAMP = "2026-09-24T00:00:00Z"
class RejectFixture(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)
held = {"agency": {"name": "株式会社Miss", "checked_at": STAMP}}
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("三上悠亚"),
json.dumps(held, ensure_ascii=False)),
(12, "performer", "松本一香", normalize_entity_name("松本一香"), "{}"),
(13, "performer", "宫西光", normalize_entity_name("宫西光"), "{}"),
(30, "agency", "株式会社Miss", normalize_entity_name("株式会社Miss"), "{}"),
(31, "agency", "LIGHT", normalize_entity_name("LIGHT"), "{}")])
self.con.executemany(
"INSERT INTO entity_membership(member_id,agency_id,source,confidence,checked_at)"
" VALUES(?,?,'minnano-av:所属事務所',1.0,?)",
[(11, 30, STAMP), (12, 31, STAMP), (13, 31, STAMP)])
self.con.execute(
"INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,created_at,updated_at)"
" VALUES(30,'official','官方网站','https://miss.co.jp/','miss.co.jp','t','t')")
self.con.execute(
"INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id,"
"metadata_json,last_synced_at) VALUES(30,'minnano-av','production','1215','{}',?)",
(STAMP,))
self.con.commit()
def reject(self, performer, agency, reason="个人经纪公司", replacement=""):
rows = plan(self.con, [{"performer": performer, "agency": agency, "reason": reason,
"replacement": replacement}])
done = apply_rows(self.con, rows, STAMP)
self.con.commit()
return rows, done
def metadata(self, entity_id):
row = self.con.execute("SELECT metadata_json FROM entity WHERE id=?", (entity_id,))
return json.loads(row.fetchone()[0] or "{}")
def count(self, sql, *params):
return self.con.execute(sql, params).fetchone()[0]
class RejectTests(RejectFixture):
def test_her_only_company_goes_away_with_its_links_and_numbers(self):
_, done = self.reject("三上悠亚", "株式会社Miss")
self.assertEqual(done, {"驳回": 1, "归属": 1, "事务所": 1, "改挂": 0, "新建事务所": 0,
"链接标签": 0})
self.assertEqual(self.count("SELECT count(*) FROM entity WHERE id=30"), 0)
self.assertEqual(self.count("SELECT count(*) FROM entity_link WHERE entity_id=30"), 0)
self.assertEqual(
self.count("SELECT count(*) FROM entity_external_ref WHERE entity_id=30"), 0)
self.assertEqual(self.con.execute("PRAGMA foreign_key_check").fetchall(), [])
def test_the_rejection_stays_on_her_and_the_stale_evidence_goes(self):
self.reject("三上悠亚", "株式会社Miss")
found = self.metadata(11)
self.assertNotIn("agency", found)
self.assertEqual(rejected_agencies(found), {normalize_entity_name("株式会社Miss")})
self.assertEqual(found["agency_rejected"][0]["source"], "user:manual")
def test_an_agency_with_other_members_keeps_its_entity(self):
rows, done = self.reject("松本一香", "LIGHT")
self.assertEqual(done["事务所"], 0)
self.assertEqual(self.count("SELECT count(*) FROM entity_membership WHERE agency_id=31"), 1)
self.assertIn("还有 1 人", str(rows[0]["note"]))
def test_rejecting_twice_records_it_once(self):
self.reject("三上悠亚", "株式会社Miss")
_, done = self.reject("三上悠亚", "株式会社Miss")
self.assertEqual(done["驳回"], 0)
self.assertEqual(len(self.metadata(11)["agency_rejected"]), 1)
def test_an_unknown_performer_is_reported_not_written(self):
rows, done = self.reject("不存在的人", "株式会社Miss")
self.assertTrue(str(rows[0]["note"]).startswith("未取得"))
self.assertEqual(set(done.values()), {0})
def test_the_replacement_becomes_her_agency_and_survives_a_reinstall(self):
"""改挂到 AV 时期的事务所：实体、读音别名、归属、证据四样一致，重跑安装得出同一家。"""
_, done = self.reject("三上悠亚", "株式会社Miss",
replacement="ONE'S DOUBLE(ワンズダブル)")
self.assertEqual((done["改挂"], done["新建事务所"]), (1, 1))
agency_id = self.count("SELECT agency_id FROM entity_membership WHERE member_id=11")
self.assertEqual(self.count("SELECT canonical_name FROM entity WHERE id=?", agency_id),
"ONE'S DOUBLE")
self.assertEqual(self.count("SELECT alias FROM entity_alias WHERE entity_id=?", agency_id),
"ワンズダブル")
self.assertEqual(self.metadata(11)["agency"]["name"], "ONE'S DOUBLE(ワンズダブル)")
self.assertEqual([item["agency"] for item in agency_plan(self.con)], ["ONE'S DOUBLE"])
def add_link(self, label, url, hostname):
self.con.execute(
"INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,created_at,updated_at)"
" VALUES(11,'official',?,?,?,'t','t')", (label, url, hostname))
self.con.commit()
return self.count("SELECT max(id) FROM entity_link")
def label(self, link_id):
return self.count("SELECT label FROM entity_link WHERE id=?", link_id)
def test_her_links_stop_speaking_for_the_rejected_agency(self):
"""官网链接上贴着的那家被驳回：有 AV 时期事务所写它，别的标签不动。"""
wrong = self.add_link("株式会社Miss", "https://ones-double.com/people/mikami-yua/",
"ones-double.com")
other = self.add_link("FALENO", "https://faleno.jp/top/actress/", "faleno.jp")
_, done = self.reject("三上悠亚", "株式会社Miss", replacement="ONE'S DOUBLE(ワンズダブル)")
self.assertEqual(done["链接标签"], 1)
self.assertEqual((self.label(wrong), self.label(other)), ("ONE'S DOUBLE", "FALENO"))
def test_without_a_replacement_the_label_falls_back_to_the_generic_one(self):
wrong = self.add_link("株式会社Miss", "https://ones-double.com/people/mikami-yua/",
"ones-double.com")
self.reject("三上悠亚", "株式会社Miss")
self.assertEqual(self.label(wrong), "官方网站")
class WriteBackTests(RejectFixture):
"""驳回之后，站上那一格原样再来一遍：两条写回路径都要跳过。"""
def test_the_installer_does_not_rebuild_a_rejected_agency(self):
self.reject("三上悠亚", "株式会社Miss")
found = self.metadata(11)
found["agency"] = {"name": "株式会社Miss", "checked_at": STAMP}
self.con.execute("UPDATE entity SET metadata_json=? WHERE id=11",
(json.dumps(found, ensure_ascii=False),))
self.assertEqual([item["agency"] for item in agency_plan(self.con)], [])
def test_relabelling_links_does_not_write_the_rejected_agency_back(self):
self.reject("三上悠亚", "株式会社Miss")
repair_labels(self.con, [{"entity_id": 11, "agency": "株式会社Miss", "verdict": "keep",
"link_id": 0, "link_kind": "official",
"new_label": "", "new_link_kind": "official"}], STAMP)
self.assertNotIn("agency", self.metadata(11))
def test_the_label_repair_rewrites_a_label_naming_a_rejected_agency(self):
"""驳回早于链接采集时，标签修复自己认出来：写着已驳回事务所的标签判 auto。"""
self.reject("三上悠亚", "株式会社Miss")
self.con.execute(
"INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,created_at,updated_at)"
" VALUES(11,'official','株式会社Miss','https://ones-double.com/people/mikami-yua/',"
"'ones-double.com','t','t')")
found = [row for row in collect_labels(self.con) if row["entity_id"] == 11]
self.assertEqual([(row["verdict"], row["new_label"], row["agency"]) for row in found],
[("auto", "官方网站", "")])
if __name__ == "__main__":
unittest.main()