"""浏览分类、来源保护、身份衔接与分页的行为回归。"""
import sqlite3
import tempfile
import unittest
from unittest import mock

from peach import entity_classification as classification, entity_identity_research as research
from peach.entities import merge_entity, upsert_asset_entity
from peach.web_entity import q_index, q_entity, q_suggest
from peach.web_catalog import q_items
from peach.web_state import WebContract
from peach.web_catalog import attach_avatar_availability
from peach.performer_alias_followup import Missing, Unavailable
from peach.sources.base import Page
from tests.support.ledger import fresh_ledger

SOUGOU = research.WIKI_ROOTS['sougouwiki'] + 'd/'


def _work_table(*cast):
    rows = ''.join(f'<tr><td>COSH-00{index}</td><td>こすっち00{index}</td>'
                   f'<td><a href="{SOUGOU}%A4%A2{index}">{name}</a></td></tr>' for index, name in enumerate(cast, 1))
    return ('<meta charset="utf-8"><div id="page-body"><div class="user-area"><table>'
            '<tr><th>NO</th><th>TITLE</th><th>ACTRESS</th></tr>' + rows + '</table></div></div>').encode()


class _Pages:
    """站上只有 `listed` 里的页；`failing` 里的地址模拟网络失败。"""

    def __init__(self, listed=None, failing=()):
        self.listed, self.failing, self.asked = listed or {}, set(failing), []

    def get(self, url):
        self.asked.append(url)
        if url in self.failing:
            raise Unavailable('Wiki 网络请求未取得')
        if url not in self.listed:
            raise Missing('站上没有这一页')
        return Page(url, self.listed[url])


