import importlib.util import json import sqlite3 import sys import tempfile import unittest from unittest import mock from pathlib import Path SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" def _load(name): spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py") module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module _script = _load("audit_code_creators") confirm = _load("confirm_code_dirs") class _Audit: """判定与清理归 `peach.code_creators`，命令行入口留在脚本里；测试两边都要够得着。""" from peach.catalog_rules import release_code_from_filename as code_from_filename from peach.catalog_rules import release_code_from_text as canonical_code from peach.code_creators import ( # noqa: F401 逐个点名，测试不跟着模块表面漂移 FIELDS, VERDICT_CODE, VERDICT_KEEP, VERDICT_SITE, VERDICT_UNCLEAR, apply_rows, classify, collect, is_filesystem_path, ) build_parser = staticmethod(_script.build_parser) run = staticmethod(_script.run) main = staticmethod(_script.main) audit = _Audit SCHEMA = """ CREATE TABLE asset( id INTEGER PRIMARY KEY, location TEXT NOT NULL, path TEXT NOT NULL, name TEXT, medium TEXT, creator TEXT, code TEXT, duration REAL, field_owners TEXT, mutation_revision INTEGER NOT NULL DEFAULT 0, UNIQUE(location,path)); CREATE TABLE entity( id INTEGER PRIMARY KEY, kind TEXT, canonical_name TEXT, normalized_name TEXT, metadata_json TEXT DEFAULT '{}', created_at TEXT, updated_at TEXT, UNIQUE(kind,normalized_name)); CREATE TABLE asset_entity( asset_id INTEGER, entity_id INTEGER, role TEXT, source TEXT, confidence REAL DEFAULT 1.0, metadata_json TEXT DEFAULT '{}', first_seen_at TEXT, last_seen_at TEXT, UNIQUE(asset_id,entity_id,role,source)); """ class CodeShapeTests(unittest.TestCase): def test_digitless_creator_names_do_not_parse_their_files_as_embedded_codes(self): from peach import code_creators assets = [{'path': f'/media/{index}.mp4', 'name': f'ABP-{index:03d}.mp4', 'code': ''} for index in range(1000)] with mock.patch.object(code_creators, 'code_from_filename', side_effect=AssertionError('无需解析影片名')): self.assertEqual(code_creators.classify('Example Creator', assets)[0], audit.VERDICT_KEEP) def test_quality_prefix_is_not_part_of_the_code(self): # 这批目录名正是本次要修的：HD 是画质，不是番号的一部分。 self.assertEqual(audit.canonical_code("HD-abp-758"), "ABP-758") self.assertEqual(audit.canonical_code("HD_MIZD-997"), "MIZD-997") self.assertEqual(audit.canonical_code("HD_hrv-041"), "HRV-041") self.assertEqual(audit.canonical_code("FHD-ssis-070"), "SSIS-070") def test_version_suffix_is_stripped(self): self.assertEqual(audit.canonical_code("pppd-937ch"), "PPPD-937") self.assertEqual(audit.canonical_code("KBI044C"), "KBI-044") self.assertEqual(audit.canonical_code("NHDTB500C"), "NHDTB-500") def test_digits_are_normalised_to_three_places(self): self.assertEqual(audit.canonical_code("abw104"), "ABW-104") self.assertEqual(audit.canonical_code("IPVR00296"), "IPVR-296") def test_date_code_systems(self): # 分隔符是片商标识：加勒比写 `-`、一本道写 `_`，同日同序号是两部不同影片， # 目录名给的是哪一个就留哪一个。 self.assertEqual(audit.canonical_code("Carib-040221-001-FHD"), "040221-001") self.assertEqual(audit.canonical_code("1pondo-071213_625-HD"), "071213_625") def test_names_that_are_not_codes(self): for name in ("涼森れむ", "suzuq", "Timepasserby", "MyElla"): self.assertIsNone(audit.canonical_code(name), name) def test_uuid_first_segment_is_not_a_code(self): self.assertIsNone( audit.canonical_code("DCE7230C-730E-45DA-9FDC-D167663BC84E")) def test_filename_noise_is_stripped_before_matching(self): self.assertEqual(audit.code_from_filename("HD-abp-758.mp4"), "ABP-758") self.assertEqual(audit.code_from_filename("HD_hrv-041-1.mp4"), "HRV-041") self.assertEqual(audit.code_from_filename("BNST033(2).jpg"), "BNST-033") self.assertEqual(audit.code_from_filename("040221-001-carib-1080p.mp4"), "040221-001") def test_online_identities_are_not_filesystem_paths(self): self.assertFalse(audit.is_filesystem_path("https://www.pixiv.net/users/93812377")) self.assertTrue(audit.is_filesystem_path(r"B:\云下载\HD-abp-758\HD-abp-758.mp4")) def _row(name, path, code=None): return {"name": name, "path": path, "code": code} class ClassifyTests(unittest.TestCase): def test_release_folder_named_after_its_own_code(self): verdict, identity, _ = audit.classify( "HD-abp-758", [_row("HD-abp-758.mp4", r"B:\云下载\HD-abp-758\HD-abp-758.mp4")]) self.assertEqual((verdict, identity), (audit.VERDICT_CODE, "ABP-758")) def test_existing_code_column_counts_as_evidence(self): verdict, identity, _ = audit.classify( "pppd-937ch", [_row("PPPD-937CH.mp4", r"B:\云下载\pppd-937ch\PPPD-937CH.mp4", "PPPD-937")]) self.assertEqual((verdict, identity), (audit.VERDICT_CODE, "PPPD-937")) def test_uploader_account_with_title_filenames_is_not_touched(self): # banbi_555 形状上像番号，但目录里全是作品标题，没有同番号文件。 verdict, _, _ = audit.classify("banbi_555", [ _row("18歳Eカップ彼氏持ち美女.mp4", r"A:\Pack From Shared\pen\banbi_555\18歳.mp4"), _row("2_2024_06_08_172104.mp4", r"A:\Pack From Shared\pen\banbi_555\2_2024.mp4"), ]) self.assertEqual(verdict, audit.VERDICT_UNCLEAR) def test_a_download_site_folder_around_the_code_is_still_a_release_folder(self): """站名、画质和分享标记贴在番号前后，目录仍是发行目录，不是创作者。 本机实测的写法：`Jav.li_MIAD573_HD`、`[98t.tv][98t.tv]ABW-251`、 `nes@第一会所@ATID-479`、`kpxvs-300MIUM-698`。整名比对认不出它们，而 11 个这样的 「创作者」名下挂着 33 条资产，在创作者索引里各占一个假身份。 """ for name, sample, code, identity in ( ("Jav.li_MIAD573_HD", "MIAD573_01.wmv", "MIAD-573", "MIAD-573"), ("[98t.tv][98t.tv]ABW-251", "ABW-251.mp4", "", "ABW-251"), ("nes@第一会所@ATID-479", "ATID-479.mp4", "", "ATID-479"), ): with self.subTest(name=name): verdict, found, _ = audit.classify( name, [_row(sample, rf"B:\云下载\{name}\{sample}", code)]) self.assertEqual((verdict, found), (audit.VERDICT_CODE, identity)) def test_an_uploader_account_shaped_like_a_code_is_not_matched_by_containment(self): """按包含关系找番号时，上传者账号是最容易被误伤的一类。 `banbi_555` 的 code 列存的就是目录名 `BANBI_555`，紧凑形一比自然「包含」。 发行番号这道形态门槛把它挡在外面，所以它仍然只是存疑，留给人看。 """ verdict, _, _ = audit.classify("banbi_555", [ _row("18歳Eカップ彼氏持ち美女.mp4", r"A:\Pack From Shared\pen\banbi_555\18歳.mp4", "BANBI_555"), ]) self.assertEqual(verdict, audit.VERDICT_UNCLEAR) def test_pixiv_artist_is_kept_out_of_the_review_entirely(self): verdict, _, _ = audit.classify( "AH18", [_row("AH18", "https://www.pixiv.net/users/93812377")]) self.assertEqual(verdict, audit.VERDICT_KEEP) def test_site_post_id_is_classified_apart_from_codes(self): verdict, identity, _ = audit.classify("fantia-3760310", [ _row("avruby.png", r"B:\xxr\fantia-3760310\avruby.png"), _row("fantia-3760310.mp4", r"B:\xxr\fantia-3760310\fantia-3760310.mp4"), ]) self.assertEqual((verdict, identity), (audit.VERDICT_SITE, "fantia-3760310")) class ApplyTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.db = Path(self.tmp.name) / "ledger.db" self.connection = sqlite3.connect(self.db) self.addCleanup(self.connection.close) self.connection.executescript(SCHEMA) # 媒介按扩展名给，和真实导入一致：目录里混着图片正是本组要覆盖的形态。 def _creator(self, entity_id, name, assets): self.connection.execute( "INSERT INTO entity(id,kind,canonical_name,normalized_name) VALUES(?,'creator',?,?)", (entity_id, name, name.casefold())) for asset_id, asset_name, path, code in assets: self.connection.execute( "INSERT INTO asset(id,location,path,name,medium,creator,code) " "VALUES(?,'local',?,?,?,?,?)", (asset_id, path, asset_name, "image" if asset_name.lower().endswith((".jpg", ".png")) else "video", name, code)) self.connection.execute( "INSERT INTO asset_entity(asset_id,entity_id,role,source) " "VALUES(?,?,'creator','legacy:asset')", (asset_id, entity_id)) def test_apply_removes_the_false_creator_and_fills_the_code(self): self._creator(1, "HD-abp-758", [ (10, "HD-abp-758.mp4", r"B:\云下载\HD-abp-758\HD-abp-758.mp4", None)]) self._creator(2, "banbi_555", [ (20, "18歳.mp4", r"A:\Pack From Shared\pen\banbi_555\18歳.mp4", None)]) self.connection.commit() rows = audit.collect(self.connection) counts = audit.apply_rows(self.connection, rows) self.assertEqual(counts["links"], 1) self.assertEqual(counts["entities"], 1) self.assertEqual(counts["codes"], 1) self.assertEqual( self.connection.execute("SELECT creator,code FROM asset WHERE id=10").fetchone(), (None, "ABP-758")) self.assertIsNone( self.connection.execute("SELECT id FROM entity WHERE id=1").fetchone()) # 存疑的上传者账号必须原样保留：实体、关系和扁平字段都不动。 self.assertIsNotNone( self.connection.execute("SELECT id FROM entity WHERE id=2").fetchone()) self.assertEqual( self.connection.execute("SELECT creator FROM asset WHERE id=20").fetchone()[0], "banbi_555") def test_the_code_it_writes_names_its_owner(self): self._creator(6, "HD-abp-758", [ (60, "HD-abp-758.mp4", r"B:\云下载\HD-abp-758\HD-abp-758.mp4", None)]) self.connection.commit() audit.apply_rows(self.connection, audit.collect(self.connection)) self.assertEqual( json.loads(self.connection.execute( "SELECT field_owners FROM asset WHERE id=60").fetchone()[0]), {"code": "script:code-creators", "creator": "script:code-creators"}) def test_a_code_the_user_cleared_stays_cleared(self): # 用户把某条的 code 清空是一次判断，补空不该把它再填回去。 self._creator(7, "Tokyo-Hot n0781-HD", [ (70, "Tokyo-Hot.mp4", r"B:\云下载\Tokyo-Hot n0781-HD\Tokyo-Hot.mp4", "n0781"), (71, "花絮.mp4", r"B:\云下载\Tokyo-Hot n0781-HD\花絮.mp4", None)]) self.connection.execute( """UPDATE asset SET field_owners='{"code":"user:manual"}' WHERE id=71""") self.connection.commit() counts = audit.apply_rows(self.connection, audit.collect(self.connection)) self.assertEqual(counts["codes"], 0) self.assertIsNone( self.connection.execute("SELECT code FROM asset WHERE id=71").fetchone()[0]) def test_existing_code_is_never_overwritten(self): self._creator(3, "MIDA-117ch", [ (30, "MIDA-117CH.mp4", r"B:\云下载\MIDA-117ch\MIDA-117CH.mp4", "MIDA-117")]) self.connection.commit() audit.apply_rows(self.connection, audit.collect(self.connection)) self.assertEqual( self.connection.execute("SELECT code FROM asset WHERE id=30").fetchone()[0], "MIDA-117") def test_site_post_id_clears_the_creator_without_writing_a_code(self): self._creator(4, "fantia-3760310", [ (40, "fantia-3760310.mp4", r"B:\xxr\fantia-3760310\fantia-3760310.mp4", None)]) self.connection.commit() audit.apply_rows(self.connection, audit.collect(self.connection)) self.assertEqual( self.connection.execute("SELECT creator,code FROM asset WHERE id=40").fetchone(), (None, None)) def test_apply_refuses_to_run_without_a_backup(self): parser = audit.build_parser() with self.assertRaises(SystemExit): audit.run(parser.parse_args(["--db", str(self.db), "--apply"])) class DurationEvidenceTests(unittest.TestCase): """目录名像番号、文件却不带番号时，片长是第二条证据（用户 2026-09-16）。""" def test_a_couple_of_minutes_of_ads_still_counts_as_the_same_release(self): confirmed, _ = confirm.duration_confirms_code(98 * 60 + 90, 98) self.assertTrue(confirmed) def test_a_different_length_is_a_different_thing(self): # `bbsxv.xyz-DOCP-324` 目录里只有一条 90 秒的广告，DOCP-324 本身 120 分钟。 confirmed, reason = confirm.duration_confirms_code(90, 120) self.assertFalse(confirmed) self.assertIn("超出容差", reason) def test_no_runtime_from_the_source_decides_nothing(self): self.assertEqual(confirm.duration_confirms_code(3600, None), (False, "来源没有给片长")) def test_the_longest_video_is_the_feature(self): """目录里还躺着论坛文宣和封面图，拿它们比片长必然对不上。""" assets = [ {"medium": "video", "duration": 5896.8}, {"medium": "video", "duration": 31.0}, {"medium": "image", "duration": None}, ] self.assertEqual(confirm.main_video_seconds(assets), 5896.8) self.assertIsNone(confirm.main_video_seconds([{"medium": "image", "duration": None}])) class ConfirmScriptTests(unittest.TestCase): class _Provider: """按番号答片长的假来源。""" def __init__(self, runtimes): self.runtimes, self.asked = runtimes, [] def query(self, code, **_): self.asked.append(code) raise LookupError("r18.dev 没有这个番号") def community(self, code, **_): runtime = self.runtimes.get(code) if runtime is None: raise LookupError("社区来源都没有这个番号") # 真实返回顺序里 javbus 排在 javdb 前面，`n0780` 它报的是 36 分。 return [("javbus", {"runtime": 36}), ("javdb", {"runtime": runtime})] def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.db = Path(self.tmp.name) / "ledger.db" self.connection = sqlite3.connect(self.db) self.addCleanup(self.connection.close) self.connection.executescript(SCHEMA) def _creator(self, entity_id, name, assets): self.connection.execute( "INSERT INTO entity(id,kind,canonical_name,normalized_name) VALUES(?,'creator',?,?)", (entity_id, name, name.casefold())) for asset_id, asset_name, path, medium, duration in assets: self.connection.execute( "INSERT INTO asset(id,location,path,name,medium,creator,duration) " "VALUES(?,'local',?,?,?,?,?)", (asset_id, path, asset_name, medium, name, duration)) self.connection.execute( "INSERT INTO asset_entity(asset_id,entity_id,role,source) " "VALUES(?,?,'creator','legacy:asset')", (asset_id, entity_id)) def test_a_release_folder_is_confirmed_by_its_runtime_and_an_account_is_not(self): self._creator(1, "Tokyo-Hot n0780-HD", [ (10, "Tokyo-Hot.mp4", r"B:\云下载\Tokyo-Hot n0780-HD\Tokyo-Hot.mp4", "video", 5896.8), (11, "封殺001.jpg", r"B:\云下载\Tokyo-Hot n0780-HD\論壇文宣\封殺001.jpg", "image", None)]) self._creator(2, "banbi_555", [ (20, "18歳.mp4", r"A:\Pack From Shared\pen\banbi_555\18歳.mp4", "video", 1800.0)]) self.connection.commit() provider = self._Provider({"n0780": 98}) rows = confirm.examine(self.connection, provider) verdicts = {row["creator"]: row["confirmed"] for row in rows} self.assertEqual(verdicts, {"Tokyo-Hot n0780-HD": "是", "banbi_555": "否"}) self.assertEqual(provider.asked, ["n0780", "BANBI-555"]) # 片长取 javdb 的：同一个番号 javbus 报 36 分，盘里那条 98 分。 self.assertEqual([row["source"] for row in rows if row["confirmed"] == "是"], ["javdb"]) counts = audit.apply_rows( self.connection, [row for row in rows if row["verdict"] == audit.VERDICT_CODE]) self.assertEqual(counts["codes"], 1) self.assertEqual( self.connection.execute("SELECT creator,code FROM asset WHERE id=10").fetchone(), (None, "n0780")) # 目录里的论坛文宣不是这部片的文件：假创作者照样摘掉，番号不给它。 self.assertEqual( self.connection.execute("SELECT creator,code FROM asset WHERE id=11").fetchone(), (None, None)) # 片长对不上的目录一条都不动。 self.assertEqual( self.connection.execute("SELECT creator FROM asset WHERE id=20").fetchone()[0], "banbi_555") def test_apply_refuses_to_run_without_a_backup(self): parser = confirm.build_parser() with self.assertRaises(SystemExit): confirm.run(parser.parse_args(["--db", str(self.db), "--apply"])) if __name__ == "__main__": unittest.main()