"""把一格装了多个艺名的别名拆成一人多名，以及不该被拆的四类括号。""" import sqlite3 import tempfile import unittest from pathlib import Path from peach.entities import split_composite_person_name from peach.metadata import normalized_performers, split_former_name from peach.metadata_auto_apply import _apply_performer_candidate from peach.migrations import upgrade from scripts.split_composite_aliases import ( CLEANUP_SOURCE, apply_review, apply_rows, collect, ) ROOT = Path(__file__).resolve().parents[1] class CompositePersonNameTests(unittest.TestCase): def test_a_packed_romaji_field_becomes_one_name_per_person(self): self.assertEqual( split_composite_person_name("Ako Momona (Kou Akemi, Mari Koizumi)"), ["Ako Momona", "Kou Akemi", "Mari Koizumi"]) self.assertEqual(split_composite_person_name("Nao (Hitomi Koike)"), ["Nao", "Hitomi Koike"]) def test_a_bracket_repeating_the_name_collapses_to_one(self): # 上游把括号无条件填满，`Rei Mizuna (Rei Mizuna)` 是模板产物不是两个人。 self.assertEqual(split_composite_person_name("Rei Mizuna (Rei Mizuna)"), ["Rei Mizuna"]) def test_names_carrying_dots_apostrophes_and_hyphens_still_split(self): self.assertEqual(split_composite_person_name("A.J. O'Neil (Mary-Ann Doe)"), ["A.J. O'Neil", "Mary-Ann Doe"]) def test_the_other_four_kinds_of_bracket_are_left_whole(self): # 括号里是厂牌消歧、角色出处、接稿状态、去重后缀——都不是这个人的另一个名字。 for name in ("AV DEBUT（本物人妻）", "アスナ(SAO)", "快慢扳机（接稿中）", "kitty(1)", "Mana(23)", "プレステージプレミアム(PRESTIGEPREMIUM)", "Egami(えがみ)", "Z(指定)ERO"): self.assertEqual(split_composite_person_name(name), [name]) def test_a_plain_name_comes_back_as_itself(self): self.assertEqual(split_composite_person_name(" 白石亚子 "), ["白石亚子"]) self.assertEqual(split_composite_person_name(""), []) class CompositeAliasSplitTests(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, ROOT / "migrations") self.con = sqlite3.connect(self.db) self.addCleanup(self.con.close) self.con.executemany( "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES(?,?,?,?,'t','t')", [(10, "performer", "藤木美夏", "藤木美夏"), (11, "performer", "里中静香", "里中静香"), (12, "series", "AV DEBUT（本物人妻）", "av debut（本物人妻）"), (13, "tag", "アスナ(SAO)", "アスナ(sao)"), (14, "performer", "Kou Akemi", "kou akemi"), (15, "performer", "新井爱丽", "新井爱丽")]) self.con.executemany( "INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)" " VALUES(?,?,?,?,1.0)", [(10, "Ako Momona (Kou Akemi, Mari Koizumi)", "ako momona (kou akemi, mari koizumi)", "r18:performer"), (11, "Rei Mizuna (Rei Mizuna)", "rei mizuna (rei mizuna)", "r18:performer"), (12, "AV DEBUT (Real Married Woman)", "av debut (real married woman)", "r18dev:series-localization"), (15, "Elly Akira (Elly Arai)", "elly akira (elly arai)", "merge:duplicate-identity")]) self.con.commit() def _rows(self): return {str(row["value"]): row for row in collect(self.con)} def test_only_the_romaji_person_shape_is_marked_auto(self): rows = self._rows() self.assertEqual(rows["Elly Akira (Elly Arai)"]["verdict"], "auto") self.assertEqual(rows["Rei Mizuna (Rei Mizuna)"]["verdict"], "auto") # 系列别名的括号里是厂牌，标签的括号里是作品出处。 self.assertEqual(rows["AV DEBUT (Real Married Woman)"]["verdict"], "review") self.assertEqual(rows["AV DEBUT（本物人妻）"]["verdict"], "review") self.assertEqual(rows["アスナ(SAO)"]["verdict"], "review") def test_a_part_that_is_another_entitys_canonical_name_goes_to_review(self): # `Kou Akemi` 已经自己占了一条实体：那意味着可能要合并，不是拆别名顺手能定的。 row = self._rows()["Ako Momona (Kou Akemi, Mari Koizumi)"] self.assertEqual(row["verdict"], "review") self.assertIn("撞 entity 14", str(row["note"])) def test_applying_replaces_the_packed_row_with_one_row_per_name(self): moved = apply_rows(self.con, collect(self.con)) self.assertEqual(moved, {"added": 3, "removed": 2}) self.assertEqual( self.con.execute("SELECT alias,source FROM entity_alias WHERE entity_id=15" " ORDER BY alias").fetchall(), [("Elly Akira", "merge:duplicate-identity"), ("Elly Arai", "merge:duplicate-identity")]) # 每一段都记在提供它的那个来源名下：这几个罗马字确实是它给的，只是被打包在一个字段里。 self.assertEqual( self.con.execute("SELECT alias,source FROM entity_alias WHERE entity_id=11") .fetchall(), [("Rei Mizuna", "r18:performer")]) def test_applying_leaves_every_review_row_untouched(self): apply_rows(self.con, collect(self.con)) for entity_id, alias in ((10, "Ako Momona (Kou Akemi, Mari Koizumi)"), (12, "AV DEBUT (Real Married Woman)")): self.assertEqual( self.con.execute("SELECT count(*) FROM entity_alias" " WHERE entity_id=? AND alias=?", (entity_id, alias)).fetchone()[0], 1) self.assertEqual( self.con.execute("SELECT canonical_name FROM entity WHERE id=13") .fetchone()[0], "アスナ(SAO)") class ReviewedNameCleanupTests(unittest.TestCase): """形状分不开、只有人能判的那些：复核 CSV 填 `verdict` 与 `target` 再执行。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.db = Path(self.tmp.name).resolve() / "ledger.db" upgrade(self.db, ROOT / "migrations") self.con = sqlite3.connect(self.db) self.addCleanup(self.con.close) self.con.executemany( "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES(?,?,?,?,'t','t')", [(20, "creator", "幼月月@一日目(土)東オ54a", "幼月月@一日目(土)東オ54a"), (21, "studio", "プレステージプレミアム", "プレステージプレミアム"), (22, "performer", "白石亚子", "白石亚子"), (23, "creator", "みどりふう", "みどりふう")]) self.con.execute( "INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)" " VALUES(21,'プレステージプレミアム(PRESTIGEPREMIUM)'," "'プレステージプレミアム(prestigepremium)','javdb:studio',1.0)") self.con.executemany( "INSERT INTO asset(id,location,path,name,medium,creator)" " VALUES(?,'online',?,?,'image',?)", [(1, "pixiv/1", "a", "幼月月@一日目(土)東オ54a"), (2, "pixiv/2", "b", "幼月月@一日目(土)東オ54a"), (3, "pixiv/3", "c", "みどりふう")]) self.con.execute( "INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)" " VALUES(1,20,'creator','pixiv',1.0)") self.con.commit() def _row(self, verdict, entity_id, value, target, kind="creator", source=""): return {"verdict": verdict, "kind": kind, "entity_id": str(entity_id), "value": value, "target": target, "source": source} def test_stripping_a_booth_suffix_renames_the_entity_and_its_assets(self): done = apply_review(self.con, [ self._row("strip", 20, "幼月月@一日目(土)東オ54a", "幼月月")]) self.assertEqual((done["strip"], done["flat"]), (1, 2)) self.assertEqual( self.con.execute("SELECT canonical_name,normalized_name FROM entity" " WHERE id=20").fetchone(), ("幼月月", "幼月月")) # 扁平投影跟着改，否则卡片还写着旧名、按旧名照样搜得到。 self.assertEqual( sorted(name for (name,) in self.con.execute( "SELECT creator FROM asset WHERE creator IS NOT NULL")), ["みどりふう", "幼月月", "幼月月"]) def test_the_stripped_name_stays_as_an_alias_under_the_cleanup_source(self): apply_review(self.con, [self._row("strip", 20, "幼月月@一日目(土)東オ54a", "幼月月")]) self.assertEqual( self.con.execute("SELECT alias,source FROM entity_alias WHERE entity_id=20") .fetchall(), [("幼月月@一日目(土)東オ54a", CLEANUP_SOURCE)]) def test_splitting_a_studio_alias_keeps_the_source_that_gave_it(self): done = apply_review(self.con, [ self._row("split", 21, "プレステージプレミアム(PRESTIGEPREMIUM)", "プレステージプレミアム | PRESTIGEPREMIUM", kind="studio", source="javdb:studio")]) self.assertEqual(done["split"], 1) self.assertEqual( sorted(alias for (alias,) in self.con.execute( "SELECT alias FROM entity_alias WHERE entity_id=21")), ["PRESTIGEPREMIUM", "プレステージプレミアム"]) def test_a_target_taken_by_another_entity_is_refused_not_written(self): # 撞上另一条实体的规范名意味着两条可能该合并，那是合并要回答的问题。 done = apply_review(self.con, [ self._row("strip", 20, "幼月月@一日目(土)東オ54a", "みどりふう")]) self.assertEqual(done["strip"], 0) self.assertEqual(len(done["blocked"]), 1) self.assertIn("撞 entity 23", done["blocked"][0]) self.assertEqual( self.con.execute("SELECT canonical_name FROM entity WHERE id=20").fetchone()[0], "幼月月@一日目(土)東オ54a") def test_rows_left_at_review_do_nothing(self): done = apply_review(self.con, [ self._row("review", 20, "幼月月@一日目(土)東オ54a", "幼月月"), self._row("strip", 20, "幼月月@一日目(土)東オ54a", "")]) self.assertEqual((done["strip"], done["split"], done["flat"]), (0, 0, 0)) self.assertEqual( self.con.execute("SELECT canonical_name FROM entity WHERE id=20").fetchone()[0], "幼月月@一日目(土)東オ54a") def test_a_performer_strip_rewrites_the_flat_tag(self): self.con.execute( "INSERT INTO asset(id,location,path,name,medium)" " VALUES(4,'online','pixiv/4','d','image')") self.con.execute( "INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)" " VALUES(4,22,'performer','javdb',1.0)") self.con.execute( "INSERT INTO asset_tag(asset_id,tag,confidence,source)" " VALUES(4,'演员:白石亚子',1.0,'javdb')") done = apply_review(self.con, [ self._row("strip", 22, "白石亚子", "白石 あこ", kind="performer")]) self.assertEqual((done["strip"], done["flat"]), (1, 1)) self.assertEqual( [tag for (tag,) in self.con.execute( "SELECT tag FROM asset_tag WHERE asset_id=4")], ["演员:白石 あこ"]) class FormerNameTests(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, ROOT / "migrations") 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(?,'performer',?,?,'t','t')", [(20, "美谷朱音", "美谷朱音"), (21, "姬川优奈", "姬川优奈")]) self.con.execute( "INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)" " VALUES(21,'月城らん','月城らん','test',1.0)") self.con.execute( "INSERT INTO asset(id,location,path,name,medium,code,size)" " VALUES(1,'R:','R:\\media\\MIZD-997.mp4','MIZD-997.mp4','video','MIZD-997',1)") def land(self, raw_name: str, external_id: str = "") -> int: performers, _warnings = normalized_performers( [{"japanese_name": raw_name, "dmm_id": external_id, "profile_source": "r18dev"}]) _apply_performer_candidate(self.con, [1], {"value": performers}, source="r18dev", confidence=0.9, metadata={}, now="t") return self.con.execute( "SELECT entity_id FROM asset_entity WHERE asset_id=1 AND role='performer'" ).fetchone()[0] def test_current_and_former_names_split_apart(self): self.assertEqual(split_former_name("美谷朱音（美谷朱里）"), ("美谷朱音", "美谷朱里")) self.assertEqual(split_former_name("七海ひな(旧・七瀬ひな)"), ("七海ひな", "七瀬ひな")) performers, _warnings = normalized_performers([{"japanese_name": "美谷朱音（美谷朱里）"}]) self.assertEqual((performers[0]["name"], performers[0]["aliases"], performers[0]["former_names"]), ("美谷朱音", ["美谷朱里"], ["美谷朱里"])) def test_brackets_that_are_not_a_former_name_stay_whole(self): # 年龄、描述、罗马字、几个旧名挤在一格：都不是「一个现名配一个旧名」。 for name in ("Mana(23)", "みく（素人）", "Rei Mizuna (Rei Mizuna)", "美谷朱音（美谷朱里、朱里）", "ゆい（20歳）"): self.assertEqual(split_former_name(name), (name, "")) def test_the_current_name_lands_on_the_entity_already_there(self): self.assertEqual(self.land("美谷朱音（美谷朱里）", "1039982"), 20) self.assertEqual(self.con.execute( "SELECT count(*) FROM entity WHERE kind='performer'").fetchone()[0], 2) self.assertEqual([row[0] for row in self.con.execute( "SELECT tag FROM asset_tag WHERE asset_id=1")], ["演员:美谷朱音"]) def test_a_new_current_name_lands_on_whoever_holds_the_former_one(self): self.assertEqual(self.land("姫川ゆうな（月城らん）"), 21) self.assertEqual(self.con.execute( "SELECT count(*) FROM entity WHERE kind='performer'").fetchone()[0], 2) self.assertEqual(self.con.execute( "SELECT source FROM entity_alias WHERE entity_id=21 AND alias='姫川ゆうな'" ).fetchone()[0], "javinizer:current-name") if __name__ == "__main__": unittest.main()