"""Seal source bytes and reviewed topic evidence using the existing intake v2.""" from datetime import datetime, timezone import argparse import hashlib import importlib.util import json from pathlib import Path import re import sqlite3 import sys from lxml import html ROOT = Path(__file__).resolve().parents[1] WORK = ROOT / 'work/five-topic-collection-20261110/supplement-01' BATCH = ROOT / 'work/database-enrichment-20261110/batches/five-topics-public-02' BATCH_ID = 'ksf-five-topics-supplement-20261110-r1' sys.path.insert(0, str(ROOT / 'server/voc-library')) import structured_batch import library_channel def sha(raw): return hashlib.sha256(raw).hexdigest() def save(file, value): file.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding='utf8') def post_id(row): host, regex = { 'smzdm': ('post.smzdm.com', r'/(?:talk/)?p/([a-z0-9]+)/'), 'foodaily': ('www.foodaily.com', r'/articles/(\d+)'), 'yicai': ('www.yicai.com', r'/news/(\d+)\.html'), 'netease': ('www.163.com', r'/dy/article/([A-Z0-9]+)\.html'), 'sspai': ('sspai.com', r'/post/(\d+)'), }[row['platform']] match = re.fullmatch('https://' + re.escape(host) + regex, row['url']) assert match, 'PRECISE_PUBLISHER_ID_REQUIRED' return match.group(1) def main(): review = json.loads((WORK / 'reviewed-selection.json').read_bytes()) dedup = json.loads((WORK / 'dedup-result.json').read_bytes()) selected = review['selected'] ids = [row['platform'] + ':' + post_id(row) for row in selected] assert dedup['ok'] and dedup['existingRecords'] == 0 and set(dedup['ids']) == set(ids), 'REMOTE_DEDUP_REQUIRED' assert len(set(ids)) == len(ids) assert not BATCH.exists(), 'FRESH_SEALED_BATCH_REQUIRED' BATCH.mkdir(parents=True) (BATCH / 'raw/original').mkdir(parents=True) batch = {'schema': 'ksf.voc-agent-batch.v2', 'batchId': BATCH_ID, 'producer': 'Codex: reviewed public community and publisher articles; quoted topic passages; author claims not independently verified', 'sources': [], 'attachments': [], 'social': []} expected = [] for row in selected: pid = post_id(row) stem = row['platform'] + '-' + pid receipt = row['httpReceipt'] raw = (WORK / row['collectionDirectory'] / receipt['responseFile']).read_bytes() assert sha(raw) == receipt['sha256'] assert sha(row['body'].encode()) == row['bodySha256'] tree = html.fromstring(raw.decode(row['sourceEncoding'])) nodes = tree.xpath(row['bodyExtractionSelector']) assert len(nodes) == 1, 'UNIQUE_BODY_LOCATOR_REQUIRED' spec = importlib.util.spec_from_file_location('base_replay', ROOT / 'scripts/collect-five-topic-public.py') base = importlib.util.module_from_spec(spec) spec.loader.exec_module(base) assert base.paragraphs(nodes[0]) == row['body'], 'ORIGINAL_BODY_REPLAY_MISMATCH' assert all(quote in row['body'] for t in row['topics'] for quote in t['passages']) original_path = 'raw/original/' + stem + '.html' (BATCH / original_path).write_bytes(raw) http_path = 'raw/original/' + stem + '.http.json' save(BATCH / http_path, receipt) for relative in [original_path, http_path]: batch['attachments'].append({'path': relative, 'sha256': sha((BATCH / relative).read_bytes())}) scope = row.get('contentScope', 'adjacent_noodle_food' if row['url'].endswith('/37590') else 'packaged_noodles') record = {'postId': pid, 'title': row['title'], 'content': row['body'], 'url': row['url'], 'contentScope': scope, 'publishedAt': row['publishedAtRaw']} adapter = {'kind': 'explicit_html_extraction_adapter_not_upstream_json', 'record': record, 'original': {'path': original_path, 'sha256': sha(raw), 'sourceUrl': row['url'], 'resolvedUrl': receipt['finalUrl'], 'collectedAt': receipt['collectedAt'], 'selector': row['bodyExtractionSelector'], 'encoding': row['sourceEncoding']}, 'researchTopics': row['topics'], 'sourceType': row['sourceType'], 'reviewNote': row['reviewNote'], 'limits': row['limits'] + review['limits']} relative = 'raw/adapter-' + stem + '.json' save(BATCH / relative, adapter) sid = 'public-' + stem batch['sources'].append({'id': sid, 'path': relative, 'sha256': sha((BATCH / relative).read_bytes()), 'platform': row['platform'], 'collectedAt': receipt['collectedAt'], 'url': row['url']}) batch['social'].append({'sourceId': sid, 'pointer': '/record', 'fields': {field: '/' + field for field in record}}) expected.append({'entity': 'social', 'id': row['platform'] + ':' + pid, 'postId': pid, 'platform': row['platform'], 'content_scope': scope, 'title': row['title'], 'content': row['body'], 'url': row['url'], 'published_at_raw': row['publishedAtRaw'], 'originalSha256': sha(raw), 'adapterSha256': sha((BATCH / relative).read_bytes()), 'adapterPath': relative, 'originalPath': original_path, 'topics': [t['topic'] for t in row['topics']], 'topicEvidence': row['topics'], 'sourceType': row['sourceType'], 'reviewNote': row['reviewNote']}) save(BATCH / 'batch.json', batch) preview = structured_batch.build(BATCH / 'batch.json', BATCH / 'staging.sqlite', False) assert preview['ok'] and preview['readOnly'] and preview['counts']['social'] == len(selected) save(BATCH / 'validation.json', preview) built = structured_batch.build(BATCH / 'batch.json', BATCH / 'staging.sqlite', True) with sqlite3.connect(BATCH / 'staging.sqlite') as connection: connection.row_factory = sqlite3.Row rows = {r['id']: dict(r) for r in connection.execute('SELECT * FROM social_posts')} assert len(rows) == len(expected) for row in expected: actual = rows[row['id']] for field in ['platform', 'content_scope', 'title', 'content', 'url', 'published_at_raw']: assert actual[field] == row[field], 'STAGING_VALUE_MISMATCH:' + field assert actual['fulltext_available'] == 1 transport, fingerprint = library_channel.pack(BATCH / 'batch.json') (BATCH / 'transport.zip').write_bytes(transport) receipt = {'kind': 'five_topic_reviewed_public_supplement', 'batchId': batch['batchId'], 'archiveSha256': fingerprint, 'archiveBytes': len(transport), 'accepted': {'social': len(expected)}, 'products': [], 'sourceTopics': review['countsByTopic'], 'validation': built, 'productionImported': False, 'createdAt': datetime.now(timezone.utc).isoformat()} save(BATCH / 'receipt.json', receipt) save(BATCH / 'expected-records.json', expected) print(json.dumps({'ok': True, 'batch': BATCH.name, 'articles': len(expected), 'countsByTopic': review['countsByTopic'], 'archiveBytes': len(transport), 'archiveSha256': fingerprint}, ensure_ascii=False)) if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('--work', default=str(WORK)) parser.add_argument('--batch', default=str(BATCH)) parser.add_argument('--batch-id', default=BATCH_ID) args = parser.parse_args() WORK, BATCH, BATCH_ID = Path(args.work).resolve(), Path(args.batch).resolve(), args.batch_id assert WORK.is_relative_to(ROOT / 'work') and BATCH.is_relative_to(ROOT / 'work'), 'WORKSPACE_OUTPUT_REQUIRED' main()