tests/test_entity_merge.py

Rama master · réplica de lectura del repositorio original.

"""实体合并：同一位女优的旧艺名与现用艺名各成一个实体时的归并。

合并不可逆，所以关系迁移、别名保留和唯一约束冲突的处理都要有回归。
"""
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path

from peach.entities import (
    canonicalize_entity_name,
    collapse_repeated_entity_name,
    merge_entity,
    normalize_entity_name,
    resolve_entity,
    upsert_asset_entity,
)
from peach.migrations import upgrade

ROOT = Path(__file__).resolve().parents[1]


class EntityMergeTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = Path(self.tmp.name) / "ledger.db"
        upgrade(self.db, ROOT / "migrations")
        self.con = sqlite3.connect(self.db)
        for asset_id in (1, 2, 3):
            self.con.execute(
                "INSERT INTO asset(id,location,path,name,medium) VALUES(?,?,?,?,'video')",
                (asset_id, "local", f"/x/{asset_id}.mp4", f"{asset_id}.mp4"))
        for entity_id, name in ((10, "橋本ありな"), (11, "新ありな")):
            self.con.execute(
                "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
                " VALUES(?,'performer',?,?,'t','t')", (entity_id, name, name.lower()))
        self.con.executemany(
            "INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)"
            " VALUES(?,?,'performer','r18',1.0)",
            [(1, 10), (2, 10), (2, 11), (3, 11)])
        self.con.commit()

    def tearDown(self):
        self.con.close()
        self.tmp.cleanup()

    def test_merge_moves_relations_and_keeps_the_old_name_searchable(self):
        moved = merge_entity(self.con, target_id=10, source_id=11,
                             source_name="新ありな", alias_source="r18:performer")
        self.con.commit()

        assets = {r[0] for r in self.con.execute(
            "SELECT asset_id FROM asset_entity WHERE entity_id=10")}
        self.assertEqual(assets, {1, 2, 3}, "三条关系都应归到保留实体")
        self.assertEqual(moved["assets"], 1, "asset 2 两边都有，去重后只迁移 asset 3")
        self.assertIsNone(self.con.execute(
            "SELECT 1 FROM entity WHERE id=11").fetchone(), "被并入的实体应删除")
        aliases = {r[0] for r in self.con.execute(
            "SELECT alias FROM entity_alias WHERE entity_id=10")}
        self.assertIn("新ありな", aliases, "旧名必须留作别名，否则按它搜索会落空")
        self.assertEqual(moved["aliases"], 0, "source 无别名时计数应为 0")

    def test_merge_moves_source_aliases_and_counts_them(self):
        self.con.execute(
            "INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)"
            " VALUES(11,'ありな','ありな','r18',1.0)")
        self.con.commit()
        moved = merge_entity(self.con, target_id=10, source_id=11,
                             source_name="新ありな", alias_source="r18:performer")
        self.con.commit()
        self.assertEqual(moved["aliases"], 1, "source 的别名应被迁移并计数")

    def test_person_merge_keeps_identity_names_and_drops_collection_spellings(self):
        for name in ("新ありな4K", "新ありな 合集", "黑丝美腿", "ABP-123"):
            self.con.execute(
                "INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)"
                " VALUES(11,?,?,'r18',1.0)", (name, normalize_entity_name(name)))
        self.con.execute(
            "INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)"
            " VALUES(11,'ありな','ありな','r18',1.0)")
        merge_entity(self.con, target_id=10, source_id=11,
                     source_name="新ありな 1080p", alias_source="user:review")
        aliases = {r[0] for r in self.con.execute("SELECT alias FROM entity_alias WHERE entity_id=10")}
        self.assertEqual(aliases, {"ありな"})
        self.assertEqual(self.con.execute("SELECT target_id FROM entity_redirect WHERE old_id=11").fetchone(), (10,))

    def test_tag_merge_keeps_the_retired_name_as_alias(self):
        for entity_id, name in ((20, "合集"), (21, "混合集")):
            self.con.execute(
                "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)"
                " VALUES(?,'tag',?,?,'t','t')", (entity_id, name, name))
        merge_entity(self.con, target_id=20, source_id=21, source_name="混合集", alias_source="retired")
        self.assertEqual(self.con.execute("SELECT alias FROM entity_alias WHERE entity_id=20").fetchall(),
                         [("混合集",)])

    def test_two_pages_on_the_same_site_both_follow_the_merge(self):
        """同一个站上的两个页面是两条引用，合并后都归目标实体（0032）。

        这两条各自对应站上一个真实页面，挂的作品不同。旧的
        `UNIQUE(entity_id,provider,external_kind)` 会丢掉其中一条，等于合并一次就少一个
        能点进去的页面。
        """
        self.con.executemany(
            "INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id)"
            " VALUES(?,'javdb','performer',?)", [(10, "d45k9"), (11, "ZX5z7")])
        self.con.commit()
        moved = merge_entity(self.con, target_id=10, source_id=11,
                             source_name="新ありな", alias_source="r18:performer")
        self.con.commit()
        self.assertEqual(moved["refs"], 1)
        self.assertEqual(moved["dropped_refs"], 0)
        kept = self.con.execute(
            "SELECT external_id FROM entity_external_ref WHERE entity_id=10"
            " ORDER BY external_id").fetchall()
        self.assertEqual(kept, [("ZX5z7",), ("d45k9",)])

    def test_merge_leaves_no_dangling_rows(self):
        merge_entity(self.con, target_id=10, source_id=11,
                     source_name="新ありな", alias_source="r18:performer")
        self.con.commit()
        for table in ("asset_entity", "entity_alias", "entity_external_ref",
                      "entity_link", "entity_search_term"):
            left = self.con.execute(
                f"SELECT count(*) FROM {table} WHERE entity_id=11").fetchone()[0]
            self.assertEqual(left, 0, f"{table} 不应残留被删实体的行")
        self.assertEqual(
            self.con.execute("PRAGMA foreign_key_check").fetchall(), [])

    def test_a_second_id_from_the_same_source_is_dropped_not_raised(self):
        """合并过的人在来源那边还是两个页面，第二个 id 只能丢掉。

        实体 8004「桥本有菜」名下挂着「橋本ありな」「新ありな」等 8 个别名，r18dev 给
        前者 1032668、给后者 1078619，账本存的是后者。按别名认出同一个人之后再插第二个
        id，撞的是 `UNIQUE(entity_id,provider,external_kind)`——插入语句的 `ON CONFLICT`
        只认主键那一组，于是一条落库把整批自动落库连同别的字段一起带走。
        """
        self.con.execute(
            "INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence) "
            "VALUES(10,'橋本ありな別名','橋本ありな別名','mapping',1.0)")
        self.con.execute(
            "INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id)"
            " VALUES(10,'r18dev','performer','1078619')")
        self.con.commit()
        entity_id = upsert_asset_entity(
            self.con, kind="performer", name="橋本ありな別名", asset_id=3,
            role="performer", source="javinizer:r18dev:performer",
            external_provider="r18dev", external_id="1032668")
        self.con.commit()
        self.assertEqual(entity_id, 10)
        self.assertEqual(
            self.con.execute(
                "SELECT external_id FROM entity_external_ref "
                "WHERE entity_id=10 AND provider='r18dev'").fetchall(),
            [("1078619",)], "先到的那条留着")
        self.assertEqual(
            self.con.execute(
                "SELECT count(*) FROM asset_entity WHERE asset_id=3 AND entity_id=10 "
                "AND source='javinizer:r18dev:performer'").fetchone()[0], 1,
            "引用写不进去不影响这条出演关系本身")

    def test_a_rescraped_credit_keeps_the_recorded_javdb_record_kind(self):
        """作品页署名带不出有碼还是無碼，重采时换掉资料也得留着之前记下的那项（#271）。"""
        self.con.execute(
            "INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id,metadata_json)"
            " VALUES(10,'javdb','performer','bkxd','{\"source\": \"javdb\", \"uncensored\": false}')")
        self.con.commit()
        upsert_asset_entity(
            self.con, kind="performer", name="橋本ありな", asset_id=1, role="performer",
            source="javdb", external_provider="javdb", external_id="bkxd",
            metadata={"source": "javdb", "provider_id": "NNPJ-466"})
        self.con.commit()
        metadata = json.loads(self.con.execute(
            "SELECT metadata_json FROM entity_external_ref WHERE external_id='bkxd'").fetchone()[0])
        self.assertEqual((metadata["provider_id"], metadata["uncensored"]), ("NNPJ-466", False))

    def test_person_upsert_collapses_repeated_name_and_reuses_unique_alias(self):
        self.con.execute(
            "INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence) "
            "VALUES(10,'Hashimoto Arina','hashimoto arina','mapping',1.0)"
        )
        entity_id = upsert_asset_entity(
            self.con, kind="performer", name="Hashimoto Arina Hashimoto Arina",
            asset_id=3, role="performer", source="javbus:performer",
        )
        self.assertEqual(entity_id, 10)
        self.assertEqual(self.con.execute(
            "SELECT count(*) FROM entity WHERE kind='performer'"
        ).fetchone()[0], 2, "旧名别名不得被重新建成第三个实体")
        self.assertEqual(collapse_repeated_entity_name("A B A B"), "A B")
        self.assertEqual(canonicalize_entity_name("performer", "画像を拡大する"), "")

    def test_studio_ingest_reuses_official_names_and_registered_aliases(self):
        target = upsert_asset_entity(self.con, kind='studio', name='Attackers',
                                    asset_id=1, role='studio', source='r18:studio')
        self.con.execute("INSERT INTO entity_alias VALUES(?,? ,?,'mapping',1)",
                         (target, 'アタッカーズ', 'アタッカーズ'))
        for name in ('ATTACKERS（アタッカーズ）', 'アタッカーズ'):
            self.assertEqual(upsert_asset_entity(self.con, kind='studio', name=name,
                asset_id=3, role='studio', source='makers:studio'), target)
            self.assertEqual(resolve_entity(self.con, 'studio', name)[0], target)
        self.assertEqual(self.con.execute("SELECT count(*) FROM entity WHERE kind='studio'").fetchone()[0], 1)

    def test_studio_ingest_preserves_ambiguous_aliases(self):
        for name in ('Alpha', 'Beta'):
            entity_id = upsert_asset_entity(self.con, kind='studio', name=name,
                asset_id=1, role='studio', source='mapping')
            self.con.execute("INSERT INTO entity_alias VALUES(?,? ,?,'mapping',1)",
                             (entity_id, 'Shared', 'shared'))
        found = upsert_asset_entity(self.con, kind='studio', name='Shared',
            asset_id=3, role='studio', source='makers:studio')
        self.assertEqual(self.con.execute('SELECT canonical_name FROM entity WHERE id=?',
                                         (found,)).fetchone()[0], 'Shared')

    def test_metadata_landing_uses_one_studio_name_for_fields_and_relations(self):
        from peach.metadata_auto_apply import _apply_metadata_candidate
        self.con.row_factory = sqlite3.Row
        target = upsert_asset_entity(self.con, kind='studio', name='Attackers',
                                    asset_id=1, role='studio', source='r18:studio')
        group = {'asset_id': '3', 'asset_path': '/x/3.mp4', 'code': '',
                 'field': 'studio', 'item_key': 'asset:3:studio'}
        candidate = {'source': 'makers', 'value': 'ATTACKERS（アタッカーズ）',
                     'confidence': 1, 'candidate_key': 'studio-identity'}
        _apply_metadata_candidate(self.con, group, candidate, '2026-10-12', 'review:makers')
        self.assertEqual(self.con.execute('SELECT studio FROM asset WHERE id=3').fetchone()[0], 'Attackers')
        self.assertEqual(self.con.execute("SELECT entity_id FROM asset_entity WHERE asset_id=3 AND role='studio'").fetchone()[0], target)


