kangshifu-board/scripts/prepare-five-topic-supplement.py

"""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()