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

"""Seal reviewed public community articles through the existing research intake."""
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
from pathlib import Path
import re
import sqlite3
import sys

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/five-topic-collection-20261110'
BATCH = ROOT / 'work/database-enrichment-20261110/batches/five-topics-public-01'
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(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding='utf8')


def main():
    spec = importlib.util.spec_from_file_location('public_collection', ROOT / 'scripts/collect-five-topic-public.py')
    collector = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(collector)
    result = json.loads((WORK / 'public-collection-result.json').read_bytes())
    assert result['status'] == 'complete'
    # Only the original community experience texts reviewed in this pass.
    approved = {'https://post.smzdm.com/talk/p/a6p3l48e/', 'https://post.smzdm.com/p/aek8v7zq/',
                'https://post.smzdm.com/p/aqmr7o3x/', 'https://post.smzdm.com/p/arqpdepg/',
                'https://post.smzdm.com/p/apm9dg09/', 'https://post.smzdm.com/p/a4d38e3k/'}
    selected = [row for row in result['sources'] if row['url'] in approved]
    assert len(selected) == len(approved) == 6, 'REVIEWED_SOURCE_MISSING'
    assert not BATCH.exists(), 'FRESH_SEALED_BATCH_REQUIRED'
    BATCH.mkdir(parents=True)
    batch = {'schema': 'ksf.voc-agent-batch.v2', 'batchId': 'ksf-five-topics-public-20261110-r1',
             'producer': 'Codex: reviewed public SMZDM article originals; exact URL and HTML extraction; author claims not independently verified',
             'sources': [], 'attachments': [], 'social': []}
    expected, topic_counts = [], {topic['id']: 0 for topic in collector.PLAN['topics']}
    for index, row in enumerate(selected):
        http = row['httpReceipt']
        raw = (WORK / 'public-sources' / http['responseFile']).read_bytes()
        assert sha(raw) == http['sha256'], 'SOURCE_HTTP_SHA_MISMATCH'
        title, body, selector, published, encoding = collector.extract(raw, http)
        assert title == row['title'] and body == row['body'], 'SOURCE_EXTRACTION_CHANGED'
        assert sha(body.encode()) == row['bodySha256'], 'BODY_SHA_MISMATCH'
        match = re.fullmatch(r'https://post\.smzdm\.com/(?:talk/)?p/([a-z0-9]+)/', row['url'])
        assert match, 'CANONICAL_POST_ID_REQUIRED'
        post_id = match.group(1)
        original_path = 'raw/original/' + post_id + '.html'
        (BATCH / 'raw/original').mkdir(parents=True, exist_ok=True)
        (BATCH / original_path).write_bytes(raw)
        receipt_path = 'raw/original/' + post_id + '.http.json'
        save(BATCH / receipt_path, http)
        for relative in [original_path, receipt_path]:
            batch['attachments'].append({'path': relative, 'sha256': sha((BATCH / relative).read_bytes())})
        record = {'postId': post_id, 'title': title, 'content': body, 'url': row['url'],
                  'contentScope': 'packaged_noodles', 'publishedAt': published}
        adapter = {'kind': 'explicit_html_extraction_adapter_not_upstream_json', 'record': record,
                   'original': {'path': original_path, 'sha256': sha(raw), 'sourceUrl': row['url'],
                                'resolvedUrl': http['finalUrl'], 'collectedAt': http['collectedAt'],
                                'selector': selector, 'encoding': encoding},
                   'researchTopics': row['topics'],
                   'limits': ['原作者体验及观点，未独立验证功效/价格/市场数字',
                              '历史原文保留原日期，不冒充近期趋势或当前商品报价',
                              '外部社区文章，不是电商购买评价，不推定中产或学生身份']}
        relative = 'raw/adapter-' + post_id + '.json'
        save(BATCH / relative, adapter)
        sid = 'public-' + post_id
        batch['sources'].append({'id': sid, 'path': relative, 'sha256': sha((BATCH / relative).read_bytes()),
                                 'platform': 'smzdm', 'collectedAt': http['collectedAt'], 'url': row['url']})
        batch['social'].append({'sourceId': sid, 'pointer': '/record',
                                'fields': {field: '/' + field for field in record if record[field] is not None}})
        topics = [item['topic'] for item in row['topics']]
        for topic in topics: topic_counts[topic] += 1
        expected.append({'entity': 'social', 'id': 'smzdm:' + post_id, 'postId': post_id, 'title': title,
                         'content': body, 'url': row['url'], 'published_at_raw': published,
                         'originalSha256': sha(raw), 'adapterSha256': sha((BATCH / relative).read_bytes()),
                         'adapterPath': relative, 'originalPath': original_path, 'topics': topics})
    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'] == 6
    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) == 6
        for expected_row in expected:
            actual = rows[expected_row['id']]
            for field in ['title', 'content', 'url', 'published_at_raw']: assert actual[field] == expected_row[field]
            assert actual['platform'] == 'smzdm' and actual['fulltext_available'] == 1
    transport, fingerprint = library_channel.pack(BATCH / 'batch.json')
    (BATCH / 'transport.zip').write_bytes(transport)
    receipt = {'kind': 'five_topic_public_community_originals', 'batchId': batch['batchId'],
               'archiveSha256': fingerprint, 'archiveBytes': len(transport), 'accepted': {'social': 6},
               'products': [], 'sourceTopics': topic_counts, 'validation': built,
               'productionImported': False, 'createdAt': datetime.now(timezone.utc).isoformat()}
    save(BATCH / 'receipt.json', receipt)
    save(BATCH / 'expected-records.json', expected)
    excluded = [{'url': r['url'], 'title': r['title'],
                 'reason': 'not selected for import: generic health claims, unattributed consumer quotes, recipe content or insufficient verified product experience'}
                for r in result['sources'] if r['url'] not in approved]
    save(WORK / 'reviewed-public-sources.json', {'selected': expected, 'excluded': excluded,
                                               'countsByTopic': topic_counts, 'sourceBodiesReviewed': 12,
                                               'acceptedUniqueOriginals': 6})
    print(json.dumps({'ok': True, 'batch': BATCH.name, 'originals': len(expected),
                      'countsByTopic': topic_counts, 'archiveSha256': fingerprint,
                      'archiveBytes': len(transport), 'productionWrites': 0}, ensure_ascii=False))


if __name__ == '__main__': main()