tests/test_duplicate_identity_merge.py
Rama master · réplica de lectura del repositorio original.
"""creator / performer 跨类重复身份审计与兼容投影回归。"""
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from support.ledger import fresh_ledger
from peach.review_csv import read_rows
from scripts.merge_duplicate_identities import (
apply_repeated_projections,
apply_rows,
build_parser,
collect,
collect_repeated_projections,
named_pairs,
run,
)
class DuplicateIdentityMergeTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db = fresh_ledger(self.tmp.name)
self.con = sqlite3.connect(self.db)
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,creator) "
"VALUES(?,'local',?,?,'video',?)",
[(1, "/x/1.mp4", "1.mp4", "小千 Qian0791"),
(2, "/x/2.mp4", "2.mp4", "小千 Qian0791"),
(3, "/x/3.mp4", "3.mp4", "NOZOMI NOZOMI"),
(4, "/x/4.mp4", "4.mp4", "NOZOMI NOZOMI")])
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at) "
"VALUES(?,?,?,?, 't','t')",
[(10, "creator", "小千 Qian0791", "小千 qian0791"),
(11, "performer", "小千", "小千"),
(12, "creator", "NOZOMI NOZOMI", "nozomi nozomi"),
(13, "performer", "NOZOMI", "nozomi"),
(14, "creator", "无关目录", "无关目录")])
self.con.executemany(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence) "
"VALUES(?,?,?,?,1.0)",
[(11, "Qian0791", "qian0791", "stash:performer"),
(13, "NOZOMI NOZOMI", "nozomi nozomi", "javbus:performer")])
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) "
"VALUES(?,?,?,?,1.0)",
[(1, 10, "creator", "legacy:asset"), (2, 10, "creator", "legacy:asset"),
(1, 11, "performer", "performer"), (2, 11, "performer", "performer"),
(3, 12, "creator", "legacy:asset"),
(3, 13, "performer", "javbus:performer"),
(4, 13, "performer", "javbus:performer"),
(1, 14, "creator", "legacy:asset"), (2, 14, "creator", "legacy:asset")])
self.con.executemany(
"INSERT INTO asset_tag(asset_id,tag,confidence,source) VALUES(?,?,1.0,?)",
[(1, "演员:小千", "performer"), (2, "演员:小千", "performer"),
(3, "演员:NOZOMI NOZOMI", "javbus:performer")])
self.con.commit()
def tearDown(self):
self.con.close()
self.tmp.cleanup()
def test_collect_uses_same_assets_aliases_and_release_provenance(self):
rows = collect(self.con)
self.assertEqual(len(rows), 2)
by_drop = {row["drop_name"]: row for row in rows}
self.assertEqual(by_drop["小千"]["keep_kind"], "creator")
self.assertIn("账号别名", by_drop["小千"]["match_evidence"])
self.assertEqual(by_drop["NOZOMI NOZOMI"]["keep_kind"], "performer")
self.assertEqual(by_drop["NOZOMI NOZOMI"]["evidence"], "发行元数据")
self.assertNotIn("无关目录", {row["keep_name"] for row in rows})
def test_apply_syncs_flat_creator_and_actor_tag_projections(self):
counts = apply_rows(self.con, collect(self.con))
self.con.commit()
self.assertIsNone(self.con.execute(
"SELECT 1 FROM entity WHERE id=11").fetchone())
self.assertIsNone(self.con.execute(
"SELECT 1 FROM entity WHERE id=12").fetchone())
self.assertEqual(self.con.execute(
"SELECT creator FROM asset WHERE id=3").fetchone()[0], None)
self.assertEqual(self.con.execute(
"SELECT creator FROM asset WHERE id=4").fetchone()[0], None,
"兼容字段要按坏名字全量清，不能只清旧 creator 关系覆盖的资产")
tags = dict(self.con.execute(
"SELECT asset_id,tag FROM asset_tag ORDER BY asset_id"))
self.assertNotIn(1, tags)
self.assertNotIn(2, tags)
self.assertEqual(tags[3], "演员:NOZOMI")
self.assertEqual(counts["actor_tags_removed"], 2)
self.assertEqual(counts["actor_tags_rewritten"], 1)
self.assertEqual(self.con.execute(
"PRAGMA foreign_key_check").fetchall(), [])
def _seed_same_kind_projection(self):
"""`小桃` / `小桃 shixiaotaone` 这类同 kind 目录名投影。
20 号是规范实体：带两个账号别名，作品集合是 {20,21,22}。21 号是
`R:\\Media\\<本名 + 账号名>` 投影出来的第二条 creator，作品 {20,21} 全落在 20 号里。
22 号名字同样由 20 号的本名和别名拼成，但作品 {22,23} 越出了 20 号的集合。
"""
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,creator) "
"VALUES(?,'local',?,?,'video',?)",
[(20, "/x/20.mp4", "20.mp4", "小桃 shixiaotaone"),
(21, "/x/21.mp4", "21.mp4", "小桃 shixiaotaone"),
(22, "/x/22.mp4", "22.mp4", "小桃 taomi"),
(23, "/x/23.mp4", "23.mp4", "小桃 taomi")])
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at) "
"VALUES(?,?,?,?, 't','t')",
[(20, "creator", "小桃", "小桃"),
(21, "creator", "小桃 shixiaotaone", "小桃 shixiaotaone"),
(22, "creator", "小桃 taomi", "小桃 taomi")])
self.con.executemany(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence) "
"VALUES(?,?,?,'stash:performer',0.9)",
[(20, "shixiaotaone", "shixiaotaone"), (20, "taomi", "taomi")])
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) "
"VALUES(?,?,?,?,1.0)",
[(20, 20, "performer", "performer"), (21, 20, "performer", "performer"),
(22, 20, "performer", "performer"),
(20, 21, "creator", "legacy:asset"), (21, 21, "creator", "legacy:asset"),
(22, 22, "creator", "legacy:asset"), (23, 22, "creator", "legacy:asset")])
self.con.commit()
def test_collect_merges_same_kind_directory_projection(self):
self._seed_same_kind_projection()
rows = [row for row in collect(self.con) if row["drop_id"] == 21]
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["keep_id"], 20)
self.assertEqual(rows[0]["keep_kind"], "creator")
self.assertEqual(rows[0]["drop_kind"], "creator")
self.assertIn("同 kind", rows[0]["match_evidence"])
def test_collect_skips_same_kind_without_asset_or_provenance_evidence(self):
self._seed_same_kind_projection()
# 21 号一旦有了自己的外部引用就不再是纯目录投影，留给人工。
self.con.execute(
"INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id) "
"VALUES(21,'stash','performer','777')")
self.con.commit()
drops = {row["drop_id"] for row in collect(self.con)}
self.assertNotIn(21, drops, "带外部引用的实体不是目录投影")
self.assertNotIn(22, drops, "作品集合越出保留方时，拼名不足以判同人")
def test_apply_rewrites_flat_creator_for_same_kind_merge(self):
self._seed_same_kind_projection()
rows = [row for row in collect(self.con) if row["drop_id"] == 21]
counts = apply_rows(self.con, rows)
self.con.commit()
self.assertIsNone(self.con.execute("SELECT 1 FROM entity WHERE id=21").fetchone())
self.assertEqual([row[0] for row in self.con.execute(
"SELECT creator FROM asset WHERE id IN (20,21) ORDER BY id")], ["小桃", "小桃"])
self.assertEqual(counts["flat_rewritten"], 2)
self.assertEqual(self.con.execute(
"SELECT creator FROM asset WHERE id=22").fetchone()[0], "小桃 taomi",
"只改写被丢弃的那个名字，不动别的目录名")
self.assertIn("小桃 shixiaotaone", [row[0] for row in self.con.execute(
"SELECT alias FROM entity_alias WHERE entity_id=20")])
self.assertEqual(self.con.execute("PRAGMA foreign_key_check").fetchall(), [])
def test_projection_audit_repairs_orphan_flat_fields_tags_and_aliases(self):
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,creator) "
"VALUES(?,'local',?,?,'video',?)",
[(5, "/x/5.mp4", "5.mp4", "瀬奈まお 瀬奈まお"),
(6, "/x/6.mp4", "6.mp4", "画像を拡大する 画像を拡大する")],
)
self.con.execute(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at) "
"VALUES(15,'performer','瀬奈まお','瀬奈まお','t','t')"
)
self.con.execute(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence) "
"VALUES(15,'瀬奈まお 瀬奈まお','瀬奈まお 瀬奈まお','user:merge',1.0)"
)
self.con.execute(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) "
"VALUES(5,15,'performer','javbus:performer',1.0)"
)
self.con.executemany(
"INSERT INTO asset_tag(asset_id,tag,confidence,source) VALUES(?,?,1.0,'javbus:performer')",
[(5, "演员:瀬奈まお 瀬奈まお"),
(6, "演员:画像を拡大する 画像を拡大する")],
)
rows = collect_repeated_projections(self.con)
by_name = {row["bad_name"]: row for row in rows}
self.assertEqual(by_name["瀬奈まお 瀬奈まお"]["action"], "use-performer")
self.assertEqual(by_name["画像を拡大する 画像を拡大する"]["action"], "remove-invalid")
counts = apply_repeated_projections(self.con, rows)
self.con.commit()
self.assertEqual(self.con.execute(
"SELECT creator FROM asset WHERE id=5").fetchone()[0], None)
self.assertEqual(self.con.execute(
"SELECT creator FROM asset WHERE id=6").fetchone()[0], None)
self.assertEqual([row[0] for row in self.con.execute(
"SELECT tag FROM asset_tag WHERE asset_id=5")], ["演员:瀬奈まお"])
self.assertEqual([row[0] for row in self.con.execute(
"SELECT tag FROM asset_tag WHERE asset_id=6")], [])
self.assertEqual([row[0] for row in self.con.execute(
"SELECT alias FROM entity_alias WHERE entity_id=15")], [])
self.assertGreaterEqual(counts["flat_cleared"], 2)
class ContainedAssetSetMergeTests(unittest.TestCase):
"""跨 kind 判据放宽到「真子集」后的回归。
只认作品集合完全相同的话，`哆米`(7 部) 与目录投影 `哆米 Dolmi24`(6 部) 这种
一侧多出一两部的就永远落在判据外——多出的那几部只被 Stash 认成 performer、
没有对应本地目录，是常态而不是反证。放宽的部分由名字那一重证据兜底。
"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db = fresh_ledger(self.tmp.name)
self.con = sqlite3.connect(self.db)
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,creator) "
"VALUES(?,'local',?,?,'video',?)",
[(1, "/x/1.mp4", "1.mp4", "哆米 Dolmi24"),
(2, "/x/2.mp4", "2.mp4", "哆米 Dolmi24"),
(3, "/x/3.mp4", "3.mp4", None),
(4, "/x/4.mp4", "4.mp4", "桉X合集")])
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at) "
"VALUES(?,?,?,?, 't','t')",
[(20, "creator", "哆米 Dolmi24", "哆米 dolmi24"),
(21, "performer", "哆米", "哆米"),
(22, "creator", "桉X合集", "桉x合集"),
(23, "performer", "桉X", "桉x")])
self.con.executemany(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence) "
"VALUES(?,?,?,?,1.0)",
[(21, "Dolmi24", "dolmi24", "stash:performer"),
(23, "lananh9496", "lananh9496", "stash:performer")])
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) "
"VALUES(?,?,?,?,1.0)",
# creator 只挂 1、2；performer 多出第 3 部——真子集，不是相等。
[(1, 20, "creator", "legacy:asset"), (2, 20, "creator", "legacy:asset"),
(1, 21, "performer", "performer"), (2, 21, "performer", "performer"),
(3, 21, "performer", "performer"),
(4, 22, "creator", "legacy:asset"),
(4, 23, "performer", "performer")])
self.con.commit()
def tearDown(self):
self.con.close()
self.tmp.cleanup()
def test_collect_merges_when_creator_assets_are_a_strict_subset(self):
rows = collect(self.con)
by_drop = {row["drop_name"]: row for row in rows}
self.assertIn("哆米", by_drop)
merged = by_drop["哆米"]
# performer 侧只有 Stash 扁平断言，保留带真实目录的 creator。
self.assertEqual(merged["keep_kind"], "creator")
self.assertEqual(merged["keep_name"], "哆米 Dolmi24")
self.assertIn("账号别名", merged["match_evidence"])
self.assertIn("真包含", merged["match_evidence"])
self.assertIn("2/3 部", merged["match_evidence"])
def test_collect_still_refuses_a_name_that_is_not_alias_backed(self):
"""`桉X合集` 的多出词「合集」不是 `桉X` 的已登记别名，仍然留给人工。
目录名里的「合集」可能是同人的合集，也可能是多人合辑；判子集只说明
文件重叠，说明不了那一条目录就是这个人。
"""
self.assertNotIn(
"桉X", {row["drop_name"] for row in collect(self.con)})
class StageNameDuplicateTests(unittest.TestCase):
"""艺名已登记为别名、那个名字却还自己占一条实体。三道闸都拿实测假阳性做样本。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db = fresh_ledger(self.tmp.name)
self.con = sqlite3.connect(self.db)
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,code) "
"VALUES(?,'local',?,?,'video',?)",
[(1, "/x/1.mp4", "1.mp4", "CEAD-275"), (2, "/x/2.mp4", "2.mp4", "SKY-250"),
(3, "/x/3.mp4", "3.mp4", "MIZD-998"), (4, "/x/4.mp4", "4.mp4", "ABP-951"),
(5, "/x/5.mp4", "5.mp4", "DPMX-016"),
(6, "/x/6.mp4", "6.mp4", "300MIUM-1219")])
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,?,?,?,'t','t')",
[(20, "performer", "森泽佳奈", "森泽佳奈"),
(21, "performer", "飯岡かなこ", "飯岡かなこ"),
(22, "performer", "白石亚子", "白石亚子"),
(23, "performer", "Ako Shiraishi", "ako shiraishi"),
(24, "performer", "绫乃梓", "绫乃梓"),
(25, "performer", "NOZOMI", "nozomi")])
self.con.executemany(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)"
" VALUES(?,?,?,?,0.9)",
[(20, "飯岡かなこ", "飯岡かなこ", "r18:performer"),
(22, "Ako Shiraishi", "ako shiraishi", "r18:performer"),
(23, "Suzu Hirasawa", "suzu hirasawa", "r18:performer"),
(23, "平沢すず", "平沢すず", "avdb-actor-mapping@8e2d5b7a"),
(24, "Nozomi", "nozomi", "r18:performer")])
self.con.executemany(
"INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id)"
" VALUES(?,?,'performer',?)",
[(23, "r18", "平沢すず"), (24, "stash", "538"), (25, "stash", "610")])
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(?,?,'performer',?,0.9)",
[(1, 20, "r18:performer"), (2, 21, "javbus:performer"),
(3, 23, "r18:performer"), (4, 22, "r18:performer"),
(5, 24, "r18:performer"), (6, 25, "javbus:performer")])
self.con.commit()
def tearDown(self):
self.con.close()
self.tmp.cleanup()
def drops(self):
return {str(row["drop_name"]): row for row in collect(self.con)}
def test_a_stage_name_registered_as_another_entity_alias_is_a_duplicate(self):
"""r18 早已把 `飯岡かなこ` 记成 `森泽佳奈` 的别名，那个名字却还自己占一条实体。"""
row = self.drops()["飯岡かなこ"]
self.assertEqual(row["keep_id"], 20)
self.assertEqual(row["keep_name"], "森泽佳奈")
self.assertIn("飯岡かなこ", str(row["match_evidence"]))
self.assertIn("r18:performer", str(row["match_evidence"]))
def test_an_entity_whose_own_aliases_name_someone_else_is_left_to_a_human(self):
"""`Ako Shiraishi` 的每个别名和 r18 引用都指向 平沢すず——规范名写错了人。
并进 `白石亚子` 就是把两位女优搅在一起，不可逆。这条要改的是规范名，不是合并。
"""
self.assertNotIn("Ako Shiraishi", self.drops())
def test_two_entities_with_different_ids_at_one_provider_are_two_people(self):
"""`NOZOMI` 是素人企划里的通名，撞上的是 `绫乃梓` 罗马字读音那条别名。
两条各有自己的 stash performer id，那就是站方也认为的两个人。
"""
self.assertNotIn("NOZOMI", self.drops())
def test_a_localization_alias_is_not_evidence_of_a_second_stage_name(self):
"""简繁与本地化别名说的是同一个名字的另一种写法，证不了这是那个人的另一个艺名。"""
self.con.execute(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(30,'performer','綾乃梓','綾乃梓','t','t')")
self.con.execute(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)"
" VALUES(24,'綾乃梓','綾乃梓','kanji-simplification',1.0)")
self.con.execute(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(1,30,'performer','legacy:asset',0.8)")
self.con.commit()
self.assertNotIn("綾乃梓", self.drops())
def test_the_merge_keeps_the_alias_holder_and_moves_its_assets(self):
counts = apply_rows(self.con, collect(self.con))
self.con.commit()
self.assertEqual(counts["merged"], 1)
self.assertIsNone(self.con.execute("SELECT 1 FROM entity WHERE id=21").fetchone())
self.assertEqual(self.con.execute(
"SELECT count(*) FROM asset_entity WHERE entity_id=20").fetchone()[0], 2)
self.assertIn("飯岡かなこ", [row[0] for row in self.con.execute(
"SELECT alias FROM entity_alias WHERE entity_id=20")])
class DirectoryIdentityTests(unittest.TestCase):
"""名录页号是身份，比名字和别名都强：换过艺名的人靠它才认得出来。"""
LINK = "http://www.prestige-av.com/special/shiraishi_ako.php"
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db = fresh_ledger(self.tmp.name)
self.con = sqlite3.connect(self.db)
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,code)"
" VALUES(?,'local',?,?,'video',?)",
[(1, "/x/1.mp4", "1.mp4", "ABP-951"), (2, "/x/2.mp4", "2.mp4", "MIZD-998")])
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,'performer',?,?,'t','t')",
[(40, "白石亚子", "白石亚子"), (41, "Ako Shiraishi", "ako shiraishi")])
self.con.executemany(
"INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)"
" VALUES(?,?,?,'r18:performer',0.9)",
[(40, "白石あこ", "白石あこ"), (41, "平沢すず", "平沢すず")])
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(?,?,'performer','r18:performer',0.9)",
[(1, 40), (2, 41)])
self.con.commit()
def tearDown(self):
self.con.close()
self.tmp.cleanup()
def install(self, entity_id, evidence, url=None):
self.con.execute(
"INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,is_sensitive,"
"metadata_json,created_at,updated_at)"
" VALUES(?,'official','T-POWERS',?,'www.prestige-av.com',0,?,'t','t')",
(entity_id, url or f"{self.LINK}?{entity_id}",
json.dumps({"evidence": evidence}, ensure_ascii=False)))
self.con.commit()
def test_two_entities_on_one_directory_page_are_one_person(self):
"""艺名换过一次，账本各存了一条；两侧各按自己的名字检索，落到同一页。"""
self.install(40, "minnano-av actress37250 资料表「公式サイト」；经「白石あこ」检索命中")
self.install(41, "minnano-av actress37250 资料表「公式サイト」；经「平沢すず」检索命中")
rows = collect(self.con)
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["keep_name"], "白石亚子")
self.assertEqual(rows[0]["drop_name"], "Ako Shiraishi")
self.assertEqual(rows[0]["evidence"], "名录页号")
self.assertIn("actress37250", str(rows[0]["match_evidence"]))
def test_the_localized_name_is_the_one_that_survives(self):
"""用户 2026-09-04 的口径：用更知名的名字，其余叫法留在别名里。"""
self.install(40, "minnano-av actress37250 经「白石あこ」检索命中")
self.install(41, "minnano-av actress37250 经「平沢すず」检索命中")
apply_rows(self.con, collect(self.con))
self.con.commit()
self.assertIsNone(self.con.execute("SELECT 1 FROM entity WHERE id=41").fetchone())
aliases = [row[0] for row in self.con.execute(
"SELECT alias FROM entity_alias WHERE entity_id=40")]
self.assertIn("Ako Shiraishi", aliases)
self.assertIn("平沢すず", aliases)
def test_two_different_pages_are_two_people(self):
"""同一家事务所的两位女优也会共用一条官网 URL——URL 不是身份，页号才是。"""
self.install(40, "minnano-av actress37250 经「白石あこ」检索命中", url=self.LINK)
self.install(41, "minnano-av actress99999 经「平沢すず」检索命中", url=self.LINK)
self.assertEqual(collect(self.con), [])
def test_a_link_without_a_page_number_is_not_evidence(self):
self.install(40, "X @shiraishi 简介外链")
self.install(41, "X @suzu 简介外链")
self.assertEqual(collect(self.con), [])
def test_three_entities_on_one_page_are_left_to_a_human(self):
"""三条要先决定并的顺序，那不是脚本该替人做的。"""
self.con.execute(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(42,'performer','平泽铃','平泽铃','t','t')")
self.con.execute(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(1,42,'performer','r18:performer',0.9)")
for entity_id in (40, 41, 42):
self.install(entity_id, f"minnano-av actress37250 经「{entity_id}」检索命中")
self.assertEqual(collect(self.con), [])
class NamedPairTests(unittest.TestCase):
"""命令行点名的一对：证据在站外时，唯一说得清来由的就是那段话。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db = fresh_ledger(self.tmp.name)
self.con = sqlite3.connect(self.db)
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,?,?,?,'t','t')",
[(1, "performer", "五十岚星兰", "五十岚星兰"),
(2, "performer", "五十嵐星蘭", "五十嵐星蘭"),
(3, "studio", "别的种类", "别的种类"),
(6, "performer", "加山優衣", "加山優衣")])
self.con.commit()
def tearDown(self):
self.con.close()
self.tmp.cleanup()
def pairs(self, *values, engaged=None):
return named_pairs(self.con, list(values), engaged)
def test_the_evidence_is_carried_through_to_the_review_file(self):
row = self.pairs("1:2:javdb 资料页把五个艺名列在一起")[0]
self.assertEqual(row["match_evidence"], "javdb 资料页把五个艺名列在一起")
self.assertEqual((row["keep_id"], row["drop_id"]), (1, 2))
self.assertEqual(row["drop_name"], "五十嵐星蘭")
def test_a_pair_without_evidence_is_refused(self):
"""合并不可逆。没有证据的一对说不清来由，比不合更糟。"""
with self.assertRaises(SystemExit):
self.pairs("1:2:")
with self.assertRaises(SystemExit):
self.pairs("1:2")
def test_two_different_kinds_are_never_the_same_identity(self):
with self.assertRaises(SystemExit):
self.pairs("1:3:随便什么理由")
def test_an_entity_already_planned_by_a_detector_is_refused(self):
"""自动判据先合会把 id 删掉，第二趟再引用同一个 id 就是合进一条不存在的实体。"""
with self.assertRaises(SystemExit):
self.pairs("1:2:证据", engaged={2})
def test_a_missing_entity_is_refused(self):
with self.assertRaises(SystemExit):
self.pairs("1:99:证据")
def test_one_entity_on_both_sides_is_refused(self):
with self.assertRaises(SystemExit):
self.pairs("1:1:证据")
def test_one_survivor_can_take_in_several_old_stage_names(self):
"""一个人有几个旧艺名就有几条要并进来，保留侧当然会重复出现。"""
rows = self.pairs("1:2:javdb 同一页", "1:6:javdb 同一页")
self.assertEqual([(row["keep_id"], row["drop_id"]) for row in rows],
[(1, 2), (1, 6)])
def test_an_entity_dropped_earlier_cannot_come_back_as_a_survivor(self):
"""前一对已经把它删了，再拿它当保留侧就是往一条不存在的实体里搬东西。"""
with self.assertRaises(SystemExit):
self.pairs("1:2:证据", "2:6:另一段证据")
def test_an_entity_dropped_earlier_cannot_be_dropped_twice(self):
with self.assertRaises(SystemExit):
self.pairs("1:2:证据", "6:2:另一段证据")
def test_a_survivor_cannot_turn_into_a_drop_later(self):
"""顺序写反：先说留它，后面又要丢它，两条不可能同时成立。"""
with self.assertRaises(SystemExit):
self.pairs("1:2:证据", "6:1:另一段证据")
def test_merging_series_rewrites_the_flat_column(self):
"""只搬关系不改 `asset.series`，卡片上还写着被丢弃的系列名。"""
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,'series',?,?,'t','t')",
[(4, "Night Safari", "night safari"), (5, "旧写法", "旧写法")])
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,code,series)"
" VALUES(?,'local',?,?,'video',?,?)",
[(1, "/x/1.mp4", "1.mp4", "AKA-022", "旧写法"),
(2, "/x/2.mp4", "2.mp4", "AKA-024", "Night Safari")])
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source)"
" VALUES(?,?,'series','r18:series')",
[(1, 5), (2, 4)])
self.con.commit()
with self.con:
counts = apply_rows(self.con, self.pairs("4:5:外部资料页说这两部同属一个系列"))
self.assertEqual(counts["merged"], 1)
self.assertEqual(
[row[0] for row in self.con.execute("SELECT series FROM asset ORDER BY id")],
["Night Safari", "Night Safari"])
class PairsOnlyTests(unittest.TestCase):
"""`--pairs-only` 只执行人工点名那批，自动判据整趟不跑。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.root = Path(self.tmp.name).resolve()
self.db = fresh_ledger(self.root)
self.con = sqlite3.connect(self.db)
self.con.executemany(
"INSERT INTO asset(id,location,path,name,medium,creator)"
" VALUES(?,'local',?,?,'video',?)",
[(1, "/x/1.mp4", "1.mp4", "小桃"), (2, "/x/2.mp4", "2.mp4", "小桃")])
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,?,?,?,'t','t')",
[(10, "creator", "小桃", "小桃"),
(11, "performer", "小桃", "小桃"),
(20, "performer", "五十岚星兰", "五十岚星兰"),
(21, "performer", "五十嵐星蘭", "五十嵐星蘭")])
self.con.executemany(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(?,?,?,?,1.0)",
[(1, 10, "creator", "legacy:asset"), (2, 10, "creator", "legacy:asset"),
(1, 11, "performer", "performer"),
(1, 21, "performer", "performer")])
self.con.commit()
self.con.close()
self.review = self.root / "merge.csv"
self.projections = self.root / "projections.csv"
def tearDown(self):
self.tmp.cleanup()
def invoke(self, *extra):
args = build_parser().parse_args([
"--db", str(self.db),
"--review-csv", str(self.review),
"--projection-review-csv", str(self.projections),
*extra,
])
return run(args)
def entity_ids(self):
connection = sqlite3.connect(self.db)
try:
return [row[0] for row in connection.execute(
"SELECT id FROM entity ORDER BY id")]
finally:
connection.close()
def test_the_detectors_still_run_without_the_switch(self):
"""对照：不给开关时，同名那组照旧进复核 CSV。"""
self.assertEqual(self.invoke(), 0)
self.assertIn("小桃", {row["drop_name"] for row in read_rows(self.review)})
def test_only_the_named_pair_reaches_the_review_file(self):
self.assertEqual(self.invoke("--pairs-only", "--pair", "20:21:javdb 同一页"), 0)
rows = read_rows(self.review)
self.assertEqual([row["drop_id"] for row in rows], ["21"])
self.assertEqual(rows[0]["evidence"], "人工指定")
self.assertEqual(read_rows(self.projections), [])
def test_the_detected_group_is_not_written_to_the_ledger(self):
self.assertEqual(self.invoke(
"--pairs-only", "--pair", "20:21:javdb 同一页",
"--apply", "--backup", str(self.root / "backup.db")), 0)
self.assertEqual(self.entity_ids(), [10, 11, 20])
def test_the_switch_without_a_pair_is_refused(self):
"""只处理人工那批却一对也没给，跑下去就是一趟什么都不做的空写入。"""
with self.assertRaises(SystemExit) as caught:
self.invoke("--pairs-only")
self.assertNotEqual(caught.exception.code, 0)
self.assertEqual(self.entity_ids(), [10, 11, 20, 21])
class ScrapedPerformerProvenanceTests(unittest.TestCase):
"""刮削链新建的 performer 也是发行元数据，保留它而不是目录名投影出来的 creator。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db = fresh_ledger(self.tmp.name)
self.con = sqlite3.connect(self.db)
self.con.execute(
"INSERT INTO asset(id,location,path,name,medium,creator)"
" VALUES(1,'local','/x/1.mp4','1.mp4','video','三上悠亜')")
self.con.executemany(
"INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES(?,?,'三上悠亜','三上悠亜','t','t')",
[(1, "creator"), (2, "performer")])
self.con.commit()
def tearDown(self):
self.con.close()
self.tmp.cleanup()
def link(self, source):
self.con.execute(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(1,1,'creator','legacy:asset',1.0)")
self.con.execute(
"INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
" VALUES(1,2,'performer',?,1.0)", (source,))
self.con.commit()
return collect(self.con)[0]
def test_a_javinizer_source_keeps_the_performer(self):
for source in ("javinizer:javdb:performer", "javinizer:r18dev:performer",
"javinizer:dmm:performer"):
with self.subTest(source=source):
self.con.execute("DELETE FROM asset_entity")
row = self.link(source)
self.assertEqual((row["keep_kind"], row["keep_id"]), ("performer", 2))
self.assertEqual(row["evidence"], "发行元数据")
def test_the_stash_flat_source_still_keeps_the_creator(self):
row = self.link("performer")
self.assertEqual((row["keep_kind"], row["keep_id"]), ("creator", 1))
if __name__ == "__main__":
unittest.main()