tests/test_r18_machine_translation_repair.py
Rama master · réplica de lectura del repositorio original.
"""早期 r18.dev 英文机翻修复：资产字段只改同源、未被用户接管的；实体只按快照日文那一侧改名。"""
import argparse
import csv
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from peach.entities import upsert_asset_entity
from peach.migrations import upgrade
from scripts.repair_r18_machine_translations import (apply_entity_rows, apply_rows, collect,
collect_entities, run)
ROOT = Path(__file__).resolve().parents[1]
class LedgerFixture(unittest.TestCase):
"""临时账本与快照目录；两组用例共用，本身不带用例。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
root = Path(self.tmp.name)
self.db = root / "ledger.db"
self.snapshots = root / "snapshots"
self.snapshots.mkdir()
upgrade(self.db, ROOT / "migrations")
self.con = sqlite3.connect(self.db)
self.con.row_factory = sqlite3.Row
def tearDown(self):
self.con.close()
self.tmp.cleanup()
def add_asset(self, asset_id: int, code: str, title: str, series: str, owners: dict[str, str]):
self.con.execute(
"INSERT INTO asset(id,location,path,name,medium,code,catalog_title,series,field_owners) "
"VALUES(?,'local',?,?,'video',?,?,?,?)",
(asset_id, f"{asset_id}.mp4", f"{asset_id}.mp4", code, title, series,
json.dumps(owners)),
)
self.con.commit()
def link_series(self, asset_id: int, name: str) -> int:
"""按自动落库那条路（`metadata_auto_apply`）的来源写法建系列实体。"""
entity_id = upsert_asset_entity(self.con, kind="series", name=name, asset_id=asset_id, role="series",
source="javinizer:r18dev:series")
self.con.commit()
return entity_id
def snapshot(self, code: str, *, title_machine=True, series_machine=True):
path = self.snapshots / f"{code}-r18dev.json"
path.write_text(json.dumps({
"id": code,
"combined": {
"dvd_id": code,
"title_en": "Machine title", "title_en_is_machine_translation": title_machine,
"title_ja": "日本語タイトル",
"series_name_en": "Machine series",
"series_name_en_is_machine_translation": series_machine,
"series_name_ja": "日本語シリーズ",
},
"translations": [{"language": "ja", "title": "日本語タイトル", "series": "日本語シリーズ"}],
}, ensure_ascii=False), encoding="utf-8")
class R18MachineTranslationRepairTests(LedgerFixture):
def test_collect_and_apply_replaces_both_verified_fields(self):
self.add_asset(1, "ABC-001", "Machine title", "Machine series", {
"catalog_title": "auto:r18dev", "series": "auto:r18dev",
})
self.snapshot("ABC-001")
rows = collect(self.con, self.snapshots)
self.assertEqual([row["action"] for row in rows], ["replace", "replace"])
self.assertEqual(apply_rows(self.con, rows), {"assets": 1, "fields": 2})
self.con.commit()
asset = self.con.execute(
"SELECT catalog_title,series,field_owners,mutation_revision FROM asset WHERE id=1").fetchone()
self.assertEqual((asset["catalog_title"], asset["series"]),
("日本語タイトル", "日本語シリーズ"))
self.assertEqual(json.loads(asset["field_owners"]), {
"catalog_title": "auto:r18dev", "series": "auto:r18dev",
})
self.assertEqual(asset["mutation_revision"], 1)
def test_changed_or_protected_values_are_not_candidates(self):
self.add_asset(2, "ABC-002", "User title", "Changed series", {
"catalog_title": "user:manual", "series": "auto:r18dev",
})
self.snapshot("ABC-002")
rows = collect(self.con, self.snapshots)
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["field"], "series")
self.assertEqual(rows[0]["action"], "skip")
self.assertEqual(rows[0]["reason"], "current-no-longer-matches-snapshot-english")
def test_bilingual_snapshot_is_enough_when_r18_did_not_mark_the_english_as_machine_translation(self):
self.add_asset(3, "ABC-003", "Machine title", "Machine series", {
"catalog_title": "auto:r18dev", "series": "auto:r18dev",
})
self.snapshot("ABC-003", title_machine=False, series_machine=False)
rows = collect(self.con, self.snapshots)
self.assertEqual({row["action"] for row in rows}, {"replace"})
self.assertEqual({row["machine_translation"] for row in rows}, {"no"})
def test_a_latin_only_spacing_variant_is_not_mistaken_for_japanese(self):
self.add_asset(5, "ABC-005", "Machine title", "1 VS 1", {
"catalog_title": "auto:r18dev", "series": "auto:r18dev",
})
self.snapshot("ABC-005")
payload = json.loads((self.snapshots / "ABC-005-r18dev.json").read_text(encoding="utf-8"))
payload["combined"]["series_name_en"] = "1 VS 1"
payload["combined"]["series_name_ja"] = "1VS1"
(self.snapshots / "ABC-005-r18dev.json").write_text(
json.dumps(payload, ensure_ascii=False), encoding="utf-8")
rows = collect(self.con, self.snapshots)
series = next(row for row in rows if row["field"] == "series")
self.assertEqual(series["action"], "skip")
self.assertEqual(series["reason"], "target-has-no-japanese-script")
def test_dry_run_writes_audit_but_not_ledger_and_apply_requires_backup(self):
self.add_asset(4, "ABC-004", "Machine title", "Machine series", {
"catalog_title": "auto:r18dev", "series": "auto:r18dev",
})
self.snapshot("ABC-004")
self.link_series(4, "Machine series")
audit = Path(self.tmp.name) / "audit.csv"
entity_audit = Path(self.tmp.name) / "entities.csv"
args = argparse.Namespace(
db=self.db, apply=False, backup=None, snapshots=self.snapshots, audit_csv=audit,
javinizer_snapshots=Path(self.tmp.name) / "javinizer", entity_audit_csv=entity_audit,
)
self.assertEqual(run(args), 0)
self.assertTrue(audit.is_file())
with entity_audit.open(encoding="utf-8-sig", newline="") as handle:
self.assertEqual([(row["current"], row["action"]) for row in csv.DictReader(handle)],
[("Machine series", "rename")])
self.assertEqual(self.con.execute(
"SELECT catalog_title FROM asset WHERE id=4").fetchone()[0], "Machine title")
self.assertEqual(self.con.execute(
"SELECT canonical_name FROM entity WHERE kind='series'").fetchone()[0], "Machine series")
args.apply = True
with self.assertRaisesRegex(SystemExit, "--backup"):
run(args)
class R18EntityNameRepairTests(LedgerFixture):
"""实体那一遍：r18 英文写法建出的系列与女优，按快照日文那一侧改名；证据不足或撞名只报不改。"""
def rows(self):
return collect_entities(self.con, self.snapshots, Path(self.tmp.name) / "javinizer")
def test_a_translated_series_is_renamed_and_keeps_the_english_as_an_alias(self):
self.add_asset(10, "ABC-010", "日本語タイトル", "Machine series", {"series": "auto:r18dev"})
self.snapshot("ABC-010")
entity_id = self.link_series(10, "Machine series")
rows = self.rows()
self.assertEqual([(row["entity_id"], row["action"], row["target"], row["reason"]) for row in rows],
[(entity_id, "rename", "日本語シリーズ", "verified-r18dev-japanese")])
counts = apply_entity_rows(self.con, rows, "2026-09-28T00:00:00+00:00")
self.assertEqual((counts["renamed"], counts["aliases"], counts["flat_rewritten"]), (1, 1, 1))
self.assertEqual(self.con.execute(
"SELECT canonical_name FROM entity WHERE id=?", (entity_id,)).fetchone()[0], "日本語シリーズ")
self.assertEqual([tuple(row) for row in self.con.execute(
"SELECT alias,source FROM entity_alias WHERE entity_id=?", (entity_id,))],
[("Machine series", "script:repair_r18_machine_translations")])
self.assertEqual(self.con.execute("SELECT series FROM asset WHERE id=10").fetchone()[0], "日本語シリーズ")
def test_a_name_the_japanese_side_writes_the_same_way_is_the_original(self):
"""DMM 登记的艺名本来就写拉丁字母（``RARA``）；日文那一侧同名就是原名，不改。"""
self.add_asset(11, "ABC-011", "日本語タイトル", "", {})
(self.snapshots / "ABC-011-r18dev.json").write_text(json.dumps({
"id": "ABC-011", "combined": {"dvd_id": "ABC-011", "actresses": [
{"name_romaji": "RARA", "name_kanji": "RARA"}]}}), encoding="utf-8")
upsert_asset_entity(self.con, kind="performer", name="RARA", asset_id=11, role="performer",
source="javinizer:r18dev:performer")
rows = self.rows()
self.assertEqual([(row["action"], row["reason"]) for row in rows], [("skip", "already-original")])
def test_a_japanese_name_owned_by_another_entity_is_reported_for_merging(self):
self.add_asset(12, "ABC-012", "日本語タイトル", "Machine series", {"series": "auto:r18dev"})
self.snapshot("ABC-012")
entity_id = self.link_series(12, "Machine series")
self.add_asset(13, "ABC-013", "日本語タイトル", "日本語シリーズ", {})
other = upsert_asset_entity(self.con, kind="series", name="日本語シリーズ", asset_id=13, role="series",
source="manual:series")
rows = self.rows()
self.assertEqual([(row["entity_id"], row["action"], row["conflict_entity_id"]) for row in rows],
[(entity_id, "merge-needed", other)])
self.assertEqual(apply_entity_rows(self.con, rows, "2026-09-28T00:00:00+00:00")["renamed"], 0)
self.assertEqual(self.con.execute(
"SELECT canonical_name FROM entity WHERE id=?", (entity_id,)).fetchone()[0], "Machine series")
def test_without_a_snapshot_the_entity_is_only_reported(self):
self.add_asset(14, "ABC-014", "日本語タイトル", "Machine series", {"series": "auto:r18dev"})
self.link_series(14, "Machine series")
self.assertEqual([(row["action"], row["reason"]) for row in self.rows()], [("skip", "missing-snapshot")])
if __name__ == "__main__":
unittest.main()