kangshifu-board/scripts/verify-five-topic-supplement.mjs

/** Read every supplemental article and its citation, then report exact new rows. */
import fs from 'node:fs';
import path from 'node:path';
import assert from 'node:assert/strict';
import {read, publicApi, load, save, sha, unwrap, checkPublication} from './lib/research-enrichment.mjs';
async function retryRead(operation) {
  for (let attempt = 0; ; attempt++) {
    try { return await operation(); }
    catch (error) {
      const code = error.cause?.code || error.code || error.name;
      if (attempt >= 2 || !['ECONNRESET', 'ETIMEDOUT', 'EAI_AGAIN', 'UND_ERR_SOCKET', 'UND_ERR_CONNECT_TIMEOUT', 'TimeoutError'].includes(code)) throw error;
      console.log(JSON.stringify({stage: 'read-only-network-retry', attempt: attempt + 1, code}));
      await new Promise(resolve => setTimeout(resolve, 1200));
    }
  }
}
const [workArg, batchArg, ...priorBatchArgs] = process.argv.slice(2);
const work = path.resolve(workArg || 'work/five-topic-collection-20261110/supplement-01');
const batchName = batchArg || 'five-topics-public-02';
const priorBatchNames = priorBatchArgs.length ? priorBatchArgs : ['five-topics-public-01'];
assert([batchName, ...priorBatchNames].every(name => /^five-topics-public-\d+$/.test(name)));
const directory = path.resolve('work/database-enrichment-20261110/batches', batchName);
const expected = load(path.join(directory, 'expected-records.json'));
const receipt = load(path.join(directory, 'receipt.json'));
const applied = unwrap(load(path.join(directory, 'apply-result.json')));
const planned = unwrap(load(path.join(directory, 'plan-result.json'))).mergePlan;
const status = unwrap(load(path.join(directory, 'status-result.json')));
checkPublication(applied, receipt);
assert.equal(applied.generation, planned.plannedGeneration);
assert(status.ok && status.publicationFound);
assert.equal(status.publishedGeneration, applied.generation);
assert.equal(sha(fs.readFileSync(path.join(directory, 'transport.zip'))), receipt.archiveSha256);
assert.equal(applied.delta.added.social_posts, expected.length);
for (const entity of ['products', 'reviews', 'specs', 'prices']) assert.equal(applied.delta.added[entity] || 0, 0);
const before = load(path.join(work, 'contract-before.json'));
const contract = await retryRead(() => read('contract'));
if (contract.generation === applied.generation) {
  for (const [table, count] of Object.entries(before.counts)) {
    assert.equal(contract.counts[table], count + (applied.delta.added[table] || 0), 'COUNT_MISMATCH:' + table);
  }
}
const generation = contract.generation;
const checks = [];
let next = 0;
await Promise.all(Array.from({length: 2}, async () => {
  while (next < expected.length) {
    const index = next++, row = expected[index];
    const checkpointPath = path.join(directory, 'verified-source-' + index + '.json');
    const checkpoint = fs.existsSync(checkpointPath) ? load(checkpointPath) : null;
    const cached = checkpoint?.record?.generation === generation && checkpoint?.source?.generation === generation;
    const response = cached ? checkpoint.record : await retryRead(() => read('get', {entity: 'social', id: row.id, generation}));
    const actual = response.record;
    assert.equal(actual.id, row.id);
    assert.equal(actual.post_id, row.postId);
    assert.equal(actual.fulltext_available, 1);
    for (const field of ['platform', 'content_scope', 'title', 'content', 'url', 'published_at_raw']) {
      assert.equal(actual[field], row[field], 'ARTICLE_READBACK_MISMATCH:' + row.id + ':' + field);
    }
    const source = cached ? checkpoint.source : await retryRead(() => read('source', {id: actual.source_record_id, generation}));
    assert.equal(source.citation.id, actual.source_record_id);
    assert.equal(source.sourcePayload.content, row.content);
    assert.equal(source.sourcePayload.id, row.postId);
    assert(source.citation.upstreamSources.some(item => item.sha256 === row.adapterSha256));
    const adapter = load(path.join(directory, row.adapterPath));
    assert.equal(adapter.original.sha256, row.originalSha256);
    assert.equal(sha(fs.readFileSync(path.join(directory, row.originalPath))), row.originalSha256);
    assert.deepEqual(adapter.researchTopics, row.topicEvidence);
    assert(row.topicEvidence.every(topic => topic.passages.every(quote => actual.content.includes(quote))));
    save(checkpointPath, {record: response, source,
      original: {...adapter.original, shaVerified: true}, topicEvidence: row.topicEvidence});
    checks.push({id: row.id, fieldsVerified: true, sourceVerified: true, originalSha256: row.originalSha256});
    save(path.join(work, 'verification-state.json'), {stage: 'get-source', completed: checks.length,
      target: expected.length, generation, updatedAt: new Date().toISOString()});
    console.log(JSON.stringify({verified: checks.length, target: expected.length, id: row.id}));
  }
}));
const previous = priorBatchNames.flatMap(name =>
  load(path.resolve('work/database-enrichment-20261110/batches', name, 'expected-records.json')));
