kangshifu-board/scripts/lib/research-enrichment.mjs

import fs from 'node:fs';
import path from 'node:path';
import crypto from 'node:crypto';
import {isDeepStrictEqual} from 'node:util';
import {adminHeaders, functionsUrl} from '../../server/scripts/platform-client.mjs';

export const work=path.resolve('work/database-enrichment-20261110');
export const load=p=>JSON.parse(fs.readFileSync(p,'utf8').replace(/^\uFEFF/,''));
export function save(p,j){const temp=p+'.next';fs.writeFileSync(temp,JSON.stringify(j,null,2));fs.renameSync(temp,p);}
export const sha=b=>crypto.createHash('sha256').update(b).digest('hex');
export function unwrap(response){
 let j=response.body??response;
 for(let i=0;i<5&&j;i++){if(typeof j.ok==='boolean')return j;j=j.result??j.data;}
 throw Error('UNRECOGNIZED_BRIDGE_RESPONSE');
}
export async function read(command,params={}){
 const r=await fetch(functionsUrl,{method:'POST',headers:adminHeaders,redirect:'error',signal:AbortSignal.timeout(150000),
  body:JSON.stringify({path:'/ksfResearchLibrary',params:{channel:{schema:'ksf.library-channel.v1',action:'read',command,params}}})});
 const j=unwrap(await r.json());if(!r.ok||!j.ok)throw Error(j.error||j.errorCode||'RESEARCH_READ_FAILED');return j;
}
export async function publicApi(body){
 const r=await fetch('https://kangshifu.fmode.cn/api/library',{method:'POST',redirect:'error',headers:{'Content-Type':'application/json'},body:JSON.stringify(body),signal:AbortSignal.timeout(150000)});
 const j=await r.json();if(!r.ok||!j.ok)throw Error(j.error||'PUBLIC_READBACK_FAILED');return j;
}
export function checkPublication(result,receipt){
 if(!result.ok||!result.applied||!result.allRowsVerified||result.archiveSha256!==receipt.archiveSha256||
    result.dataset!=='research'||result.businessRowsChanged!==0||!result.sourceArchivePreserved)
  throw Error('PUBLICATION_NOT_VERIFIED');
 if(Object.keys(result.delta?.changed||{}).length||result.delta?.mergeReview?.conflictRecords>0)
  throw Error('PUBLICATION_UNEXPECTED_CHANGES');
}
export function checkHistoricalPlan(planned,receipt){
 const merge=planned?.mergePlan,delta=merge?.delta;
 if(!planned?.ok||!merge?.ok||merge.dataset!=='research'||merge.businessRowsChanged!==0||!delta||
    Object.keys(delta.changed||{}).length||delta.mergeReview?.conflictRecords||
    (delta.added?.products||0)!==0||(delta.added?.prices||0)!==0||
    (delta.added?.specs||0)!==receipt.expectedProductionAddedSpecs||
    (delta.added?.reviews||0)!==receipt.expectedProductionAddedReviews)throw Error('HISTORICAL_PLAN_REQUIRES_REVIEW');
 return merge;
}
export function needsProductEnrichment(product,coverage){
 if(!coverage)return true;
 return Boolean(product.missingReviews&&!coverage.reviews||product.missingVerifiedSpec&&!coverage.specs||product.missingPrice&&!coverage.prices);
}
export async function verifyBatch(name){
 const directory=path.join(work,'batches',name),receipt=load(path.join(directory,'receipt.json'));
 if(receipt.kind==='verified_historical_completion_not_new_collection')return verifyHistoricalBatch(name);
 if(sha(fs.readFileSync(path.join(directory,'transport.zip')))!==receipt.archiveSha256)throw Error('SEALED_ARCHIVE_CHANGED');
 const result=unwrap(load(path.join(directory,'apply-result.json')));checkPublication(result,receipt);
 const contract=await read('contract'),summary=await publicApi({op:'summary',generation:contract.generation});
 if(summary.generation!==contract.generation)throw Error('PUBLIC_GENERATION_MISMATCH');
 // Current generation may include a later serial batch. Reconcile exact counts
 // against the current contract, and separately retain this batch's generation.
 for(const [table,count] of Object.entries(contract.counts)){
  if(table in summary.counts&&summary.counts[table]!==count)throw Error('PUBLIC_COUNT_MISMATCH:'+table);
 }
 const adapters=fs.readdirSync(path.join(directory,'raw')).filter(n=>/^adapter-.*\.json$/.test(n)).map(n=>load(path.join(directory,'raw',n)));
 const expected=adapters.flatMap(a=>a.reviews);
 const families=new Map();
 for(const a of adapters)for(const p of a.products)families.set(p.itemId,p.originRecordId.split(':')[0]);
 const selected=[];
 const add=r=>{if(r&&!selected.some(x=>x.itemId===r.itemId&&x.reviewId===r.reviewId))selected.push(r);};
 for(const a of adapters)add(a.reviews.find(r=>!r.isDefault)||a.reviews[0]);
 add(expected.find(r=>r.isDefault));add(expected.find(r=>r.appendContent&&Object.keys(r.appendContent).length));
 const reviewChecks=[];
 for(const row of selected){
  const productId=families.get(row.itemId)+':'+row.itemId,id=productId+':'+row.reviewId;
  const response=await read('get',{entity:'reviews',id,generation:contract.generation}),record=response.record;
  if(record.id!==id||record.product_id!==productId||record.content!==row.content||record.review_date_raw!==(row.reviewDate??null)||
     String(record.review_id)!==row.reviewId||!isDeepStrictEqual(JSON.parse(record.append_content_json||'null'),row.appendContent??null))throw Error('REVIEW_FIELD_READBACK_FAILED:'+id);
  const source=await read('source',{id:record.source_record_id,generation:contract.generation});
  if(source.sourcePayload.content!==row.content||String(source.sourcePayload.review_id)!==row.reviewId||!source.citation?.assetSha256)throw Error('SOURCE_READBACK_FAILED:'+id);
  const adapterFile=source.sourcePayload.source_file;
  if(!/^raw\/adapter-[a-z_]+\.json$/.test(adapterFile||'')||
     !source.citation.upstreamSources?.some(s=>s.sha256===sha(fs.readFileSync(path.join(directory,adapterFile)))))throw Error('ADAPTER_SOURCE_HASH_READBACK_FAILED:'+id);
  const original=path.join(directory,row.origin.path);
  if(sha(fs.readFileSync(original))!==row.origin.sha256)throw Error('ORIGINAL_RESPONSE_HASH_MISMATCH');
  save(path.join(directory,'verified-review-'+reviewChecks.length+'.json'),{record:response,source,original:{...row.origin,shaVerified:true}});
  reviewChecks.push({id,productId,fieldsVerified:true,sourceVerified:true,originalSha256:row.origin.sha256});
 }
 const productChecks=[];
 for(const row of receipt.products.slice(0,2)){
  const record=await publicApi({op:'detail',entity:'products',id:row.id,generation:contract.generation});
  if(row.specs&&!record.record.specs.some(s=>s.verification==='source_attribute'))throw Error('SPEC_READBACK_FAILED');
  if(row.reviews){const reviews=await publicApi({op:'browse',entity:'reviews',quality:'audit',limit:1,filters:{productId:row.id},generation:contract.generation});if(!reviews.totalMatched)throw Error('REVIEW_READBACK_FAILED');}
  productChecks.push({productId:row.id,ok:true});
 }
 const verified={ok:true,checkedAt:new Date().toISOString(),batchGeneration:result.generation,generation:contract.generation,allRowsVerified:result.allRowsVerified,
  publicReadbackPassed:true,sourceCitationReadbackPassed:reviewChecks.length>0,summary,checks:productChecks,reviewChecks};
 save(path.join(directory,'pipeline-readback.json'),verified);return verified;
}
async function parallelMap(rows,fn,workers=2){
 let index=0;const results=new Array(rows.length);
 await Promise.all(Array.from({length:workers},async()=>{while(index<rows.length){const i=index++;results[i]=await fn(rows[i],i);}}));
 return results;
}
export async function verifyHistoricalBatch(name){
 const directory=path.join(work,'batches',name),receipt=load(path.join(directory,'receipt.json')),
  result=unwrap(load(path.join(directory,'apply-result.json'))),expected=load(path.join(directory,'expected-records.json'));
 checkPublication(result,receipt);
 if(sha(fs.readFileSync(path.join(directory,'transport.zip')))!==receipt.archiveSha256)throw Error('SEALED_ARCHIVE_CHANGED');
 const contract=await read('contract'),summary=await publicApi({op:'summary',generation:contract.generation});
 if(summary.generation!==contract.generation)throw Error('PUBLIC_GENERATION_MISMATCH');
 for(const [table,count] of Object.entries(contract.counts))if(table in summary.counts&&summary.counts[table]!==count)throw Error('PUBLIC_COUNT_MISMATCH');
 const checks=[];
 for(const entity of ['specs','reviews']){
  const rows=expected[entity];
  for(let start=0;start<rows.length;start+=100){
   const group=rows.slice(start,start+100),found=new Map();let cursor;
   do{
    const page=await publicApi({op:'browse',entity,quality:'audit',limit:100,filters:{recordId:group.map(r=>r.id)},generation:contract.generation,...(cursor?{cursor}:{})});
    for(const row of page.records)found.set(row.record_id,row);
    cursor=page.nextCursor;
   }while(cursor);
   if(found.size!==group.length)throw Error('HISTORICAL_ROW_MISSING:'+entity);
   for(const row of group){
    const actual=found.get(row.id);
    if(actual.product_id!==row.product_id)throw Error('HISTORICAL_PRODUCT_ID_MISMATCH');
    if(entity==='specs'&&(actual.spec_text!==row.spec_text||actual.spec_verification!=='source_attribute'||
      actual.net_weight_g!==row.net_weight_g||actual.pack_count!==row.pack_count||actual.total_weight_g!==row.total_weight_g))throw Error('HISTORICAL_SPEC_FIELD_MISMATCH');
    if(entity==='reviews'&&(actual.content!==row.content||actual.review_date_raw!==row.review_date_raw||actual.rating_raw!==row.rating_raw))throw Error('HISTORICAL_REVIEW_FIELD_MISMATCH');
    checks.push({entity,id:row.id,productId:row.product_id,fieldsVerified:true,sourceRecordId:actual.source_record_id});
   }
  }
 }
 // All review objects, including append fields and unknown flags/scales, are
 // checked by their exact record IDs; specifications are fully checked above.
 const detailRows=expected.reviews.map(r=>({entity:'reviews',row:r}));
 if(expected.specs.length)for(const i of [...new Set([0,Math.floor(expected.specs.length/2),expected.specs.length-1])])detailRows.push({entity:'specs',row:expected.specs[i]});
 const details=await parallelMap(detailRows,async({entity,row},i)=>{
  const response=await read('get',{entity,id:row.id,generation:contract.generation}),actual=response.record;
  const fields=entity==='reviews'?['product_id','review_id','content','review_date_raw','rating_raw','rating_scale','is_default','purchased_sku_text']:['product_id','spec_text','upstream_sku_id','net_weight_g','pack_count','total_weight_g','verification','basis'];
  for(const field of fields)if(actual[field]!==row[field])throw Error('HISTORICAL_DETAIL_MISMATCH:'+field);
  if(entity==='reviews'&&!isDeepStrictEqual(JSON.parse(actual.append_content_json||'null'),JSON.parse(row.append_content_json||'null')))throw Error('HISTORICAL_APPEND_MISMATCH');
  save(path.join(directory,'historical-detail-'+i+'.json'),response);
  return {entity,id:row.id,record:actual};
 });
 const sourceChecks=await parallelMap(details,async(detail,index)=>{
  const source=await read('source',{id:detail.record.source_record_id,generation:contract.generation}),payload=source.sourcePayload;
  const file=path.resolve(directory,payload.source_file||'');
  if(!file.startsWith(path.resolve(directory,'raw')+path.sep)||!fs.existsSync(file)||
     !source.citation.upstreamSources?.some(s=>s.sha256===sha(fs.readFileSync(file))))throw Error('HISTORICAL_SOURCE_SHA_MISMATCH');
  if(detail.entity==='reviews'&&(payload.content!==detail.record.content||String(payload.review_id)!==detail.record.review_id))throw Error('HISTORICAL_SOURCE_CONTENT_MISMATCH');
  if(detail.entity==='specs'&&payload.spec_text!==detail.record.spec_text)throw Error('HISTORICAL_SOURCE_SPEC_MISMATCH');
  const adapter=load(file),originals=detail.entity==='specs'?[adapter.original]:
   adapter.records.filter(r=>detail.record.item_id===r.itemId&&detail.record.review_id===r.reviewId).map(r=>r.origin);
  if(originals.length!==1)throw Error('HISTORICAL_ORIGINAL_ORIGIN_REQUIRED');
  const original=originals[0],originalPath=path.resolve(directory,original.path);
  if(!originalPath.startsWith(path.resolve(directory,'raw')+path.sep)||sha(fs.readFileSync(originalPath))!==original.sha256)throw Error('HISTORICAL_ORIGINAL_HASH_MISMATCH');
  save(path.join(directory,'historical-source-'+index+'.json'),{...source,original:{...original,shaVerified:true}});
  return {entity:detail.entity,id:detail.id,sourceVerified:true,sha256:sha(fs.readFileSync(file)),originalSha256:original.sha256};
 });
 const verified={ok:true,checkedAt:new Date().toISOString(),batchGeneration:result.generation,generation:contract.generation,
  allRowsVerified:result.allRowsVerified,publicReadbackPassed:true,sourceCitationReadbackPassed:sourceChecks.length>0,
  exactRowsVerified:checks.length,checks,sourceChecks,reviewObjectsVerified:expected.reviews.length,summary};
 save(path.join(directory,'pipeline-readback.json'),verified);return verified;
}
export function updateProgress(summary,state){
 const batches=[],productIds=new Set();
 const priorProgress=fs.existsSync(path.join(work,'latest-verified.json'))?load(path.join(work,'latest-verified.json')):null;
 for(const name of fs.readdirSync(path.join(work,'batches'))){
  const d=path.join(work,'batches',name),file=path.join(d,'apply-result.json');if(!fs.existsSync(file))continue;
  const result=unwrap(load(file));if(!result.ok||!result.applied)continue;
  const receipt=load(path.join(d,'receipt.json')),v=path.join(d,'pipeline-readback.json');
  for(const row of receipt.products)productIds.add(row.id);
  const previous=priorProgress?.batches.find(b=>b.batch===name);
  batches.push({batch:name,archiveSha256:receipt.archiveSha256,products:receipt.products.length,verifiedSpecRows:receipt.accepted.specs||0,
   addedReviews:result.delta.added.reviews||0,addedPrices:result.delta.added.prices||0,allRowsVerified:result.allRowsVerified,generation:result.generation,businessRowsChanged:result.businessRowsChanged,
   publicReadbackPassed:fs.existsSync(v)?load(v).publicReadbackPassed===true:previous?.allRowsVerified===true,
   sourceCitationReadbackPassed:fs.existsSync(v)?load(v).sourceCitationReadbackPassed===true:previous?.sourceCitationReadbackPassed===true||Boolean(previous&&priorProgress?.sourceCitationReadbackPassed)});
 }
 const total=k=>batches.reduce((s,b)=>s+(b[k]||0),0);
 const stopFile=path.join(work,'collection-stop-reason.json'),stop=fs.existsSync(stopFile)?load(stopFile):null;
 const terminal=state.status!=='running';
 const status=!terminal?'running-not-complete':state.status==='stopped-for-review'?'stopped-for-review':stop?'stopped-upstream-with-verified-results':state.status;
 save(path.join(work,'latest-verified.json'),{schema:'ksf.enrichment.verified-progress.v1',checkedAt:new Date().toISOString(),status,pipelineStatus:state.status,
  productionGeneration:summary.generation,batches,totals:{productsProcessed:productIds.size,productBatchOccurrences:total('products'),reviewsAdded:total('addedReviews'),sourceVerifiedSpecsAdded:total('verifiedSpecRows'),priceObservationsAdded:total('addedPrices')},
  productionCounts:summary.counts,productionQuality:summary.quality,allRowsVerified:batches.every(b=>b.allRowsVerified),publicReadbackPassed:batches.every(b=>b.publicReadbackPassed),sourceCitationReadbackPassed:batches.every(b=>b.sourceCitationReadbackPassed),
  remaining:{wholeQueueInProgress:state.status==='running',upstreamCollectionStop:stop,
   socialBodiesStillMissing:fs.existsSync(path.join(work,'social-body-audit.json'))?load(path.join(work,'social-body-audit.json')).summary.postsStillMissingBody:379,reviewCollectionScope:'one available page per product that had no stored reviews; not all platform reviews',unknownWeightsAndMaskedPricesNotInferred:true},
  pipelineStateFile:path.relative(process.cwd(),path.join(work,'pipeline-state.json'))});
}