kangshifu-board/scripts/prepare-five-topic-multisource-20261111.py

"""Seal reviewed HTML originals, precise excerpts and government documents."""
from datetime import datetime,timezone
import hashlib
import importlib.util
import json
from pathlib import Path
import sqlite3
import sys
from lxml import html

ROOT=Path(__file__).resolve().parents[1]
WORK=ROOT/(sys.argv[1] if len(sys.argv)>1 else 'work/five-topic-research-20261111')
BATCH=ROOT/'work/database-enrichment-20261110/batches'/(sys.argv[2] if len(sys.argv)>2 else 'five-topics-multisource-20261111-01')
BATCH_ID=sys.argv[3] if len(sys.argv)>3 else 'ksf-five-topics-multisource-20261111-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 main():
 review=json.loads((WORK/'reviewed-selection.json').read_bytes())
 dedup=json.loads((WORK/'dedup-result.json').read_bytes());assert dedup['ok']
 selected=dedup['selected'];assert selected and not BATCH.exists()
 spec=importlib.util.spec_from_file_location('replay_source',ROOT/'scripts/collect-five-topic-public.py')
 base=importlib.util.module_from_spec(spec);spec.loader.exec_module(base)
 (BATCH/'raw/original').mkdir(parents=True)
 batch={'schema':'ksf.voc-agent-batch.v2','batchId':BATCH_ID,
  'producer':'Codex reviewed multi-source topic collection with explicit source boundaries',
  'sources':[],'attachments':[],'social':[],'documents':[],'clauses':[]}
 expectations={name:[] for name in ['social','documents','clauses']}
 source_metadata={}
 def attach(name,raw):
  (BATCH/name).write_bytes(raw);batch['attachments'].append({'path':name,'sha256':sha(raw)})
 for i,r in enumerate(selected):
  original=(WORK/r['collectionDirectory']/r['httpReceipt']['responseFile']).read_bytes()
  assert sha(original)==r['httpReceipt']['sha256']
  tree=html.fromstring(original.decode(r['sourceEncoding']));nodes=tree.xpath(r['bodyExtractionSelector']);assert len(nodes)==1
  body=base.paragraphs(nodes[0])
  if 'paragraphIndexes' in r:
   lines=body.splitlines();body='\n'.join(lines[n] for n in r['paragraphIndexes'])
  assert body==r['body'] and sha(body.encode())==r['bodySha256'],'SOURCE_REPLAY_FAILED'
  assert all(quote in body for t in r['topics'] for quote in t['passages'])
  stem=str(i+1).zfill(2)+'-'+r['id'].replace(':','-')
  op='raw/original/'+stem+'.html';attach(op,original)
  hp='raw/original/'+stem+'.http.json';attach(hp,json.dumps(r['httpReceipt'],ensure_ascii=False,indent=2).encode())
  if r['entity']=='social':
   record={'postId':r['postId'],'title':r['title'],'content':body,'url':r['url'],
           'publishedAt':r['publishedAtRaw'],'contentScope':r['contentScope']}
  else:
   tp='raw/original/'+stem+'.txt';attach(tp,body.encode())
   record={'documentId':r['id'],'title':r['title'],'sourceUrl':r['url'],
     'declaredStatus':r['declaredStatus'],'statusAsOf':datetime.now(timezone.utc).date().isoformat(),
     'issuedDate':r['publishedAtRaw'],'fullText':body,'expectedSha256':sha(body.encode()),'textFile':tp,
     'coverage':json.dumps({'sourceType':r['sourceType'],'topics':r['topics'],'dateEvidence':r['dateEvidence'],
        'limits':r['limits']+review['limits']},ensure_ascii=False)}
  adapter={'kind':'explicit_html_extraction_adapter_not_upstream_json','record':record,
   'original':{'path':op,'sha256':sha(original),'sourceUrl':r['httpReceipt']['requestedUrl'],
       'resolvedUrl':r['httpReceipt']['finalUrl'],'encoding':r['sourceEncoding'],'selector':r['bodyExtractionSelector'],
       'paragraphIndexes':r.get('paragraphIndexes'),'collectedAt':r['httpReceipt']['collectedAt']},
   'researchTopics':r['topics'],'sourceType':r['sourceType'],'reviewNote':r['reviewNote'],
   'limits':r['limits']+review['limits']}
  if r['entity']=='documents':
   adapter['clauses']=[{'documentId':r['id'],'clause':c['name'],'excerpt':c['excerpt'],
    'excerptMethod':'HTML正文逐字摘录；XPath '+r['bodyExtractionSelector']+'；非PDF逐字核验',
    'applicationNote':c['application']} for c in r['clauses']]
  ap='raw/adapter-'+stem+'.json';save(BATCH/ap,adapter);sid='source-'+stem
  batch['sources'].append({'id':sid,'path':ap,'sha256':sha((BATCH/ap).read_bytes()),
       'platform':r['platform'],'url':r['url'],'collectedAt':r['httpReceipt']['collectedAt']})
  batch[r['entity']].append({'sourceId':sid,'pointer':'/record','fields':{k:'/'+k for k in record}})
  for j,c in enumerate(adapter.get('clauses',[])):
   batch['clauses'].append({'sourceId':sid,'pointer':'/clauses/'+str(j),'fields':{k:'/'+k for k in c}})
  source_metadata[ap]={'adapterPath':ap,'adapterSha256':sha((BATCH/ap).read_bytes()),'originalPath':op,'originalSha256':sha(original),
    'topics':[t['topic'] for t in r['topics']],'topicEvidence':r['topics'],'sourceType':r['sourceType'],'reviewNote':r['reviewNote']}
 save(BATCH/'batch.json',batch)
 dry=structured_batch.build(BATCH/'batch.json',BATCH/'staging.sqlite',False);save(BATCH/'validation.json',dry)
 assert dry['ok'] and dry['networkRequests']==0 and dry['databaseWrites']==0
 built=structured_batch.build(BATCH/'batch.json',BATCH/'staging.sqlite',True)
 with sqlite3.connect(BATCH/'staging.sqlite') as c:
  c.row_factory=sqlite3.Row
  for entity,table in [('social','social_posts'),('documents','policy_documents'),('clauses','policy_clauses')]:
   for record in c.execute('SELECT * FROM '+table):
    actual=dict(record)
    payload=json.loads(c.execute('SELECT payload_json FROM source_records WHERE id=?',(actual['source_record_id'],)).fetchone()[0])
    ap=payload['source_file'];meta=source_metadata[ap]
    assert entity!='documents' or actual['attachment_hash_state']=='matched'
    assert entity!='social' or actual['fulltext_available']==1
    expectations[entity].append({'entity':entity,'id':actual['id'],'record':actual,**meta})
 transport,fingerprint=library_channel.pack(BATCH/'batch.json');(BATCH/'transport.zip').write_bytes(transport)
 receipt={'kind':'five_topic_reviewed_multisource','batchId':batch['batchId'],'archiveSha256':fingerprint,'archiveBytes':len(transport),
   'accepted':{k:len(v) for k,v in expectations.items()},'products':[],'sourceTopics':dedup['countsByTopic'],'validation':built,
   'createdAt':datetime.now(timezone.utc).isoformat(),'productionImported':False}
 save(BATCH/'receipt.json',receipt);save(BATCH/'expected-records.json',expectations)
 save(BATCH/'reviewed-source-index.json',selected)
 print(json.dumps({'ok':True,'batch':BATCH.name,'accepted':receipt['accepted'],'archiveBytes':len(transport)},ensure_ascii=False))

if __name__=='__main__':main()