const allIds = [...previous, ...expected].map(row => row.id);
assert.equal(new Set(allIds).size, allIds.length);
const summary = await retryRead(() => publicApi({op: 'summary', generation}));
assert.equal(summary.generation, generation);
for (const [table, count] of Object.entries(contract.counts)) assert.equal(summary.counts[table], count);
const page = await retryRead(() => publicApi({op: 'browse', entity: 'social', quality: 'audit', limit: 100,
  filters: {recordId: allIds}, generation}));
assert.equal(page.totalMatched, allIds.length);
assert.equal(page.records.length, allIds.length);
assert.deepEqual(new Set(page.records.map(row => row.record_id)), new Set(allIds));
assert(page.records.every(row => row.fulltext_available === 1));
const names = {wellness: '中式养生', student: '学生（宿舍与夜宵）', premium: '高端升级（中产）', stockpile: '居家囤货', novelty: '潮流尝鲜'};
const topics = Object.entries(names).map(([id, name]) => ({id, name,
  previousArticles: previous.filter(row => row.topics.includes(id)).length,
  newArticles: expected.filter(row => row.topics.includes(id)).length,
  totalArticles: [...previous, ...expected].filter(row => row.topics.includes(id)).length,
  newRecordIds: expected.filter(row => row.topics.includes(id)).map(row => row.id),
  relationCounts: expected.flatMap(row => row.topicEvidence.filter(item => item.topic === id))
    .reduce((result, item) => ({...result, [item.relation]: (result[item.relation] || 0) + 1}), {})}));
const sourceTypes = [...previous, ...expected].reduce((result, row) => {
  const key = row.sourceType === 'editorial_purchase_guide' ? 'editorial_purchase_guide' :
    /^(smzdm|sspai):/.test(row.id) ? 'community_experience' : 'industry_reporting_or_opinion';
  result[key] = (result[key] || 0) + 1;
  return result;
}, {});
const publication = {ok: true, checkedAt: new Date().toISOString(), generation, batchGeneration: applied.generation,
  allRowsVerified: true, exactArticleObjectsVerified: checks.length, sourceCitationReadbackPassed: true,
  publicReadbackPassed: true, checks, summary};
save(path.join(directory, 'pipeline-readback.json'), publication);
const result = {schema: 'ksf.five-topic.supplement-verified.v1', checkedAt: publication.checkedAt,
  generation, batchGeneration: applied.generation, topics, newArticles: expected.length,
  previousArticles: previous.length, totalUniqueArticles: allIds.length, sourceTypes,
  batches: [...priorBatchNames, batchName],
  allRowsVerified: true, publicReadbackPassed: true, sourceCitationReadbackPassed: true,
  newProducts: 0, newEcommerceReviews: 0, productionCounts: contract.counts,
  limits: load(path.join(work, 'reviewed-selection.json')).limits,
  crossTopicMemberships: topics.reduce((sum, row) => sum + row.newArticles, 0)};
save(path.join(work, 'verified-results.json'), result);
save(path.resolve(work, '../latest-verified.json'), result);
save(path.join(work, 'verification-state.json'), {stage: 'complete', completed: checks.length,
  target: expected.length, generation, publicReadbackPassed: true, updatedAt: publication.checkedAt});
const table = ['| 课题 | 上轮已入库 | 本轮补入 | 累计关联资料 |', '|---|---:|---:|---:|',
  ...topics.map(row => `| ${row.name} | ${row.previousArticles} | ${row.newArticles} | ${row.totalArticles} |`)];
fs.writeFileSync(path.join(work, '补采数量.md'), `# 五课题补采与入库数量\n\n核验时间：${result.checkedAt}\n\n${table.join('\n')}\n\n本轮去重新增${expected.length}篇，累计去重共${allIds.length}篇；课题间允许交叉，表内行数不能直接相加。累计社区体验${sourceTypes.community_experience}篇，行业报道或评论${sourceTypes.industry_reporting_or_opinion}篇，编辑采购指南${sourceTypes.editorial_purchase_guide || 0}篇。\n\n本轮${checks.length}/${expected.length}具体记录、来源与原件SHA校验通过；看板接口${allIds.length}/${allIds.length}均可读。新代际${applied.generation}。\n\n本轮未新增电商商品、购买评价、价格和规格。资料保留原日期、原HTML、提取定位、主题对应原句及来源。\n\n${result.limits.map(text => '- ' + text).join('\n')}\n`, 'utf8');
console.log(JSON.stringify({ok: true, generation, newArticles: result.newArticles,
  totalUniqueArticles: result.totalUniqueArticles, sourceTypes, topics}));