class EntityClassificationTests(unittest.TestCase):
    def setUp(self):
        self.directory=tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.db=fresh_ledger(self.directory.name)
        self.connection=sqlite3.connect(self.db)
        self.addCleanup(self.connection.close)
        self.connection.row_factory=sqlite3.Row

    def entity(self,name,kind='creator'):
        self.connection.execute("INSERT INTO asset(location,path,name,medium,creator,studio) VALUES('local',?,?, 'video',?,'Publisher')",
            (f'B:\\西方\\Publisher\\Publisher.24.12.02.{name.replace(" ",".")}.mp4',f'{name}.mp4',name))
        asset_id=self.connection.execute('SELECT last_insert_rowid()').fetchone()[0]
        entity_id=upsert_asset_entity(self.connection,kind=kind,name=name,asset_id=asset_id,role=kind,source='legacy:asset')
        self.connection.commit()
        return asset_id,entity_id

    def test_account_work_portraits_without_person_identity_are_marked_as_stand_ins(self):
        asset_id,entity_id=self.entity('Publisher Account')
        self.connection.execute("UPDATE asset SET snapshot_path='sheet.jpg' WHERE id=?",(asset_id,))
        self.connection.commit()
        contract=WebContract(self.db)
        with mock.patch.object(WebContract,'has_avatar',return_value=True):
            row={'entity_id':entity_id,'rep':asset_id,'has_image':True}
            attach_avatar_availability(contract,[row])
            self.assertTrue(row['has_avatar'])
            self.assertTrue(row['avatar_stand_in'])
            self.assertTrue(row['has_image'])
            self.claim(entity_id,'identity','person')
            self.connection.commit()
            attach_avatar_availability(contract,[row])
            self.assertTrue(row['has_avatar'])
            self.assertNotIn('avatar_stand_in',row)

    def claim(self,entity_id,facet,value,**kwargs):
        classification.write_claim(self.connection,entity_id=entity_id,facet=facet,value=value,
            source='source:publisher',source_url='https://publisher.test/profile',evidence='发行方资料',status='observed',confidence=1,**kwargs)

    def finding(self,name,entity_id):
        return {'entity_id':entity_id,'name':name,'cast_studios':['Publisher'],
            'cast_source_url':'https://publisher.test/profile', 'claims':[
                {'facet':'identity','value':'person','source_url':'https://publisher.test/profile','evidence':'发行方署名','status':'observed','confidence':1},
                {'facet':'occupation','value':'adult_performer','source_url':'https://publisher.test/profile','evidence':'发行出演名单','status':'observed','confidence':1}]}

    def test_multiple_markets_and_occupations_are_independent_of_entity_role(self):
        _,entity_id=self.entity('Known Person',kind='performer')
        for facet,value in [('identity','person'),('occupation','adult_performer'),('occupation','model'),('market','japanese_av'),('market','western_adult')]:
            self.claim(entity_id,facet,value)
        self.connection.commit()
        data=q_entity(WebContract(self.db),{'kind':'performer','name':'Known Person'})
        self.assertEqual(data['identity_labels'],['女优','西方'])
        self.assertEqual(len(classification.classifications(self.connection,[entity_id])[entity_id]),5)
        self.assertNotIn('classifications',data)

    def test_candidates_and_search_hits_do_not_qualify_for_trusted_filters(self):
        _,known=self.entity('Known Person'); _,unknown=self.entity('Unknown Person')
        self.claim(known,'account_role','seller')
        classification.write_claim(self.connection,entity_id=unknown,facet='account_role',value='seller',source='script:lookup',evidence='同名搜索命中')
        self.connection.commit()
        data=q_index(WebContract(self.db),'performers',limit=1,category='blogger')
        self.assertEqual(data['items'],[])
        self.assertFalse(data['has_more'])
        self.assertEqual(data['categories'],{})
        unclassified=q_index(WebContract(self.db),'performers')
        self.assertEqual([row['entity_id'] for row in unclassified['items']],[unknown])
        self.assertEqual(unclassified['items'][0]['identity_labels'],[])
        self.assertEqual(q_index(WebContract(self.db),'creators',category='seller')['items'][0]['entity_id'],known)
        self.assertEqual(q_index(WebContract(self.db),'creators',category="person' OR 1=1")['items'],[])

    def test_personal_accounts_publishers_and_animation_have_distinct_categories(self):
        _,person=self.entity('Personal Account')
        self.claim(person,'identity','person')
        _,publisher=self.entity('Publishing Account')
        self.claim(publisher,'identity','account')
        self.claim(publisher,'account_role','publisher')
        _,animation=self.entity('Animation Account')
        self.claim(animation,'identity','account')
        self.claim(animation,'account_role','publisher')
        self.claim(animation,'occupation','animator')
        _,collection=self.entity('Resource Collection')
        self.claim(collection,'identity','collection')
        self.claim(collection,'occupation','content_creator')
        self.connection.commit()
        contract=WebContract(self.db)
        self.assertCountEqual([r['entity_id'] for r in q_index(contract,'performers',category='blogger')['items']],
                              [person,publisher])
        self.assertEqual(q_entity(contract,{'kind':'creator','name':'Resource Collection'})['identity_labels'],[])
        self.assertNotIn(collection,[r['entity_id'] for r in q_index(contract,'performers')['items']])
        self.assertEqual(q_index(contract,'performers',category='animation')['items'][0]['entity_id'],animation)

    def test_existing_jav_directory_identity_is_shared_by_profiles_and_paginated_filters(self):
        expected = []
        for provider in sorted(classification.JAV_DIRECTORIES):
            _, entity_id = self.entity(provider, kind='performer')
            self.connection.execute('INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id) '
                                    'VALUES(?,?,?,?)', (entity_id,provider,'performer',str(entity_id)))
            expected.append(entity_id)
        self.connection.commit()
        before = self.connection.total_changes
        for provider in sorted(classification.JAV_DIRECTORIES):
            profile = q_entity(WebContract(self.db),{'kind':'performer','name':provider})
            self.assertEqual(profile['identity_labels'],['女优'])
            self.assertEqual(classification.classifications(self.connection,[profile['id']])[profile['id']],[])
        found = []
        for offset in range(len(expected)):
            page = q_index(WebContract(self.db),'performers',category='japanese_av',limit=1,offset=offset)
            found.extend(row['entity_id'] for row in page['items'])
            self.assertEqual(page['categories'],{'japanese_av':len(expected)})
        self.assertCountEqual(found,expected)
        self.assertEqual(self.connection.total_changes,before)

    def test_collection_refs_general_directories_and_candidates_do_not_assert_jav_identity(self):
        for provider, external_kind in [('minnano-av','production'),('babepedia','performer'),
                                        ('stash','performer'),('r18','performer_name'),('kmib','performer')]:
            _, entity_id = self.entity(provider, kind='performer')
            self.connection.execute('INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id) '
                                    'VALUES(?,?,?,?)', (entity_id,provider,external_kind,str(entity_id)))
            classification.write_claim(self.connection,entity_id=entity_id,facet='market',value='japanese_av',
                                       source='script:search',evidence='未核验同名搜索')
        self.connection.commit()
        data = q_index(WebContract(self.db),'performers')
        self.assertEqual(len(data['items']),5)
        self.assertTrue(all(row['identity_labels']==[] for row in data['items']))
        self.assertEqual(q_index(WebContract(self.db),'performers',category='japanese_av')['items'],[])

    def test_category_counts_respect_search_and_visible_works_without_being_limited_to_one_page(self):
        _, seller = self.entity('Seller Match')
        self.claim(seller,'account_role','seller')
        self.entity('Creator Match'); self.entity('Outside')
        asset_id, hidden = self.entity('Vanished Match')
        self.connection.execute("UPDATE asset SET disposal='vanished' WHERE id=?",(asset_id,))
        self.connection.commit()
        data = q_index(WebContract(self.db),'creators',q='Match',limit=1,category='seller')
        self.assertEqual(data['categories'],{'seller':1})
        self.assertEqual([row['entity_id'] for row in data['items']],[seller])

    def test_fc2_release_does_not_determine_performer_occupation(self):
        asset_id, entity_id = self.entity('Known Person',kind='performer')
        self.claim(entity_id,'market','japanese_av')
        self.connection.execute("UPDATE asset SET code='FC2-PPV-1234567',region='west' WHERE id=?",(asset_id,))
        self.connection.commit()
        for category in ('japanese_av','western'):
            page = q_index(WebContract(self.db),'performers',category=category,limit=1)
            self.assertEqual(page['categories'],{'japanese_av':1,'western':1})
            self.assertEqual(page['items'][0]['identity_labels'],['女优','西方'])
            self.assertEqual(page['items'][0]['identity_categories'],['japanese_av','western'])
        self.assertEqual(q_index(WebContract(self.db),'performers',category='amateur')['items'],[])
        self.connection.execute("UPDATE asset SET disposal='vanished' WHERE id=?",(asset_id,))
        self.connection.commit()
        self.assertEqual(classification.summaries(self.connection,[entity_id])[entity_id]['identity_labels'],['女优'])

    def test_amateur_scene_tag_does_not_classify_a_jav_performer_as_amateur(self):
        asset_id, entity_id = self.entity('JAV Performer',kind='performer')
        self.claim(entity_id,'market','japanese_av')
        self.connection.execute("UPDATE asset SET code='SIRO-5333' WHERE id=?",(asset_id,))
        upsert_asset_entity(self.connection,kind='tag',name='素人',asset_id=asset_id,
                            role='tag',source='test:scene-tag')
        self.connection.commit()
        contract=WebContract(self.db)
        profile=q_entity(contract,{'kind':'performer','name':'JAV Performer'})
        self.assertEqual(profile['identity_labels'],['女优'])
        self.assertEqual(q_index(contract,'performers',category='amateur')['items'],[])
        self.assertEqual(q_index(contract,'performers')['categories'],{'japanese_av':1})

    def work(self,name,code=None,*,tag=None,disposal=None):
        """给已有出演者再挂一部作品。"""
        asset_id=self.connection.execute("INSERT INTO asset(location,path,name,medium,code,disposal) VALUES('local',?,?,'video',?,?)",
            (f'B:\\作品\\{name}\\{code or "clip"}-{self.connection.execute("SELECT count(*) FROM asset").fetchone()[0]}.mp4',
             'clip.mp4',code,disposal)).lastrowid
        upsert_asset_entity(self.connection,kind='performer',name=name,asset_id=asset_id,role='performer',source='legacy:asset')
        if tag:
            upsert_asset_entity(self.connection,kind='tag',name=tag,asset_id=asset_id,role='tag',source='test:scene-tag')
        return asset_id

    def test_amateur_identity_follows_trusted_claims_or_all_amateur_releases(self):
        _, confirmed = self.entity('Nonprofessional Cast',kind='performer')
        self.claim(confirmed,'occupation','amateur_performer')
        fc2_id, fc2_only = self.entity('FC2 Cast',kind='performer')
        self.connection.execute("UPDATE asset SET code='FC2-PPV-1234567' WHERE id=?",(fc2_id,))
        self.work('FC2 Cast',tag='素人')
        self.work('FC2 Cast',disposal='vanished')
        mixed_id, _ = self.entity('Mixed Cast',kind='performer')
        self.connection.execute("UPDATE asset SET code='FC2-PPV-7654321' WHERE id=?",(mixed_id,))
        self.work('Mixed Cast','ABP-001')
        _, guessed = self.entity('Guessed Cast',kind='performer')
        classification.write_claim(self.connection,entity_id=guessed,facet='occupation',
            value='amateur_performer',source='script:search',evidence='未核验推断')
        jav_id, jav = self.entity('Known JAV Cast',kind='performer')
        self.connection.execute("UPDATE asset SET code='FC2-PPV-1111111' WHERE id=?",(jav_id,))
        self.claim(jav,'market','japanese_av')
        self.connection.commit()
        contract=WebContract(self.db)
        page=q_index(contract,'performers',category='amateur')
        self.assertEqual(sorted(item['entity_id'] for item in page['items']),sorted([confirmed,fc2_only]))
        self.assertEqual(page['categories']['amateur'],2)
        self.assertEqual(q_entity(contract,{'kind':'performer','name':'Mixed Cast'})['identity_labels'],[])
        self.assertEqual(q_entity(contract,{'kind':'performer','name':'Guessed Cast'})['identity_labels'],[])
        self.assertEqual(q_entity(contract,{'kind':'performer','name':'Known JAV Cast'})['identity_labels'],['女优'])
        self.claim(confirmed,'occupation','adult_performer')
        self.claim(fc2_only,'market','western_adult')
        self.connection.commit()
        self.assertEqual(q_index(contract,'performers',category='amateur')['items'],[])

    def test_amateur_tags_from_the_vision_model_do_not_classify_a_performer(self):
        asset_id, styled = self.entity('Styled Cast',kind='performer')
        self.connection.execute("INSERT INTO asset_tag(asset_id,tag,confidence,source) VALUES(?,'素人',0.6,'vision_creator')",(asset_id,))
        for source in ('vision_creator','vision_creator_review'):
            upsert_asset_entity(self.connection,kind='tag',name='素人',asset_id=asset_id,role='tag',source=source)
        self.connection.commit()
        self.assertEqual(classification.summaries(self.connection,[styled])[styled]['identity_categories'],[])
        upsert_asset_entity(self.connection,kind='tag',name='素人',asset_id=asset_id,role='tag',source='name')
        self.connection.commit()
        self.assertEqual(classification.summaries(self.connection,[styled])[styled]['identity_categories'],['amateur'])

    def test_animation_author_appears_in_artist_directory_with_its_own_category(self):
        _, animator = self.entity('Animator')
        self.claim(animator,'occupation','animator')
        _, artist = self.entity('Artist')
        self.claim(artist,'occupation','artist')
        self.connection.commit()
        page = q_index(WebContract(self.db),'performers',category='animation')
        self.assertEqual([row['entity_id'] for row in page['items']],[animator])
        self.assertEqual(page['categories'],{'animation':1})
        self.assertEqual(q_index(WebContract(self.db),'creators')['items'],[])

    def test_real_accounts_share_artist_directory_and_keep_avatars_and_profile_routes(self):
        _, actor = self.entity('Actor', kind='performer')
        _, blogger = self.entity('Blogger')
        _, seller = self.entity('Seller')
        _, animator = self.entity('Animator')
        self.claim(seller,'account_role','seller')
        self.claim(animator,'occupation','animator')
        self.connection.commit()
        artists = q_index(WebContract(self.db),'performers')
        self.assertCountEqual([row['entity_id'] for row in artists['items']],[actor,blogger,animator])
        self.assertEqual({row['entity_id']:row['entity_kind'] for row in artists['items']},{actor:'performer',blogger:'creator',animator:'creator'})
        separate = q_index(WebContract(self.db),'creators')
        self.assertCountEqual([row['entity_id'] for row in separate['items']],[seller])

    def test_person_account_can_belong_to_blogger_and_seller_directories(self):
        _, person = self.entity('Person Account')
        self.claim(person,'identity','person')
        self.claim(person,'account_role','seller')
        self.connection.commit()
        contract = WebContract(self.db)
        self.assertEqual(q_index(contract,'performers')['items'][0]['entity_id'],person)
        self.assertEqual(q_index(contract,'creators')['items'][0]['entity_id'],person)
        self.assertEqual(q_entity(contract,{'kind':'creator','name':'Person Account'})['identity_labels'],['网黄博主','卖家'])

    def test_western_cast_occupation_does_not_assert_a_japanese_av_career(self):
        _, entity_id = self.entity('Western Performer',kind='performer')
        self.claim(entity_id,'occupation','adult_performer')
        self.claim(entity_id,'market','western_adult')
        self.connection.commit()
        profile = q_entity(WebContract(self.db),{'kind':'performer','name':'Western Performer'})
        self.assertEqual(profile['identity_labels'],['西方'])
        self.assertEqual(q_index(WebContract(self.db),'performers',category='japanese_av')['items'],[])

    def test_script_cannot_approve_or_supply_unvalidated_values(self):
        _,entity_id=self.entity('Known Person')
        for patch in [{'status':'approved'}, {'status':'observed','source_url':'http://example.test'}]:
            with self.assertRaises(ValueError):
                classification.write_claim(self.connection,entity_id=entity_id,facet='identity',value='person',source='script:lookup',evidence='推断',**patch)
        with self.assertRaises(ValueError):
            self.claim(entity_id,'occupation','unknown_profession')
        with self.assertRaisesRegex(ValueError,'代码判据'):
            classification.write_claim(self.connection,entity_id=entity_id,facet='identity',value='person',
                source='script:identity-research',source_url='https://publisher.test/profile',
                evidence='研究清单判断',status='observed',confidence=1)

    def test_only_code_sources_make_observed_claims_trusted(self):
        _,research=self.entity('Research Person',kind='performer')
        _,parsed=self.entity('Parsed Person',kind='performer')
        self.connection.execute("INSERT INTO entity_classification VALUES(?,'market','japanese_av',"
            "'script:identity-research','https://publisher.test/a','研究清单判断','observed',1,'t')",(research,))
        self.claim(parsed,'market','japanese_av')
        self.connection.commit()
        page=q_index(WebContract(self.db),'performers',category='japanese_av')
        self.assertEqual([row['entity_id'] for row in page['items']],[parsed])
        self.assertEqual(page['categories'],{'japanese_av':1})

    def test_research_observations_downgrade_to_candidates_and_code_facts_stay(self):
        _,research=self.entity('Research Person',kind='performer')
        _,parsed=self.entity('Parsed Person',kind='performer')
        self.connection.execute("INSERT INTO entity_classification VALUES(?,'market','japanese_av',"
            "'script:identity-research','https://publisher.test/a','研究清单判断','observed',0.8,'t')",(research,))
        self.connection.execute("INSERT INTO entity_identity_link VALUES(?,?,'same_person',"
            "'script:identity-research','https://publisher.test/a','研究清单判断','observed','t')",(research,parsed))
        self.claim(parsed,'market','japanese_av')
        plan=classification.untrusted_observed(self.connection)
        self.assertEqual([(row['entity_id'],row['source']) for row in plan['entity_classification']],
                         [(research,'script:identity-research')])
        self.assertEqual(len(plan['entity_identity_link']),1)
        self.assertEqual(classification.downgrade_untrusted_observed(self.connection),
                         {'entity_classification':1,'entity_identity_link':1})
        rows=self.connection.execute('SELECT entity_id,status,confidence,evidence FROM entity_classification ORDER BY entity_id').fetchall()
        self.assertEqual([tuple(row) for row in rows],
                         [(research,'candidate',0.8,'研究清单判断'),(parsed,'observed',1.0,'发行方资料')])
        self.assertEqual(classification.untrusted_observed(self.connection),
                         {'entity_classification':[],'entity_identity_link':[]})

    def test_name_wiki_work_list_through_the_research_list_lands_as_candidate_release(self):
        _,entity_id=self.entity('COSH こすっち')
        classification.write_claim(self.connection,entity_id=entity_id,facet='identity',value='unknown',
                                   source=research.SOURCE,evidence='公开身份来源未取得')
        keys=research.lookup_keys('COSH こすっち',['こすっち'])
        self.assertEqual(keys,['COSH こすっち','こすっち'])
        self.assertEqual(research.lookup_keys('secret_japan'),['secret_japan'])
        listed={SOUGOU+'%A4%B3%A4%B9%A4%C3%A4%C1':_work_table('八ッ橋さい子','本多由奈','北川ゆず')}
        finding=research.wiki_finding(entity_id,'COSH こすっち',keys,{'sougouwiki':_Pages(listed),'av_neme':_Pages()})
        with self.connection:
            research.apply(self.connection,research.plan(self.connection,[finding]))
        claims={(row['value'],row['status']) for row in self.connection.execute(
            "SELECT value,status FROM entity_classification WHERE entity_id=? AND facet='identity'",(entity_id,))}
        self.assertEqual([claim['status'] for claim in finding['claims']],['observed','rejected'])
        self.assertEqual(claims,{('release','candidate'),('unknown','rejected')})

    def test_name_wiki_lookup_records_pages_asked_and_leaves_failed_fetches_unfetched(self):
        missing=research.wiki_finding(7,'lucky',['lucky'],{'sougouwiki':_Pages(),'av_neme':_Pages()})
        self.assertEqual([(claim['value'],claim['status']) for claim in missing['claims']],[('unknown','candidate')])
        self.assertIn('sougouwiki「lucky」（页不存在）',missing['claims'][0]['evidence'])
        self.assertIn('av_neme「lucky」（页不存在）',missing['claims'][0]['evidence'])
        broken=_Pages(failing=[SOUGOU+'lucky'])
        self.assertIsNone(research.wiki_finding(7,'lucky',['lucky'],{'sougouwiki':broken,'av_neme':_Pages()}))
        few=_Pages({SOUGOU+'lucky':_work_table('八ッ橋さい子','本多由奈')})
        finding=research.wiki_finding(7,'lucky',['lucky'],{'sougouwiki':few,'av_neme':_Pages()})
        self.assertIn('sougouwiki「lucky」（不是作品一览）',finding['claims'][0]['evidence'])

    def test_cast_role_repair_and_restore_preserve_all_business_fields(self):
        asset_id,entity_id=self.entity('Known Person')
        self.connection.execute('UPDATE asset SET play_count=7,rating=4 WHERE id=?',(asset_id,));self.connection.commit()
        before=dict(self.connection.execute('SELECT * FROM asset WHERE id=?',(asset_id,)).fetchone())
        frozen=research.plan(self.connection,[self.finding('Known Person',entity_id)])
        self.assertEqual(len(frozen[0]['cast']),1)
        with self.connection:
            receipt=research.apply(self.connection,frozen)
        self.assertEqual(self.connection.execute('SELECT role FROM asset_entity WHERE asset_id=?',(asset_id,)).fetchone()[0],'performer')
        self.assertEqual(self.connection.execute('SELECT group_concat(DISTINCT status) FROM entity_classification').fetchone()[0],'candidate')
        self.assertEqual(self.connection.execute('SELECT group_concat(status) FROM entity_identity_link').fetchone()[0],'candidate')
        self.assertEqual(classification.related_identities(self.connection,entity_id),[])
        with self.connection:
            research.restore(self.connection,receipt)
        after=dict(self.connection.execute('SELECT * FROM asset WHERE id=?',(asset_id,)).fetchone())
        before.pop('mutation_revision');after.pop('mutation_revision')
        self.assertEqual(before,after)
        self.assertEqual(self.connection.execute('SELECT count(*) FROM entity').fetchone()[0],1)
        self.assertEqual(self.connection.execute('SELECT count(*) FROM entity_identity_link').fetchone()[0],0)
        self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(),[])

    def test_stale_plan_and_history_changes_block_repair_or_restore(self):
        asset_id,entity_id=self.entity('Known Person')
        frozen=research.plan(self.connection,[self.finding('Known Person',entity_id)])
        self.connection.execute('UPDATE asset SET creator=? WHERE id=?',('User choice',asset_id));self.connection.commit()
        with self.assertRaisesRegex(ValueError,'已过期'):
            with self.connection: research.apply(self.connection,frozen)
        frozen=research.plan(self.connection,[self.finding('Known Person',entity_id)])
        with self.connection: receipt=research.apply(self.connection,frozen)
        self.connection.execute('UPDATE asset SET play_count=9 WHERE id=?',(asset_id,));self.connection.commit()
        with self.assertRaisesRegex(ValueError,'后续改动'):
            with self.connection: research.restore(self.connection,receipt)

    def test_manual_creator_relation_is_not_reclassified_as_cast(self):
        asset_id,entity_id=self.entity('Known Person')
        self.connection.execute("UPDATE asset_entity SET source='user:manual' WHERE asset_id=?",(asset_id,));self.connection.commit()
        self.assertEqual(research.plan(self.connection,[self.finding('Known Person',entity_id)])[0]['cast'],[])

    def test_entity_merge_transfers_classification_with_foreign_keys_disabled(self):
        _,target=self.entity('Known Person');_,source=self.entity('Old Name')
        self.claim(source,'occupation','model');self.connection.commit()
        with self.connection:
            merge_entity(self.connection,target_id=target,source_id=source,source_name='Old Name',alias_source='user:review')
        self.assertEqual(classification.classifications(self.connection,[target])[target][0]['label'],'模特')
        self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(),[])

    def part(self, number, code='FC2-PPV-3312576', disposal=None):
        name=f'{code}-{number}.mp4'
        cursor=self.connection.execute(
            "INSERT INTO asset(location,path,name,medium,code,size,duration,disposal) "
            "VALUES('local',?,?,'video',?,100,600,?)", (f'B:\\番号\\{code}\\{name}',name,code,disposal))
        for kind, label in [('performer','Cast Person'),('performer','Costar'),('studio','FC2 Studio'),('tag','Test Tag')]:
            upsert_asset_entity(self.connection,kind=kind,name=label,asset_id=cursor.lastrowid,role=kind,source='test:cast')
        self.connection.commit()
        return cursor.lastrowid

    def test_merge_deduplicates_identical_source_facts_and_keeps_latest_check_time(self):
        _, target = self.entity('Target'); _, source = self.entity('Source')
        for entity_id, checked in ((source, '2026-10-07T01:00:00+00:00'),
                                   (target, '2026-10-06T01:00:00+00:00')):
            classification.write_claim(self.connection, entity_id=entity_id, facet='identity',
                value='person', source='source:official', evidence='同一份署名',
                source_url='https://publisher.test/person', status='observed', confidence=1, checked_at=checked)
        self.connection.commit()
        with self.connection:
            merge_entity(self.connection, target_id=target, source_id=source,
                         source_name='Source', alias_source='user:review')
        rows = classification.classifications(self.connection, [target])[target]
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]['checked_at'], '2026-10-07T01:00:00+00:00')

    def test_merge_rejects_different_facts_from_the_same_source(self):
        _, target = self.entity('Target'); _, source = self.entity('Source')
        for entity_id, evidence in ((source, '记录甲'), (target, '记录乙')):
            classification.write_claim(self.connection, entity_id=entity_id, facet='identity',
                value='person', source='source:official', evidence=evidence, status='observed', confidence=1,
                source_url='https://publisher.test/person')
        self.connection.commit()
        with self.assertRaisesRegex(ValueError, '来源冲突'), self.connection:
            merge_entity(self.connection, target_id=target, source_id=source,
                         source_name='Source', alias_source='user:review')
        self.assertEqual(self.connection.execute('SELECT count(*) FROM entity WHERE id IN (?,?)',
                                                (target, source)).fetchone()[0], 2)

    def test_fc2_collection_counts_once_in_profiles_rosters_tags_and_search(self):
        for number in range(1,20): self.part(number)
        contract=WebContract(self.db)
        profile=q_entity(contract,{'kind':'performer','name':'Cast Person'})
        self.assertEqual(profile['asset_count'],1)
        self.assertEqual(profile['related_performers'][0]['n'],1)
        self.assertEqual(profile['tags'][0]['n'],1)
        for kind in ('performers','studios','tags'):
            self.assertEqual(q_index(contract,kind)['items'][0]['n'],1)
        suggest=q_suggest(contract,'Cast Person',kinds=['performer'])
        self.assertEqual(suggest['groups'][0]['items'][0]['n'],1)
        page=q_items(contract,{'performer':'Cast Person','limit':'5'})
        self.assertEqual(page['work_total'],1)
        self.assertEqual(page['total'],19)
        self.assertEqual(len(page['items']),5)
        self.assertEqual(page['items'][0]['part_group']['count'],19)
        self.assertEqual(self.connection.execute('SELECT count(*) FROM asset').fetchone()[0],19)

    def test_fc2_work_counts_keep_visibility_and_invalidate_with_ledger_changes(self):
        self.part(1);self.part(3,disposal='trash');self.part(4,disposal='vanished')
        contract=WebContract(self.db)
        self.assertEqual(q_entity(contract,{'kind':'performer','name':'Cast Person'})['asset_count'],1)
        self.part(1,code='FC2-PPV-7777777')
        self.assertEqual(q_entity(contract,{'kind':'performer','name':'Cast Person'})['asset_count'],2)
        self.assertEqual(q_items(contract,{'performer':'Cast Person'})['work_total'],2)
        self.assertEqual(self.connection.execute('SELECT count(*) FROM asset_entity').fetchone()[0],16)

    def test_ambiguous_fc2_encodes_are_counted_as_individual_files(self):
        self.part(1);self.part(2)
        self.part('1-1080p')
        profile=q_entity(WebContract(self.db),{'kind':'performer','name':'Cast Person'})
        self.assertEqual(profile['asset_count'],3)
        self.assertEqual(q_items(WebContract(self.db),{'performer':'Cast Person'})['work_total'],3)

    def test_cross_role_merge_routes_only_explicitly_marked_aliases(self):
        _, target=self.entity('Known Person',kind='performer')
        _, source=self.entity('Old Account Name')
        with self.connection:
            merge_entity(self.connection,target_id=target,source_id=source,
                         source_name='Old Account Name',alias_source='user:identity-merge:creator')
        self.assertEqual(q_entity(WebContract(self.db),{'kind':'creator','name':'Old Account Name'}),
                         {'redirect':{'id':target,'kind':'performer','name':'Known Person'}})
        self.assertEqual(q_entity(WebContract(self.db),{'kind':'studio','name':'Known Person'}),{'error':'not found'})
        self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(),[])

    def test_matching_names_without_a_merge_are_distinct_identities(self):
        self.entity('Known Person',kind='performer')
        self.assertEqual(q_entity(WebContract(self.db),{'kind':'creator','name':'Known Person'}),{'error':'not found'})