tests/test_metadata_creator_attributions.py
Rama master · réplica de lectura del repositorio original.
"""创作者审计保留独立身份、保护人工判断，并可整批回滚。"""
import sqlite3
import tempfile
import unittest
import json
from pathlib import Path
from types import SimpleNamespace
from peach.classification import creator_collection_base, is_structural_creator
from peach.entities import (apply_directory_rejections, creator_directory_key, derived_directory_rejections,
merge_entity, normalize_entity_name, upsert_asset_entity)
from peach.field_owners import write_owned_fields
from peach.entity_classification import write_claim
from peach.metadata_creator_attributions import apply_plan, collect, restore
from tests.support.ledger import fresh_ledger
from scripts import audit_creator_attributions as audit
class CreatorAttributionTests(unittest.TestCase):
def setUp(self):
self.directory = tempfile.TemporaryDirectory()
self.addCleanup(self.directory.cleanup)
self.db = fresh_ledger(Path(self.directory.name).resolve())
self.connection = sqlite3.connect(self.db)
self.addCleanup(self.connection.close)
def asset(self, name, *, path=None, studio=None, code=None, source='legacy:asset', metadata=None):
cursor = self.connection.execute(
"INSERT INTO asset(location,path,name,medium,creator,studio,code) VALUES('local',?,?,'video',?,?,?)",
(path or f'B:\\创作者\\{name}\\clip.mp4', (path or 'clip.mp4').rsplit('\\', 1)[-1], name, studio, code))
asset_id = cursor.lastrowid
# 历史错误直接造在夹具里，摄取入口本身须拒绝发行平台。
held = self.connection.execute("SELECT id FROM entity WHERE kind='creator' AND canonical_name=?", (name,)).fetchone()
if held:
entity_id = held[0]
else:
entity_id = self.connection.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,created_at,updated_at) VALUES('creator',?,?,'t','t')",
(name, name.casefold())).lastrowid
self.connection.execute("INSERT INTO asset_entity(asset_id,entity_id,role,source,metadata_json) VALUES(?,?,'creator',?,?)",
(asset_id, entity_id, source, metadata or '{}'))
return asset_id, entity_id
def plan(self):
return [row for row in collect(self.connection) if row['action'] in {'remove','replace'}]
def test_rejected_account_attribution_blocks_directory_reingest_and_allows_manual_review(self):
asset_id, entity_id = self.asset('Parent Account')
self.connection.execute("DELETE FROM asset_entity WHERE asset_id=?", (asset_id,))
self.connection.execute("INSERT INTO review_decision(category,item_key,status,updated_at) "
"VALUES('creator-attribution',?,'rejected','t')", (f'{asset_id}:parent account',))
self.assertIsNone(upsert_asset_entity(self.connection, kind='creator',name='Parent Account',
asset_id=asset_id,role='creator',source='scan:directory'))
self.assertEqual(self.connection.execute('SELECT count(*) FROM asset_entity WHERE asset_id=?',(asset_id,)).fetchone()[0],0)
self.assertEqual(upsert_asset_entity(self.connection,kind='creator',name='Parent Account',
asset_id=asset_id,role='creator',source='user:manual'),entity_id)
def test_collection_quality_suffix_uses_the_rejected_canonical_account_decision(self):
asset_id, _ = self.asset('Parent Account')
self.connection.execute("INSERT INTO review_decision(category,item_key,status,updated_at) "
"VALUES('creator-attribution',?,'rejected','t')", (f'{asset_id}:parent account',))
self.assertIsNone(upsert_asset_entity(self.connection, kind='creator',name='Parent Account4K',
asset_id=asset_id,role='creator',source='scan:directory'))
def test_directory_rejection_covers_new_files_and_subdirectories(self):
self.connection.execute("INSERT INTO review_decision(category,item_key,status,updated_at) "
"VALUES('creator-attribution',?,'rejected','t')",
(creator_directory_key('Parent Account', r'B:\创作者\Parent Account\Other Channel'),))
blocked = [r'B:\创作者\Parent Account\Other Channel\new.mp4',
r'B:\创作者\Parent Account\other channel\图片\new.jpg']
for path in blocked:
asset_id, _ = self.asset('Parent Account', path=path)
with self.subTest(path=path):
self.assertIsNone(upsert_asset_entity(self.connection, kind='creator', name='Parent Account',
asset_id=asset_id, role='creator', source='scan:directory'))
self.assertIsNotNone(upsert_asset_entity(self.connection, kind='creator', name='Parent Account',
asset_id=asset_id, role='creator', source='user:manual'))
own_id, entity_id = self.asset('Parent Account', path=r'B:\创作者\Parent Account\own.mp4')
self.assertEqual(upsert_asset_entity(self.connection, kind='creator', name='Parent Account',
asset_id=own_id, role='creator', source='scan:directory'), entity_id)
def test_fully_rejected_subtrees_yield_topmost_directory_rejections(self):
root = r'B:\创作者\Parent Account'
rejected = [rf'{root}\Other Channel\a.mp4', rf'{root}\Other Channel\图片\b.jpg',
rf'{root}\Mixed\c.mp4', r'B:\卖家\Reseller\Parent Account\e.mp4']
for path in rejected:
asset_id, _ = self.asset('Parent Account', path=path)
self.connection.execute("INSERT INTO review_decision(category,item_key,status,note,updated_at) "
"VALUES('creator-attribution',?,'rejected','另一账号','t')", (f'{asset_id}:parent account',))
self.asset('Parent Account', path=rf'{root}\Mixed\own.mp4')
proposals = derived_directory_rejections(self.connection)
self.assertEqual([(row['directory'], row['subtree_assets']) for row in proposals],
[(rf'{root}\Other Channel', 2), (r'B:\卖家\Reseller\Parent Account', 1)])
self.assertEqual(apply_directory_rejections(self.connection, proposals, reviewer='script:test'), 2)
self.assertEqual(derived_directory_rejections(self.connection), [])
new_id, _ = self.asset('Elsewhere', path=rf'{root}\Other Channel\新\d.mp4')
self.assertIsNone(upsert_asset_entity(self.connection, kind='creator', name='Parent Account',
asset_id=new_id, role='creator', source='scan:directory'))
def test_collection_suffix_respects_rejection_without_a_canonical_account(self):
asset_id, _ = self.asset('Parent Account4K')
self.connection.execute("INSERT INTO review_decision(category,item_key,status,updated_at) "
"VALUES('creator-attribution',?,'rejected','t')", (f'{asset_id}:parent account',))
self.assertIsNone(upsert_asset_entity(self.connection, kind='creator', name='Parent Account4K',
asset_id=asset_id, role='creator', source='scan:directory'))
def test_merged_person_alias_ingests_into_the_preserved_artist(self):
asset_id, creator_id = self.asset('Former Name')
artist_id = upsert_asset_entity(self.connection,kind='performer',name='Current Name',
asset_id=asset_id,role='performer',source='release:cast',metadata={'official':'preserved'})
merge_entity(self.connection,target_id=artist_id,source_id=creator_id,source_name='Former Name',
alias_source='user:identity-merge:creator')
another_id, _ = self.asset('Independent Account')
found = upsert_asset_entity(self.connection,kind='creator',name='Former Name',
asset_id=another_id,role='creator',source='scan:directory',metadata={'other':'data'})
self.assertEqual(found,artist_id)
self.assertIsNone(self.connection.execute("SELECT id FROM entity WHERE kind='creator' AND canonical_name='Former Name'").fetchone())
self.assertEqual(self.connection.execute('SELECT role FROM asset_entity WHERE asset_id=? AND entity_id=?',
(another_id,artist_id)).fetchone(),('performer',))
self.assertEqual(json.loads(self.connection.execute('SELECT metadata_json FROM entity WHERE id=?',
(artist_id,)).fetchone()[0]),{'official':'preserved'})
def test_unmerged_artist_alias_keeps_an_independent_account(self):
asset_id, _ = self.asset('Independent Account')
artist_id = upsert_asset_entity(self.connection,kind='performer',name='Current Name',
asset_id=asset_id,role='performer',source='release:cast')
self.connection.execute("INSERT INTO entity_alias VALUES(?, 'Shared Alias','shared alias','source:directory',1)",(artist_id,))
found = upsert_asset_entity(self.connection,kind='creator',name='Shared Alias',
asset_id=asset_id,role='creator',source='scan:directory')
self.assertNotEqual(found,artist_id)
def test_content_month_quality_and_repost_directories_are_not_accounts(self):
names = ['zj','11月','AI修复','黑丝','美腿足交','7sht.me','98T.la202202092146']
for name in names:
asset_id, _ = self.asset(name)
self.assertIsNone(upsert_asset_entity(self.connection,kind='creator',name=name,
asset_id=asset_id,role='creator',source='legacy:asset'))
self.asset('Santa'); self.asset('banbi_555'); self.asset('alice.example.com')
self.assertCountEqual([row['current_creator'] for row in self.plan()],names)
self.asset('zj',path=r'B:\手动归属\zj\clip.mp4',source='user:manual')
self.assertCountEqual([row['current_creator'] for row in self.plan()],names[1:])
self.assertEqual({row['relation_source']:row['action'] for row in collect(self.connection)
if row['current_creator']=='zj'}, {'legacy:asset':'review', 'user:manual':'keep'})
def test_directory_names_built_from_genre_vocabulary_are_not_accounts(self):
for name in ('黑丝美腿', 'jk足交合集', '极品美腿天花板', '合集-巨乳 多创作者', '美腿 1080p',
'2023年5月', '4K', '맨발_모음', '검스A맨발B_풋잡', '&网红套图（走光）'):
with self.subTest(name=name):
self.assertTrue(is_structural_creator(name))
for name in ('Alice', '海生', 'jk小美', 'fc2美臀OLさくら', 'Lena Anderson', 'analove', '黑丝小仙女'):
with self.subTest(name=name):
self.assertFalse(is_structural_creator(name))
def test_collection_annotations_reduce_to_the_account_name(self):
for name in ('someone 合集', 'someone合辑', 'someone全集', 'someone最新12v', 'someone 30v 40g',
'someone V12 80GB', 'someone(2)', 'someone 去重版 4K', 'someone 120GB', 'someone 1080p'):
with self.subTest(name=name):
self.assertEqual(creator_collection_base(name), 'someone')
for name in ('someone 2509', 'someone', 'v12', '合集'):
with self.subTest(name=name):
self.assertEqual(creator_collection_base(name), name)
def test_month_and_quality_collections_reuse_accounts_and_restore_all_relations(self):
_, target = self.asset('muchi_tina',source='user:manual')
ids = [self.asset(name)[0] for name in ['muchi_tina2025.4','muchi_tina10月','muchi_tina2025年一月']]
self.connection.commit()
frozen = self.plan()
self.assertEqual([row['proposed_creator'] for row in frozen],['muchi_tina']*3)
with self.connection:
receipt = apply_plan(self.connection,frozen)
for asset_id in ids:
self.assertEqual(self.connection.execute('SELECT creator FROM asset WHERE id=?',(asset_id,)).fetchone()[0],'muchi_tina')
self.assertEqual(self.connection.execute("SELECT entity_id FROM asset_entity WHERE asset_id=? AND role='creator'",(asset_id,)).fetchone()[0],target)
with self.connection:
restore(self.connection,receipt)
self.assertEqual(len(self.plan()),3)
self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(),[])
def test_versioned_release_directories_and_publisher_numbers_are_excluded(self):
samples = {
'WAAA-415_6K-C':r'B:\云下载\WAAA-415_6K-C\@Milan@ty999.me_WAAA-415_6K-C.mp4',
'wavr00178pl':r'B:\云下载\wavr00178pl\wavr00178.part1.mp4',
'fellatiojapan-090-':r'B:\创作者\fellatiojapan-090-\fellatiojapan-090-AyumuNakahori-1080p\090-AyumuNakahori-1080p.mp4',
'fellatiojapan-324':r'B:\云下载\fellatiojapan-324\2048社区 - big2048.com@fellatiojapan-324.mp4',
}
for name,path in samples.items():
asset_id,_ = self.asset(name,path=path)
self.assertIsNone(upsert_asset_entity(self.connection,kind='creator',name=name,
asset_id=asset_id,role='creator',source='scan:directory'))
self.asset('banbi_555',path=r'B:\创作者\banbi_555\scene title.mp4')
self.asset('ABW-987',path=r'B:\创作者\ABW-987\another title.mp4')
self.assertCountEqual([row['current_creator'] for row in self.plan()],samples)
def test_release_directory_does_not_assign_its_code_to_unrelated_files(self):
examples = [('WAAA-415_6K-C', 'HMN-597.mp4'),
('WAAA-415_6K-C', 'daily life.mp4'),
('gachincoppv-1009-HD', 'gachincoppv-1009-HD1.wmv'),
('gachincoppv-1009-HD', 'gachincoppv-1009-HD3.wmv')]
ids = []
for name, filename in examples:
asset_id, _ = self.asset(name, path=f'B:\\云下载\\{name}\\{filename}')
ids.append(asset_id)
self.assertIsNone(upsert_asset_entity(self.connection,kind='creator',name=name,
asset_id=asset_id,role='creator',source='scan:directory'))
self.asset('banbi_555', path=r'B:\创作者\banbi_555\holiday.mp4')
frozen = self.plan()
self.assertCountEqual([row['asset_id'] for row in frozen], ids)
with self.connection:
receipt = apply_plan(self.connection, frozen)
self.assertEqual(self.connection.execute('SELECT count(*) FROM asset WHERE creator IS NULL AND code IS NULL').fetchone()[0], 4)
self.assertEqual(self.connection.execute('SELECT count(*) FROM asset').fetchone()[0], 5)
with self.connection:
restore(self.connection, receipt)
self.assertCountEqual([row['asset_id'] for row in self.plan()], ids)
def test_tokyo_publisher_requires_registered_studio_and_matching_release_file(self):
asset_id,_ = self.asset('Tokyo',studio='东京热',code='n1042',path=r'B:\云下载\TokyoHot-n1042.mp4')
studio_id = upsert_asset_entity(self.connection,kind='studio',name='东京热',
asset_id=asset_id,role='studio',source='release:studio')
self.connection.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source) VALUES(?,'Tokyo Hot','tokyo hot','release')",(studio_id,))
self.asset('Tokyo',studio='东京热',code='n1042',path=r'B:\创作者\Tokyo\unrelated.mp4')
self.assertEqual([row['asset_id'] for row in self.plan()],[asset_id])
def test_source_confirmed_publisher_lands_as_studio_and_is_reversible(self):
asset_id, entity_id = self.asset('Published Brand')
write_claim(self.connection,entity_id=entity_id,facet='account_role',value='studio',
source='source:official',source_url='https://publisher.test/',evidence='发行目录将出演者与厂牌分列',
status='observed',confidence=1)
self.connection.commit()
self.assertIsNone(upsert_asset_entity(self.connection,kind='creator',name='Published Brand',
asset_id=asset_id,role='creator',source='scan:directory'))
self.assertIsNone(upsert_asset_entity(self.connection,kind='creator',name='Published Brand2025.05',
asset_id=asset_id,role='creator',source='scan:directory'))
with self.connection:
receipt = apply_plan(self.connection,self.plan())
self.assertEqual(self.connection.execute('SELECT creator,studio FROM asset WHERE id=?',(asset_id,)).fetchone(),(None,'Published Brand'))
self.assertEqual(len(receipt['created_entities']),1)
with self.connection:
restore(self.connection,receipt)
self.assertEqual(self.connection.execute('SELECT creator,studio FROM asset WHERE id=?',(asset_id,)).fetchone(),('Published Brand',None))
self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(),[])
def test_confirmed_studio_and_aliases_are_excluded_from_automatic_creator_ingest(self):
asset_id, entity_id = self.asset('Published Brand')
write_claim(self.connection,entity_id=entity_id,facet='account_role',value='studio',
source='source:official',source_url='https://publisher.test/',evidence='发行厂牌官网',
status='observed',confidence=1)
self.connection.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source) VALUES(?,'Brand.test','brand.test','release')",(entity_id,))
receipt = apply_plan(self.connection,self.plan())
studio_id = receipt['created_entities'][0]['id']
merge_entity(self.connection,target_id=studio_id,source_id=entity_id,
source_name='Published Brand',alias_source='user:identity-merge:creator')
for name in ('Published Brand','Published Brand2025.05','Brand.test','Brand.test2025.05'):
with self.subTest(name=name):
self.assertIsNone(upsert_asset_entity(self.connection,kind='creator',name=name,
asset_id=asset_id,role='creator',source='scan:directory'))
self.assertEqual(self.connection.execute("SELECT count(*) FROM entity WHERE kind='creator'").fetchone()[0],0)
self.assertEqual(self.connection.execute('SELECT entity_id,role FROM asset_entity WHERE asset_id=?',(asset_id,)).fetchall(),[(studio_id,'studio')])
self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(),[])
manual_id = upsert_asset_entity(self.connection,kind='creator',name='Published Brand',
asset_id=asset_id,role='creator',source='user:manual')
self.assertIsNotNone(manual_id)
def test_unconfirmed_studio_claim_does_not_block_creator_ingest(self):
asset_id, _ = self.asset('Alice')
studio_id = upsert_asset_entity(self.connection,kind='studio',name='Brand',
asset_id=asset_id,role='studio',source='source:catalog')
write_claim(self.connection,entity_id=studio_id,facet='account_role',value='studio',
source='source:search',evidence='待核验的同名检索',status='candidate',confidence=0.5)
self.assertIsNotNone(upsert_asset_entity(self.connection,kind='creator',name='Brand',
asset_id=asset_id,role='creator',source='scan:directory'))
def test_created_studio_with_subsequent_alias_refuses_atomic_restore(self):
asset_id, entity_id = self.asset('Published Brand')
write_claim(self.connection,entity_id=entity_id,facet='account_role',value='studio',
source='source:official',source_url='https://publisher.test/',evidence='发行厂牌官网',
status='observed',confidence=1)
self.connection.commit()
with self.connection:
receipt = apply_plan(self.connection,self.plan())
studio_id = receipt['created_entities'][0]['id']
self.connection.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source) VALUES(?,'Brand','brand','user:manual')",(studio_id,))
self.connection.commit()
with self.assertRaises(ValueError):
with self.connection:
restore(self.connection,receipt)
self.assertEqual(self.connection.execute('SELECT creator,studio FROM asset WHERE id=?',(asset_id,)).fetchone(),(None,'Published Brand'))
self.assertEqual(self.connection.execute('SELECT count(*) FROM entity_alias WHERE entity_id=?',(studio_id,)).fetchone()[0],1)
def test_collection_ingest_reuses_existing_account_and_preserves_profile(self):
_, target = self.asset('muchi_tina',source='user:manual')
self.connection.execute("UPDATE entity SET metadata_json='{\"profile\":\"kept\"}' WHERE id=?",(target,))
asset_id,_ = self.asset('other')
returned = upsert_asset_entity(self.connection,kind='creator',name='muchi_tina2025.05',
asset_id=asset_id,role='creator',source='scan:directory')
self.assertEqual(returned,target)
self.assertEqual(self.connection.execute('SELECT metadata_json FROM entity WHERE id=?',(target,)).fetchone()[0],'{"profile":"kept"}')
def test_platform_is_not_ingested_as_creator(self):
asset_id, _ = self.asset('RealSeller')
entity_id = upsert_asset_entity(self.connection, kind='creator', name='Myfans',
asset_id=asset_id, role='creator', source='user:manual')
self.assertIsNone(entity_id)
self.assertEqual(self.connection.execute("SELECT count(*) FROM entity WHERE canonical_name='Myfans'").fetchone()[0], 0)
def test_audit_covers_all_sources_and_preserves_accounts_and_manual_identity(self):
self.asset('Myfans')
self.asset('Alice', path='https://example.test/users/alice')
self.asset('ConfirmedSeller', source='user:watermark')
self.asset('banbi_555', path=r'B:\创作者\banbi_555\a title.mp4')
self.asset('FC2-PPV', source='user:manual')
rows = collect(self.connection)
self.assertEqual(len(rows), 5)
self.assertEqual([row['current_creator'] for row in rows if row['action']=='remove'], ['Myfans'])
self.assertEqual({row['verdict'] for row in rows}, {'platform','online_account','source_identity','identifier_candidate'})
def test_studio_aliases_and_dated_scene_titles_have_file_evidence(self):
asset_id, _ = self.asset('Das', studio='ダスッ！')
studio_id = upsert_asset_entity(self.connection, kind='studio', name='ダスッ！',
asset_id=asset_id, role='studio', source='release:studio')
self.connection.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source) VALUES(?,'Das','das','release')", (studio_id,))
self.asset('Freya Mayer Ready to Serve You', path=r'B:\西方\DarkRoomVR.21.04.02.Freya.Mayer.Ready.to.Serve.You.XXX\movie.mp4', studio='DarkRoomVR')
self.asset('Jane Doe', path=r'B:\西方\Studio.21.04.02.Jane.Doe.XXX\movie.mp4')
self.asset('DPMI-058_000', path=r'B:\云下载\DPMI-058_000\dpmi00058hhb_000.mp4')
self.asset('ABW-123', path=r'B:\云下载\ABW-123\ABW-123.mp4', code='ABW-123')
self.assertEqual({row['current_creator']: row['verdict'] for row in self.plan()}, {
'Das':'studio', 'Freya Mayer Ready to Serve You':'release_title',
'DPMI-058_000':'release_identifier', 'ABW-123':'release_identifier'})
def test_repair_and_restore_preserve_identity_metadata_and_personal_records(self):
asset_id, entity_id = self.asset('Myfans', metadata='{"evidence":"folder"}')
self.connection.execute('UPDATE asset SET play_count=9,rating=4 WHERE id=?', (asset_id,))
self.connection.commit()
plan = self.plan()
with self.connection:
manifest = apply_plan(self.connection, plan)
self.assertEqual(len(manifest['relations']), 1)
self.assertEqual(self.connection.execute('SELECT creator,play_count,rating FROM asset WHERE id=?', (asset_id,)).fetchone(), (None,9,4))
self.assertEqual(self.connection.execute('SELECT canonical_name FROM entity WHERE id=?', (entity_id,)).fetchone(), ('Myfans',))
with self.connection:
restore(self.connection, manifest)
self.assertEqual(self.connection.execute('SELECT creator,play_count,rating FROM asset WHERE id=?', (asset_id,)).fetchone(), ('Myfans',9,4))
self.assertEqual(self.connection.execute('SELECT metadata_json FROM asset_entity WHERE asset_id=?', (asset_id,)).fetchone(), ('{"evidence":"folder"}',))
self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(), [])
def test_stale_plan_and_later_edits_reject_whole_batch(self):
asset_id, _ = self.asset('Myfans')
other_id, _ = self.asset('FC2-PPV')
self.connection.commit()
plan = self.plan()
write_owned_fields(self.connection, [asset_id], {'creator':'Seller'}, 'user:manual')
self.connection.commit()
with self.assertRaisesRegex(ValueError, '计划'):
with self.connection:
apply_plan(self.connection, plan)
self.assertEqual(self.connection.execute('SELECT count(*) FROM asset_entity').fetchone()[0], 2)
with self.connection:
manifest = apply_plan(self.connection, self.plan())
write_owned_fields(self.connection, [other_id], {'creator':'VerifiedSeller'}, 'user:manual')
self.connection.commit()
with self.assertRaisesRegex(ValueError, '后续改动'):
with self.connection:
restore(self.connection, manifest)
self.assertEqual(self.connection.execute('SELECT creator FROM asset WHERE id=?', (other_id,)).fetchone()[0], 'VerifiedSeller')
def test_protected_and_independent_identity_are_reviewed(self):
asset_id, entity_id = self.asset('Myfans')
write_owned_fields(self.connection, [asset_id], {'creator':'Myfans'}, 'review:identity')
self.asset('FC2-PPV')
self.connection.execute("INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id) VALUES(?, 'site','creator','seller-12')", (entity_id,))
self.assertEqual([row['current_creator'] for row in self.plan()], ['FC2-PPV'])
def test_publisher_moves_to_existing_studio_and_restores_as_a_batch(self):
studio_asset, _ = self.asset('Alice')
studio_id = upsert_asset_entity(self.connection, kind='studio', name='Vixen',
asset_id=studio_asset, role='studio', source='release:studio', metadata={'official':'preserved'})
studio_before = self.connection.execute('SELECT metadata_json,updated_at FROM entity WHERE id=?', (studio_id,)).fetchone()
asset_id, creator_id = self.asset('Vixen', path=r'B:\云下载\A Scene [Vixen.com]\scene.mp4')
self.connection.commit()
with self.connection:
manifest = apply_plan(self.connection, self.plan())
self.assertEqual(self.connection.execute('SELECT creator,studio FROM asset WHERE id=?', (asset_id,)).fetchone(), (None,'Vixen'))
self.assertEqual(self.connection.execute('SELECT entity_id,role FROM asset_entity WHERE asset_id=?', (asset_id,)).fetchall(), [(studio_id,'studio')])
self.assertEqual(self.connection.execute('SELECT metadata_json,updated_at FROM entity WHERE id=?', (studio_id,)).fetchone(), studio_before)
with self.connection:
restore(self.connection, manifest)
self.assertEqual(self.connection.execute('SELECT creator,studio FROM asset WHERE id=?', (asset_id,)).fetchone(), ('Vixen',None))
self.assertEqual(self.connection.execute('SELECT entity_id,role FROM asset_entity WHERE asset_id=?', (asset_id,)).fetchall(), [(creator_id,'creator')])
def test_site_post_identity_requires_matching_file_and_preserves_accounts(self):
self.asset('[7sht.me]legsjapan-955', path=r'B:\云下载\[7sht.me]legsjapan-955\legsjapan-955.mp4')
self.asset('fellatiojapan-312', path=r'B:\云下载\fellatiojapan-312\fellatiojapan-312.mp4')
self.asset('legsjapan-999', path=r'B:\创作者\legsjapan-999\a title.mp4')
self.assertEqual({row['current_creator'] for row in self.plan()}, {'[7sht.me]legsjapan-955','fellatiojapan-312'})
def test_restore_refuses_a_creator_relation_added_without_flat_field_changes(self):
asset_id, _ = self.asset('Myfans')
self.connection.commit()
with self.connection:
manifest = apply_plan(self.connection, self.plan())
upsert_asset_entity(self.connection, kind='creator', name='NewSeller', asset_id=asset_id,
role='creator', source='user:manual')
self.connection.commit()
with self.assertRaisesRegex(ValueError, '关系已发生后续改动'):
with self.connection:
restore(self.connection, manifest)
def performers(self, *names):
for name in names:
self.connection.execute("INSERT INTO entity(kind,canonical_name,normalized_name,created_at,updated_at) "
"VALUES('performer',?,?,'t','t')", (name, normalize_entity_name(name)))
def shaped(self, account, filenames, source='legacy:asset'):
for filename in filenames:
self.asset(account, path=f'B:\\创作者\\{account}\\{filename}', source=source)
return {row['verdict'] for row in collect(self.connection) if row['current_creator'] == account}
def test_series_naming_with_rotating_cast_is_a_review_candidate_not_a_repair(self):
self.performers('八ッ橋さい子', '本多由奈', '北川ゆず')
self.assertEqual(self.shaped('COSH こすっち', ['こすっち001 八ッ橋さい子.mp4', 'こすっち002 八ッ橋さい子.mp4',
'こすっち003 本多由奈.mp4', 'こすっち005 北川ゆず.mp4']), {'label_shape'})
uralesbian = ['Akari Asagiri & Eri Hirasawa - 120 - Sensual Kiss.mp4',
'Shino Aoi & Mai Miori - 108 - Double Masturbation.mp4', 'ChieKobayashi-ShinoAoi-039-1080p.mp4']
self.assertEqual(self.shaped('UraLesbian', uralesbian, source='user:manual-organize'), {'label_shape'})
self.assertEqual(self.plan(), [])
def test_series_naming_does_not_hold_back_a_verified_studio_repair(self):
_, entity_id = self.asset('Label Works', path='B:\\创作者\\Label Works\\Ann Lee & Mia Rose - 01 - Kiss.mp4')
write_claim(self.connection, entity_id=entity_id, facet='account_role', value='studio', source='source:publisher',
source_url='https://label.test/', evidence='官网', status='observed', confidence=1)
verdicts = self.shaped('Label Works', ['Ann Lee & Leo King - 02 - Pool.mp4', 'Sam Fox & Mia Rose - 03 - Day.mp4'])
self.assertEqual(verdicts, {'verified_studio'})
def test_couple_accounts_titles_and_stray_files_keep_their_account_verdicts(self):
self.performers('Mia Rose', 'Leo King', 'Ann Lee')
couple = ['Mia Rose & Leo King - 01 - Beach.mp4', 'Mia Rose & Ann Lee - 02 - Pool.mp4',
'Mia Rose & Leo King - 03 - Night.mp4', 'Mia Rose & Sam Fox - 04 - Day.mp4']
titles = ['Account 012 Beach Day.mp4', 'Account 013 Hot Tub.mp4', 'Account 014 Night Out.mp4']
stray = ['MariHirose-RiaKurumi-488-1080p.mp4', 'clip1.mp4', 'clip2.mp4', 'clip3.mp4']
for account, names in [('Couple', couple), ('Account', titles), ('RiaKurumi', stray)]:
self.assertNotIn('label_shape', self.shaped(account, names))
def test_cli_preserves_existing_foreign_key_violations_and_keeps_a_restore_manifest(self):
_, entity_id = self.asset('Myfans')
self.connection.execute("INSERT INTO asset_entity(asset_id,entity_id,role,source) VALUES(99999,?,'tag','legacy')", (entity_id,))
self.connection.commit()
baseline = self.connection.execute('PRAGMA foreign_key_check').fetchall()
manifest = self.db.parent / 'manifest.json'
args = SimpleNamespace(db=self.db, apply=True, backup=self.db.parent/'before.db', restore=None, manifest=manifest)
audit._write_ledger(args, {'rows':self.plan()})
payload = json.loads(manifest.read_text(encoding='utf-8'))
self.assertEqual(payload['status'], 'committed')
self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(), baseline)
args.backup = self.db.parent / 'repaired.db'
args.restore = manifest
args.manifest = None
audit._write_ledger(args, payload)
self.assertEqual(self.connection.execute('SELECT creator FROM asset').fetchone()[0], 'Myfans')
self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(), baseline)
if __name__ == '__main__':
unittest.main()