kangshifu-board/scripts/retrieve-five-topic-library.mjs

/** Read the five requested research scopes from one fixed, current generation. */
import fs from 'node:fs';
import path from 'node:path';
import {read, save, load, sha} from './lib/research-enrichment.mjs';
const work = path.resolve('work/five-topic-collection-20261110');
const plan = load(path.join(work, 'collection-plan.json'));
const contract = load(path.join(work, 'contract-before.json'));
const directory = path.join(work, 'library-pages');
fs.mkdirSync(directory, {recursive: true});
const stateFile = path.join(work, 'library-retrieval-state.json');
const state = {startedAt: new Date().toISOString(), generation: contract.generation, pid: process.pid,
  status: 'running', topics: {}, productionWrites: 0};
const alive = pid => {try {process.kill(pid, 0); return true;} catch {return false;}};
if (fs.existsSync(stateFile)) {
  const previous = load(stateFile);
  if (previous.status === 'running' && alive(previous.pid)) throw Error('TOPIC_RETRIEVER_ALREADY_RUNNING');
}
const record = () => {state.updatedAt = new Date().toISOString(); save(stateFile, state);};
const inScope = row => {
  if (['raw_noodle', 'sauce', 'other_food'].includes(row.category)) return false;
  if (row.entity === 'social' && row.content_scope && row.content_scope !== 'packaged_noodles') return false;
  if (row.entity === 'products') {
    const title = row.title || '';
    if (/挂面|生面条|鲜面条|鲜湿面|手擀面粉|面粉|拌面酱|拌面汁|堂食/.test(title)) return false;
    if (!/方便面|泡面|速食|即食|拉面|拌面|捞面|汤面|热干面|乌冬面|干脆面|杯面|桶面|碗面|面皮|凉皮|火鸡面|面饼/.test(title)) return false;
  }
  return true;
};
async function retrieve(topic) {
  const rows = [], seen = new Set();
  let cursor, page = 0, totalMatched;
  const progress = state.topics[topic.id] = {name: topic.name, pages: 0, retrieved: 0, status: 'running'};
  const groups = topic.keywords.slice(0, 20);
  do {
    page++;
    const file = path.join(directory, topic.id + '-' + String(page).padStart(3, '0') + '.json');
    const input = {entities: ['products', 'reviews', 'social'], quality: 'audit', limit: 100,
      text: {all: [{any: groups}]}, ...(cursor ? {cursor} : {})};
    const querySha256 = sha(Buffer.from(JSON.stringify({input, generation: contract.generation})));
    let result;
    if (fs.existsSync(file)) {
      const cached = load(file);
      if (cached.querySha256 === querySha256 && cached.result?.generation === contract.generation) result = cached.result;
    }
    if (!result) {
      result = await read('search', {input, generation: contract.generation});
      save(file, {querySha256, input, receivedAt: new Date().toISOString(), result});
    }
    if (result.generation !== contract.generation) throw Error('GENERATION_CHANGED');
    if (totalMatched === undefined) totalMatched = result.totalMatched;
    if (result.totalMatched !== totalMatched) throw Error('TOPIC_COUNT_CHANGED');
    for (const row of result.records) {
      const key = row.entity + ':' + row.record_id;
      if (seen.has(key)) throw Error('DUPLICATE_CURSOR_ROW');
      seen.add(key); rows.push(row);
    }
    progress.pages = page; progress.retrieved = rows.length; progress.totalMatched = totalMatched; record();
    cursor = result.nextCursor;
  } while (cursor);
  if (rows.length !== totalMatched) throw Error('TOPIC_CURSOR_INCOMPLETE');
  const accepted = [], excluded = [];
  for (const row of rows) {
    let reason;
    if (!inScope(row)) reason = 'OUTSIDE_PACKAGED_NOODLE_SCOPE';
    else if (row.entity === 'reviews' && !(row.match?.matchedTerms || []).length) reason = 'PRODUCT_TITLE_MATCH_ONLY_NOT_REVIEW_BODY';
    if (reason) excluded.push({entity: row.entity, id: row.record_id, reason});
    else accepted.push(row);
  }
  const counts = {};
  for (const row of accepted) counts[row.entity] = (counts[row.entity] || 0) + 1;
  const reviews = accepted.filter(r => r.entity === 'reviews');
  const reviewText = new Set(reviews.map(r => r.content_fingerprint));
  Object.assign(progress, {status: 'complete', accepted: accepted.length, counts,
    distinctReviewTexts: reviewText.size, excluded: excluded.length});
  save(path.join(work, 'library-' + topic.id + '.json'), {schema: 'ksf.five-topic.existing-library.v1',
    topic, generation: contract.generation, checkedAt: new Date().toISOString(), upstreamNewCollection: false,
    allPagesRetrieved: true, rawCandidates: totalMatched, counts,
    sourceIdsPresent: accepted.every(r => r.source_record_id && r.citation?.assetSha256),
    distinctReviewTexts: reviewText.size, records: accepted, excluded,
    limits: ['字面关键词候选；不证明功效、收入阶层或真实购买人群',
      '评价仅计正文或追评命中，商品标题带场景词不当成评价证言',
      '社媒与电商分开，不将相同原文记录当作独立消费者',
      '中式养生中的食材词不自动代表具有保健功效']});
  record();
  console.log(JSON.stringify({topic: topic.name, pages: page, counts, excluded: excluded.length}));
}
try {
  record();
  let next = 0;
  await Promise.all(Array.from({length: 2}, async () => {
    while (next < plan.topics.length) {const topic = plan.topics[next++]; await retrieve(topic);}
  }));
  const after = await read('contract');
  save(path.join(work, 'contract-after-library-retrieval.json'), after);
  if (after.generation !== contract.generation) throw Error('GENERATION_CHANGED');
  state.status = 'complete'; state.finishedAt = new Date().toISOString(); record();
} catch (error) {
  state.status = 'failed'; state.errorCode = /^[A-Z][A-Z0-9_:.-]+$/.test(error.message) ? error.message : 'TOPIC_READ_FAILED';
  record(); console.error(JSON.stringify({ok: false, errorCode: state.errorCode})); process.exitCode = 1;
}