/** 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}));