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

/** Preserve an honest queue state after a collector stops and its publisher drains. */
import fs from 'node:fs';
import path from 'node:path';
import {work,load,save} from './lib/research-enrichment.mjs';
const alive=pid=>{if(!pid)return false;try{process.kill(pid,0);return true;}catch{return false;}};
const stateFile=path.join(work,'pipeline-state.json'),state=load(stateFile);
if(alive(state.pid)||alive(state.collectorPid))throw Error('PIPELINE_STILL_ACTIVE');
for(const name of fs.readdirSync(path.join(work,'batches'))){
 const d=path.join(work,'batches',name),attempt=path.join(d,'apply-attempt.json');
 if(!fs.existsSync(attempt))continue;
 const file=path.join(d,'apply-result.json'),verified=path.join(d,'pipeline-readback.json');
 if(!fs.existsSync(file)||!load(file).body?.result?.allRowsVerified||!fs.existsSync(verified)||!load(verified).ok)
  throw Error('UNRESOLVED_PUBLICATION_REQUIRES_REVIEW');
}
const stopFile=path.join(work,'collection-stop-reason.json');
if(fs.existsSync(stopFile)){
 save(path.join(work,'pipeline-state.before-stop-reconcile-'+Date.now()+'.json'),state);
 state.status='stopped-upstream-with-verified-results';state.stage='collected-material-verified';
 delete state.errorCode;
 state.collection={...state.collection,stopReason:load(stopFile),queueComplete:false};
 state.updatedAt=new Date().toISOString();save(stateFile,state);
}
const coverage=new Map();
for(const name of fs.readdirSync(path.join(work,'batches'))){
 const directory=path.join(work,'batches',name),file=path.join(directory,'apply-result.json');if(!fs.existsSync(file))continue;
 const result=load(file).body.result;if(!result.ok||!result.applied||!result.allRowsVerified)continue;
 for(const row of load(path.join(directory,'receipt.json')).products){
  const known=coverage.get(row.id)||{reviews:0,specs:0,prices:0,batches:[]};
  for(const field of ['reviews','specs','prices'])known[field]+=row[field]||0;
  known.batches.push(name);coverage.set(row.id,known);
 }
}
// A published product can still lack reviews, specifications or a numeric price.
// This manifest retains those individual gaps for the next recovery pass.
const queue=load(path.join(work,'product-queue.json')),remaining=[];
for(const p of queue){
 const known=coverage.get(p.productId)||{reviews:0,specs:0,prices:0,batches:[]};
 const gaps={reviews:p.missingReviews&&!known.reviews,sourceVerifiedSpecs:p.missingVerifiedSpec&&!known.specs,price:p.missingPrice&&!known.prices};
 if(Object.values(gaps).some(Boolean))remaining.push({productId:p.productId,gaps,published:known,
  recheckRequired:'retry original linked record; successful request alone does not guarantee valid rows or all fields'});
}
const manifest={schema:'ksf.enrichment.pending-recovery.v1',checkedAt:new Date().toISOString(),
 status:state.status,upstreamStop:state.collection?.stopReason||null,productsWithRemainingFields:remaining.length,
 missingReviews:remaining.filter(r=>r.gaps.reviews).length,missingSourceVerifiedSpecs:remaining.filter(r=>r.gaps.sourceVerifiedSpecs).length,
 missingPrices:remaining.filter(r=>r.gaps.price).length,products:remaining};
save(path.join(work,'pending-recovery.json'),manifest);
console.log(JSON.stringify({status:state.status,productsWithRemainingFields:manifest.productsWithRemainingFields,
 missingReviews:manifest.missingReviews,missingSourceVerifiedSpecs:manifest.missingSourceVerifiedSpecs,missingPrices:manifest.missingPrices}));