class ZeroWidthNameTests(unittest.TestCase):
    """规范名里的零宽字符。

    上游译名夹带过两个实例：performer 的 `‌斋藤亚美里` 和 creator 的
    `比特ビット‌`。界面上和普通名字一模一样，问题全在比对上。
    """

    def test_a_zero_width_char_does_not_survive_into_the_canonical_name(self):
        self.assertEqual(canonicalize_entity_name("performer", "‌斋藤亚美里"),
                         "斋藤亚美里")
        self.assertEqual(canonicalize_entity_name("creator", "比特ビット‌"),
                         "比特ビット")

    def test_the_normalised_name_matches_across_the_invisible_difference(self):
        """`strip()` 不认零宽字符是空白，`normalized_name` 于是带着它。

        后果不是显示出错，是 `upsert_asset_entity` 按 `normalized_name` 找不到已有
        实体，同一个人存成两条；按名字搜也一个都搜不到。
        """
        self.assertEqual(normalize_entity_name("比特ビット‌"),
                         normalize_entity_name("比特ビット"))

    def test_the_joiner_inside_an_emoji_is_left_alone(self):
        """U+200D 在 emoji 序列里是连字符，剥掉会把一个字形拆成三个。

        创作者名字里带 emoji 是常事，所以它不在剥除名单里。
        """
        self.assertEqual(canonicalize_entity_name("creator", "👨‍👩‍👧"),
                         "👨‍👩‍👧")


if __name__ == "__main__":
    unittest.main()