tests/test_company_profiles.py
Rama master · réplica de lectura del repositorio original.
"""公司事实的来源、日期精度、填空与撤回。"""
import json
import sqlite3
import tempfile
import unittest
from contextlib import closing
from pathlib import Path
from types import SimpleNamespace
from peach import company_profiles
from peach.entities import upsert_asset_entity
from peach import web_contract
from support.ledger import fresh_ledger
class CompanyProfileTests(unittest.TestCase):
def test_dates_keep_precision_and_reject_copyright_or_invalid_dates(self):
for raw, expected in [('2002年5月', '2002-05'), ('1999年3月1日', '1999-03-01'), ('2016', '2016'), ('平成16年7月21日', '2004-07-21')]:
self.assertEqual(company_profiles.date_text(raw), expected)
for raw in ['© 2002', '20周年', '2025年2月30日', '2002年（ブランド発売）']:
self.assertIsNone(company_profiles.date_text(raw))
def test_official_tables_separate_company_and_brand_and_detect_conflicts(self):
html = '<table><tr><th>設立</th><td>2002年5月</td></tr><tr><th>ブランド設立</th><td>2003年</td></tr><tr><th>主要取引先</th><td>WILL</td></tr></table><dl><dt>所在地</dt><dd>東京都中野区</dd></dl><a href="/company/">会社概要 Company</a>'
parsed = company_profiles.extract(html, 'https://brand.example/')
self.assertEqual(parsed['facts']['founded']['value'], '2002-05')
self.assertEqual(parsed['facts']['launched']['value'], '2003')
self.assertNotIn('parent', parsed['facts'])
self.assertNotIn('country', parsed['facts'])
self.assertEqual(parsed['company_pages'], ['https://brand.example/company/'])
conflict = company_profiles.extract(html + '<dl><dt>設立</dt><dd>2001</dd></dl>', 'https://brand.example/')
self.assertNotIn('founded', conflict['facts'])
self.assertEqual(conflict['conflicts'][0]['field'], 'founded')
def test_fill_protects_existing_facts_and_revert_preserves_metadata(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
connection = sqlite3.connect(fresh_ledger(Path(temporary.name).resolve()))
self.addCleanup(connection.close)
connection.execute("INSERT INTO entity(id,kind,canonical_name,normalized_name,metadata_json,created_at,updated_at) VALUES(1,'studio','Brand','brand','{\"provider\":\"catalog\"}','2026-10-07','2026-10-07')")
fact = {'value': '2002', 'source_url': 'https://brand.example/company', 'evidence': '設立：2002年'}
self.assertEqual(company_profiles.fill(connection, 1, {'founded': fact}, source='auto:company-profile', batch='batch1')['written'], ['founded'])
conflict = company_profiles.fill(connection, 1, {'founded': {**fact, 'value': '2001'}}, source='auto:company-profile', batch='batch2')
self.assertEqual(conflict, {'written': [], 'conflicts': ['founded']})
self.assertEqual(company_profiles.planned_revert(connection, 'auto:company-profile', 'batch2'), [])
rows = company_profiles.planned_revert(connection, 'auto:company-profile', 'batch1')
self.assertEqual(company_profiles.revert(connection, rows), 1)
self.assertEqual(json.loads(connection.execute('SELECT metadata_json FROM entity').fetchone()[0]), {'provider': 'catalog'})
def test_group_company_tables_are_scoped_by_the_entity_names(self):
html = '<table><tr><th>社名</th><td>ソフト・オン・デマンド株式会社</td></tr><tr><th>設立</th><td>1995年11月</td></tr></table><table><tr><th>社名</th><td>SODクリエイト株式会社</td></tr><tr><th>設立</th><td>1999年3月</td></tr></table>'
result = company_profiles.extract_for_entity(html, 'https://group.example/about_us/info/', ['SOD Create', 'SODクリエイト'])
self.assertEqual(result['facts']['founded']['value'], '1999-03')
self.assertEqual(result['facts']['legal_name']['value'], 'SODクリエイト株式会社')
unnamed = company_profiles.extract_for_entity(html, 'https://group.example/about_us/info/', ['SOD Create'])
self.assertEqual(unnamed['facts'], {})
def test_terms_operator_comes_from_the_page_sentence_and_listings_give_no_group(self):
html = '<h1>Vixen Media Group</h1><p>OUR BRANDS Each studio is a world. Explore Vixen, Blacked, Blacked Raw, Tushy, and Deeper, all available on Vixen Plus</p>'
self.assertEqual(company_profiles.page_facts(html, 'https://www.vixengroup.com/', ['Tushy'])['facts'], {})
terms = '<p>These terms of service are entered into between you and General Media Systems, LLC (the Company).</p>'
facts = company_profiles.page_facts(terms, 'https://www.anybrand.example/terms', ['Any'])['facts']
self.assertEqual(facts['operator']['value'], 'General Media Systems, LLC')
self.assertEqual(facts['operator']['evidence'], 'These terms of service are entered into between you and General Media Systems, LLC (')
self.assertEqual(company_profiles.page_facts(terms, 'https://www.anybrand.example/news', ['Any'])['facts'], {})
def test_group_statements_without_labels_are_not_observed(self):
cases = [
('<p>SODグループ</p><p>SODクリエイト株式会社</p>', 'https://corporate.sod.co.jp/business/softondemand/', ['SOD Create']),
('<h2>メーカー一覧</h2><p>FALENO GROUP</p><p>素人CLOVER PROFILE</p>', 'https://falenogroup.com/makers/', ['素人CLOVER']),
('<dl><dt>2006</dt><dd>組織拡大の為、ティーパワーズ株式会社を設立</dd></dl>', 'https://www.t-powers.co.jp/company/', ['T-POWERS']),
]
for html, url, names in cases:
with self.subTest(url=url):
self.assertEqual(company_profiles.page_facts(html, url, names)['facts'], {})
def test_replay_reads_saved_pages_and_drops_cross_page_conflicts(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
root = Path(temporary.name)
(root / 'a.html').write_text('<table><tr><th>設立</th><td>2002年5月</td></tr><tr><th>所在地</th><td>東京都</td></tr></table>', encoding='utf-8')
(root / 'b.html').write_text('<dl><dt>設立</dt><dd>2001年</dd></dl>', encoding='utf-8')
(root / 'c.html').write_text('<dl><dt>商号</dt><dd>株式会社ブランド</dd></dl>', encoding='utf-8')
pages = [{'url': 'https://brand.example/', 'final_url': 'https://www.brand.example/company', 'http_status': 200, 'cache': str(root / 'a.html')},
{'url': 'https://brand.example/about', 'http_status': 404},
{'url': 'https://brand.example/x', 'final_url': 'https://brand.example/x', 'http_status': 200, 'cache': str(root / 'b.html')},
{'url': 'https://registry.example/brand', 'final_url': 'https://registry.example/brand', 'http_status': 200, 'cache': str(root / 'c.html')},
{'url': 'https://brand.example/y', 'final_url': 'https://brand.example/y', 'cache': str(root / 'c.html')}]
result = company_profiles.replay(pages, ['Brand'], {'brand.example'})
self.assertEqual(list(result['facts']), ['location'])
self.assertEqual(result['facts']['location']['source_url'], 'https://www.brand.example/company')
def test_observed_fact_replaces_a_candidate_and_revert_lists_candidates(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
connection = sqlite3.connect(fresh_ledger(Path(temporary.name).resolve()))
self.addCleanup(connection.close)
held = {'company_profile': {
'group': {'value': 'Group', 'source_url': 'https://brand.example/', 'status': 'candidate', 'source': 'auto:company-profile', 'batch': 'old'},
'location': {'value': '東京都', 'source_url': 'https://brand.example/', 'status': 'candidate', 'source': 'auto:company-profile', 'batch': 'old'}}}
connection.execute("INSERT INTO entity(id,kind,canonical_name,normalized_name,metadata_json,created_at,updated_at) VALUES(1,'studio','Brand','brand',?,'2026-10-07','2026-10-07')",
(json.dumps(held),))
self.assertEqual(company_profiles.planned_revert(connection, 'auto:company-profile', 'old'),
[{'entity_id': 1, 'entity': 'Brand', 'fields': ['group', 'location']}])
fact = {'value': '大阪府', 'source_url': 'https://brand.example/company', 'evidence': '所在地：大阪府', 'status': 'observed'}
self.assertEqual(company_profiles.fill(connection, 1, {'location': fact}, source='auto:company-profile', batch='new')['written'], ['location'])
profile = json.loads(connection.execute('SELECT metadata_json FROM entity').fetchone()[0])['company_profile']
self.assertEqual(profile['location']['value'], '大阪府')
self.assertEqual(list(company_profiles.public_profile({'company_profile': profile})), ['location'])
def test_landing_writes_only_replayed_facts_and_ignores_listed_aliases_and_links(self):
from scripts import harvest_company_profiles as harvest
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
root = Path(temporary.name).resolve()
path = fresh_ledger(root)
with sqlite3.connect(path) as connection:
connection.execute("INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at) VALUES(1,'studio','Brand','brand','2026-10-07','2026-10-07')")
connection.execute("INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,created_at,updated_at) VALUES(1,'official','Brand','https://brand.example/','brand.example','2026-10-07','2026-10-07')")
connection.close()
(root / 'page.html').write_text('<table><tr><th>設立</th><td>2002年5月</td></tr></table>', encoding='utf-8')
listed = {'value': 'Brand Group', 'source_url': 'https://brand.example/', 'evidence': 'Brand Group', 'status': 'observed'}
plan = {'entities': [{'entity_id': 1, 'name': 'Brand', 'status': '已核查',
'facts': {'founded': {**listed, 'value': '2002-05'}, 'group': listed},
'pages': [{'url': 'https://brand.example/company', 'final_url': 'https://brand.example/company', 'http_status': 200, 'cache': str(root / 'page.html')}],
'aliases': [{'value': 'ブランド', 'source_url': 'https://brand.example/', 'status': 'observed'}],
'links': [{'entity_id': 1, 'link_kind': 'catalog', 'label': 'Brand', 'url': 'https://catalog.example/brand', 'status': 'observed'}]}]}
(root / 'plan.json').write_text(json.dumps(plan, ensure_ascii=False), encoding='utf-8')
args = SimpleNamespace(input=root / 'plan.json', db=path, apply=True, backup=root / 'backup.db',
batch='auto:company-profile@test', output=root / 'receipt.json')
self.assertEqual(harvest.land(args), 0)
with closing(sqlite3.connect(path)) as connection:
profile = json.loads(connection.execute('SELECT metadata_json FROM entity WHERE id=1').fetchone()[0])['company_profile']
self.assertEqual(list(profile), ['founded'])
self.assertEqual(profile['founded']['evidence'], '設立：2002年5月')
self.assertEqual(connection.execute('SELECT count(*) FROM entity_alias').fetchone()[0], 0)
self.assertEqual(connection.execute('SELECT url FROM entity_link').fetchall(), [('https://brand.example/',)])
change = json.loads((root / 'receipt.json').read_text(encoding='utf-8'))['changes'][0]
self.assertEqual(change['not_replayed'], ['group'])
self.assertEqual(change['ignored'], {'aliases': 1, 'links': 1})
def test_repair_demotes_unreplayable_fields_and_drops_batch_aliases_and_links(self):
from scripts import repair_company_profile_landing as repair
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
root = Path(temporary.name).resolve()
path = fresh_ledger(root)
batch = 'auto:company-profile@old'
stamp = {'source': 'auto:company-profile', 'batch': batch, 'status': 'observed', 'source_url': 'https://brand.example/company'}
profile = {'founded': {**stamp, 'value': '2002-05', 'evidence': '設立：2002年5月'},
'group': {**stamp, 'value': 'Brand Group', 'evidence': 'Brand Group：Brand'},
'location': {'source': 'auto:company-profile', 'batch': 'other', 'status': 'observed', 'value': '東京都',
'source_url': 'https://brand.example/company'}}
with closing(sqlite3.connect(path)) as connection, connection:
connection.execute("INSERT INTO entity(id,kind,canonical_name,normalized_name,metadata_json,created_at,updated_at) VALUES(1,'studio','Brand','brand',?,'2026-10-07','2026-10-07')",
(json.dumps({'company_profile': profile}, ensure_ascii=False),))
connection.execute("INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,created_at,updated_at) VALUES(1,'official','Brand','https://brand.example/','brand.example','2026-10-07','2026-10-07')")
connection.execute("INSERT INTO entity_link(entity_id,link_kind,label,url,hostname,metadata_json,created_at,updated_at) VALUES(1,'catalog','Brand','https://catalog.example/brand','catalog.example',?,'2026-10-07','2026-10-07')",
(json.dumps({'source': 'auto:company-profile', 'batch': batch}),))
connection.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source) VALUES(1,'ブランド','ブランド',?)", (batch,))
(root / 'page.html').write_text('<table><tr><th>設立</th><td>2002年5月</td></tr></table>', encoding='utf-8')
plan = {'entities': [{'entity_id': 1, 'pages': [{'url': 'https://brand.example/company', 'final_url': 'https://brand.example/company',
'http_status': 200, 'cache': str(root / 'page.html')}]}]}
(root / 'plan.json').write_text(json.dumps(plan), encoding='utf-8')
common = ['--db', str(path), '--batch', batch, '--plan', str(root / 'plan.json'), '--output', str(root / 'receipt.json')]
self.assertEqual(repair.main(common), 0)
with closing(sqlite3.connect(path)) as connection:
self.assertEqual(connection.execute('SELECT count(*) FROM entity_alias').fetchone()[0], 1)
self.assertEqual(repair.main([*common, '--apply', '--backup', str(root / 'backup.db')]), 0)
with closing(sqlite3.connect(path)) as connection:
held = json.loads(connection.execute('SELECT metadata_json FROM entity').fetchone()[0])['company_profile']
self.assertEqual({key: fact['status'] for key, fact in held.items()},
{'founded': 'observed', 'group': 'candidate', 'location': 'observed'})
self.assertEqual(connection.execute('SELECT count(*) FROM entity_alias').fetchone()[0], 0)
self.assertEqual(connection.execute('SELECT url FROM entity_link').fetchall(), [('https://brand.example/',)])
self.assertEqual(json.loads((root / 'receipt.json').read_text(encoding='utf-8'))['summary'],
{'keep': 1, 'evidence': 0, 'candidate': 1, 'aliases': 1, 'links': 1})
def test_resume_retries_rate_limited_entities(self):
from scripts import harvest_company_profiles as harvest
self.assertTrue(harvest.settled({'status': '已核查', 'pages': [{'http_status': 200}]}))
self.assertTrue(harvest.settled({'status': '未取得', 'reason': '没有可访问的已确认官网', 'pages': []}))
self.assertFalse(harvest.settled({'status': '已核查', 'pages': [{'http_status': 200}, {'http_status': 429}]}))
self.assertFalse(harvest.settled({'status': '未取得', 'pages': [{'http_status': 403}]}))
self.assertFalse(harvest.settled({'status': '未取得', 'pages': [{'reason': 'ConnectTimeout'}]}))
def test_public_profile_requires_public_source_and_known_nonempty_values(self):
facts = {key: {'value': value, 'source_url': url} for key, value, url in [('country','日本','https://brand.example/company'), ('operator','secret','https://localhost/admin'), ('founded','','https://brand.example/'), ('telephone','123','https://brand.example/')]}
self.assertEqual(list(company_profiles.public_profile({'company_profile': facts})), ['country'])
def test_company_profile_survives_catalog_ingest(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
root = Path(temporary.name).resolve()
path = fresh_ledger(root)
connection = sqlite3.connect(path)
self.addCleanup(connection.close)
connection.execute("INSERT INTO asset(id,location,path,name,medium) VALUES(1,'local','C:/test/a.mp4','a.mp4','video')")
entity_id = upsert_asset_entity(connection, kind='studio', name='Brand', asset_id=1, role='studio', source='scan:filename')
fact = {'value': '2002', 'source_url': 'https://brand.example/company'}
company_profiles.fill(connection, entity_id, {'founded': fact}, source='auto:company-profile', batch='batch1')
upsert_asset_entity(connection, kind='studio', name='Brand', asset_id=1, role='studio', source='scan:filename', metadata={'provider': 'fresh'})
metadata = json.loads(connection.execute('SELECT metadata_json FROM entity WHERE id=?', (entity_id,)).fetchone()[0])
self.assertEqual(metadata['provider'], 'fresh')
self.assertEqual(metadata['company_profile']['founded']['value'], '2002')
connection.commit()
contract = web_contract.WebContract(path, avatar_root=root / 'avatars', logo_root=root / 'logos', cover_root=root / 'covers')
response = web_contract.q_entity(contract, {'kind': 'studio', 'name': 'Brand'})
self.assertEqual(response['company_profile']['founded']['value'], '2002')
if __name__ == '__main__':
unittest.main()