tests/test_fc2_sellers.py
Rama master · réplica de lectura del repositorio original.
"""卖家证据、账号冲突、馆藏匹配及按批撤回的行为验证。"""
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from peach import auto_landing_revert, fc2_sellers
from peach.entities import upsert_asset_entity
from peach.entity_classification import write_claim
from peach.web_entity import q_entity, q_index
from peach.web_state import WebContract
from tests.support.ledger import fresh_ledger
class Fc2SellerTests(unittest.TestCase):
def setUp(self):
self.directory = tempfile.TemporaryDirectory()
self.addCleanup(self.directory.cleanup)
self.root = Path(self.directory.name).resolve()
self.db = fresh_ledger(self.root)
self.connection = sqlite3.connect(self.db)
self.connection.row_factory = sqlite3.Row
self.addCleanup(self.connection.close)
self.cache = self.root / 'cache'
self.cache.mkdir()
self.batch = fc2_sellers.SOURCE + '@test'
def asset(self, code):
row = self.connection.execute("INSERT INTO asset(location,path,name,medium,code,play_count,rating,creator) "
"VALUES('local',?,?,'video',?,8,4,'User Account')", (f'B:\\FC2\\{code}.mp4',f'{code}.mp4',code))
self.connection.commit()
return row.lastrowid
def snapshot(self, code='FC2-PPV-1234567', name='Seller', account='seller', provider='fc2ppvdb', **changes):
number = code.rsplit('-',1)[1]
page = f'https://fc2ppv-db.com/ja/videos/{number}' if provider == 'fc2ppvdb' else f'https://fc2cmadb.com/articles/{number}'
row = {'id':code, 'label':name, 'seller_url':f'https://adult.contents.fc2.com/users/{account}/', 'source_url':page, **changes}
(self.cache / f'{code}-{provider}.json').write_text(json.dumps(row),encoding='utf-8')
return row
def install(self):
with self.connection:
return fc2_sellers.install(self.connection,fc2_sellers.collect(self.connection,self.cache),batch=self.batch)
def undo(self):
with self.connection:
aliases = auto_landing_revert.planned_aliases(self.connection,fc2_sellers.SOURCE,self.batch)
refs = auto_landing_revert.planned_refs(self.connection,fc2_sellers.SOURCE,self.batch)
links = auto_landing_revert.planned_links(self.connection,fc2_sellers.SOURCE,self.batch)
sellers = fc2_sellers.planned_revert(self.connection,fc2_sellers.SOURCE,self.batch)
for row in aliases:
self.connection.execute('DELETE FROM entity_alias WHERE entity_id=? AND source=?',(row['entity_id'],row['source']))
for row in refs:
self.connection.execute('DELETE FROM entity_external_ref WHERE provider=? AND external_kind=? AND external_id=?',(row['provider'],row['kind'],row['id']))
for row in links:
self.connection.execute('DELETE FROM entity_link WHERE id=?',(row['id'],))
fc2_sellers.revert(self.connection,sellers)
def test_seller_account_and_split_files_share_work_without_changing_cast_or_history(self):
asset_id = self.asset('FC2-PPV-1234567-1')
self.asset('FC2-PPV-1234567-2')
performer = upsert_asset_entity(self.connection,kind='performer',name='Cast',asset_id=asset_id,role='performer',source='user:manual')
self.connection.commit()
before = [tuple(row) for row in self.connection.execute('SELECT * FROM asset ORDER BY id')]
cast = tuple(self.connection.execute('SELECT * FROM asset_entity WHERE entity_id=?',(performer,)).fetchone())
self.snapshot()
result = self.install()
self.assertEqual(len(result['entity_ids']),1)
contract = WebContract(self.db)
page = q_index(contract,'creators')
self.assertEqual(page['items'][0]['identity_labels'],['卖家'])
self.assertEqual(page['items'][0]['n'],1)
self.assertEqual(len(q_entity(contract,{'kind':'creator','name':'Seller'})['links']),1)
self.assertEqual([tuple(row) for row in self.connection.execute('SELECT * FROM asset ORDER BY id')],before)
self.assertEqual(tuple(self.connection.execute('SELECT * FROM asset_entity WHERE entity_id=?',(performer,)).fetchone()),cast)
self.assertEqual(self.connection.execute('PRAGMA foreign_key_check').fetchall(),[])
self.assertEqual(self.install()['changes'],0)
self.undo()
self.assertEqual(self.connection.execute("SELECT count(*) FROM entity WHERE kind='creator'").fetchone()[0],0)
self.assertEqual([tuple(row) for row in self.connection.execute('SELECT * FROM asset ORDER BY id')],before)
def test_conflicting_source_accounts_are_skipped_and_nonlibrary_records_are_ignored(self):
self.asset('FC2-PPV-1234567')
self.snapshot()
self.snapshot(provider='fc2cmadb',account='other')
self.snapshot(code='FC2-PPV-7654321')
plan = fc2_sellers.collect(self.connection,self.cache)
self.assertEqual(plan['accounts'],[])
self.assertEqual(plan['skipped'][0]['reason'],'作品来源给出不同卖家账号')
def test_reuse_by_account_preserves_user_name_metadata_reference_and_seller_claim(self):
one = self.asset('FC2-PPV-1234567')
two = self.asset('FC2-PPV-7654321')
entity_id = upsert_asset_entity(self.connection,kind='creator',name='User Name',asset_id=one,role='creator',
source='user:manual',external_provider='fc2',external_id='seller',metadata={'user':'kept'})
write_claim(self.connection,entity_id=entity_id,facet='account_role',value='seller',source='user:manual',status='approved',evidence='用户确认')
self.connection.commit()
entity = tuple(self.connection.execute('SELECT * FROM entity WHERE id=?',(entity_id,)).fetchone())
ref = tuple(self.connection.execute('SELECT * FROM entity_external_ref WHERE entity_id=?',(entity_id,)).fetchone())
self.snapshot(code='FC2-PPV-7654321',name='Public Name')
self.install()
self.assertEqual(tuple(self.connection.execute('SELECT * FROM entity WHERE id=?',(entity_id,)).fetchone()),entity)
self.assertEqual(tuple(self.connection.execute('SELECT * FROM entity_external_ref WHERE entity_id=?',(entity_id,)).fetchone()),ref)
self.assertEqual(self.connection.execute('SELECT alias FROM entity_alias WHERE entity_id=?',(entity_id,)).fetchone()[0],'Public Name')
self.undo()
self.assertEqual(self.connection.execute('SELECT count(*) FROM asset_entity WHERE entity_id=? AND asset_id=?',(entity_id,two)).fetchone()[0],0)
self.assertEqual(tuple(self.connection.execute('SELECT * FROM entity WHERE id=?',(entity_id,)).fetchone()),entity)
self.assertEqual(self.connection.execute('SELECT count(*) FROM entity_classification WHERE entity_id=?',(entity_id,)).fetchone()[0],1)
def test_same_name_different_account_and_invalid_public_evidence_do_not_land(self):
one = self.asset('FC2-PPV-1234567')
upsert_asset_entity(self.connection,kind='creator',name='Seller',asset_id=one,role='creator',source='user:manual',external_provider='fc2',external_id='other')
self.connection.commit()
row = self.snapshot()
self.assertEqual(fc2_sellers.collect(self.connection,self.cache)['accounts'],[])
for changes in ({'source_url':'https://fc2ppv-db.com/ja/videos/7654321'},
{'seller_url':'https://adult.contents.fc2.com.evil.test/users/seller/'},
{'seller_url':'https://adult.contents.fc2.com/users/seller/?token=private'},
{'label':'FC2-PPV'}):
self.assertIsNone(fc2_sellers.seller_record({**row,**changes},'fc2ppvdb'))
def test_new_accounts_with_shared_names_are_conflicts(self):
self.asset('FC2-PPV-1234567')
self.asset('FC2-PPV-7654321')
self.snapshot()
self.snapshot(code='FC2-PPV-7654321',account='other')
plan = fc2_sellers.collect(self.connection,self.cache)
self.assertEqual(plan['accounts'],[])
self.assertEqual(len(plan['skipped']),2)
self.assertTrue(all(row['reason']=='不同 FC2 账号使用相同卖家名称' for row in plan['skipped']))
def test_revert_preserves_later_user_account_name(self):
self.asset('FC2-PPV-1234567')
self.snapshot()
entity_id = self.install()['entity_ids'][0]
self.connection.execute("UPDATE entity SET canonical_name='User Seller',normalized_name='userseller' WHERE id=?",(entity_id,))
self.connection.commit()
self.undo()
self.assertEqual(self.connection.execute('SELECT canonical_name FROM entity WHERE id=?',(entity_id,)).fetchone()[0],'User Seller')
def test_rejected_seller_name_is_skipped_while_other_accounts_still_land(self):
rejected = self.asset('FC2-PPV-1234567')
self.asset('FC2-PPV-7654321')
self.connection.execute("INSERT INTO review_decision(category,item_key,status,note,updated_at) "
"VALUES('creator-attribution',?,'rejected','','2026-10-08T00:00:00+00:00')",
(f'{rejected}:seller',))
self.connection.commit()
self.snapshot()
self.snapshot(code='FC2-PPV-7654321',name='Other Seller',account='other')
result = self.install()
self.assertEqual(len(result['entity_ids']),1)
self.assertEqual([(row['code'],row['reason']) for row in result['skipped']],[('FC2-PPV-1234567','摄取判定不收这个卖家名')])
names = [row[0] for row in self.connection.execute("SELECT canonical_name FROM entity WHERE kind='creator'")]
self.assertEqual(names,['Other Seller'])
self.assertEqual(self.connection.execute("SELECT count(*) FROM entity_external_ref WHERE external_id='seller'").fetchone()[0],0)
def test_held_account_gets_no_alias_or_claim_from_a_work_that_ingest_rejects(self):
one = self.asset('FC2-PPV-1234567')
two = self.asset('FC2-PPV-7654321')
entity_id = upsert_asset_entity(self.connection,kind='creator',name='User Name',asset_id=one,role='creator',
source='user:manual',external_provider='fc2',external_id='seller')
self.connection.execute("INSERT INTO review_decision(category,item_key,status,note,updated_at) "
"VALUES('creator-attribution',?,'rejected','','2026-10-08T00:00:00+00:00')",
(f'{two}:{fc2_sellers.normalize_entity_name("User Name")}',))
self.connection.commit()
self.snapshot(code='FC2-PPV-7654321',name='Public Name')
result = self.install()
self.assertEqual(result['entity_ids'],[])
self.assertEqual([row['asset_id'] for row in result['skipped']],[two])
for table in ('entity_alias','entity_classification','entity_link'):
self.assertEqual(self.connection.execute(f'SELECT count(*) FROM {table} WHERE entity_id=?',(entity_id,)).fetchone()[0],0)
def test_collection_names_are_planned_against_the_account_that_ingest_picks(self):
one = self.asset('FC2-PPV-1234567')
two = self.asset('FC2-PPV-7654321')
held = upsert_asset_entity(self.connection,kind='creator',name='Studio',asset_id=one,role='creator',source='user:manual')
self.connection.commit()
self.snapshot(code='FC2-PPV-7654321',name='Studio 4K')
plan = fc2_sellers.collect(self.connection,self.cache)
self.assertEqual(plan['accounts'][0]['entity_id'],held)
self.install()
self.assertEqual(self.connection.execute("SELECT entity_id FROM asset_entity WHERE asset_id=? AND role='creator'",(two,)).fetchone()[0],held)
self.connection.execute("UPDATE entity_external_ref SET external_id='other' WHERE entity_id=?",(held,))
self.connection.commit()
(self.cache / 'FC2-PPV-7654321-fc2ppvdb.json').unlink()
self.snapshot(code='FC2-PPV-1234567',name='Studio 4K',account='third')
plan = fc2_sellers.collect(self.connection,self.cache)
self.assertEqual([row['reason'] for row in plan['skipped']],['同名对应不同 FC2 账号'])
def test_works_in_the_recycle_bin_get_no_seller(self):
asset_id = self.asset('FC2-PPV-1234567')
self.connection.execute("UPDATE asset SET disposal='trash' WHERE id=?",(asset_id,))
self.connection.commit()
self.snapshot()
self.assertEqual(fc2_sellers.collect(self.connection,self.cache)['accounts'],[])
def test_stale_work_plan_rolls_back_and_revert_keeps_later_user_metadata(self):
asset_id = self.asset('FC2-PPV-1234567')
self.snapshot()
plan = fc2_sellers.collect(self.connection,self.cache)
self.connection.execute("UPDATE asset SET code='ABW-001' WHERE id=?",(asset_id,))
self.connection.commit()
with self.assertRaisesRegex(ValueError,'已过期'):
with self.connection:
fc2_sellers.install(self.connection,plan,batch=self.batch)
self.connection.execute("UPDATE asset SET code='FC2-PPV-1234567' WHERE id=?",(asset_id,))
self.connection.commit()
entity_id = self.install()['entity_ids'][0]
self.connection.execute("UPDATE entity SET metadata_json=json_set(metadata_json,'$.user','kept') WHERE id=?",(entity_id,))
self.connection.commit()
self.undo()
self.assertEqual(json.loads(self.connection.execute('SELECT metadata_json FROM entity WHERE id=?',(entity_id,)).fetchone()[0])['user'],'kept')