kangshifu-board/scripts/collect-topic-research.cjs

/** Catalog-driven topical collection with preserved HTTP bytes and a single-run lock. */
const fs = require('node:fs');
const path = require('node:path');
const crypto = require('node:crypto');
const {createRequire} = require('node:module');
const {AsyncLocalStorage} = require('node:async_hooks');
const fromVoc = createRequire('C:/Users/JSHY/Doubao/skills/voc-intelligence/package.json');
const {searchVocApis, getVocApiDoc} = fromVoc('./mcp/src/tools/voc-api-catalog-run.js');
const {readCredentialContext} = fromVoc('./mcp/src/core/credentials.js');
const gateways = {social: fromVoc('./mcp/src/providers/voc-gateway.js').callSocialGateway,
  ecommerce: fromVoc('./mcp/src/providers/ecommerce-gateway.js').callEcommerceGateway};
const planFile = path.resolve(process.argv[2] || '');
const plan = JSON.parse(fs.readFileSync(planFile, 'utf8'));
const directory = path.dirname(planFile), rawRoot = path.join(directory, 'collected');
fs.mkdirSync(rawRoot, {recursive: true});
const sha = bytes => crypto.createHash('sha256').update(bytes).digest('hex');
const save = (file, data) => {
  const next = file + '.next';
  fs.writeFileSync(next, JSON.stringify(data, null, 2));
  fs.renameSync(next, file);
};
const lockFile = path.join(directory, 'collector.lock.json');
if (fs.existsSync(lockFile)) {
  const old = JSON.parse(fs.readFileSync(lockFile, 'utf8'));
  let alive = false; try { process.kill(old.pid, 0); alive = true; } catch {}
  if (alive) throw Error('TOPIC_COLLECTOR_ALREADY_RUNNING');
  fs.renameSync(lockFile, lockFile + '.stale-' + Date.now());
}
fs.writeFileSync(lockFile, JSON.stringify({pid: process.pid, startedAt: new Date().toISOString()}), {flag: 'wx'});
const context = new AsyncLocalStorage(), nativeFetch = global.fetch;
global.fetch = async (url, init = {}) => {
  const response = await nativeFetch(url, {...init, signal: init.signal || AbortSignal.timeout(60000)});
  const current = context.getStore();
  if (current) {
    const bytes = Buffer.from(await response.clone().arrayBuffer());
    if (bytes.length > 45000000) throw Error('RESPONSE_TOO_LARGE');
    const file = current.key + '--attempt-' + (current.attempts.length + 1) + '.body';
    fs.writeFileSync(path.join(rawRoot, file), bytes);
    current.attempts.push({responseFile: file, httpStatus: response.status,
      bytes: bytes.length, sha256: sha(bytes), receivedAt: new Date().toISOString()});
    current.lastBytes = bytes;
  }
  return response;
};
(async () => {
  const credentials = readCredentialContext({});
  if (!credentials.newApiToken && !credentials.sessionToken) throw Error('VOC_CREDENTIAL_REQUIRED');
  const docs = new Map(), records = [];
  for (const platform of [...new Set(plan.jobs.map(j => j.platform))]) {
    const catalog = await searchVocApis({platform, query: '搜索', limit: 30});
    save(path.join(directory, 'catalog-' + platform + '.json'), catalog);
  }
  for (const id of [...new Set(plan.jobs.map(j => j.endpoint))]) {
    const doc = await getVocApiDoc({id});
    if (doc.status !== 'ok') throw Error('ENDPOINT_DOCUMENT_REQUIRED');
    docs.set(id, doc.data.endpoint);
    save(path.join(directory, 'endpoint-' + id.replace(/[^a-z0-9_-]/gi, '_') + '.json'), doc);
  }
  let stopReason = null;
  for (let index = 0; index < plan.jobs.length; index++) {
    const job = plan.jobs[index], endpoint = docs.get(job.endpoint);
    const key = String(index + 1).padStart(3, '0') + '-' + job.topic + '-' + job.platform + '-' + Date.now();
    const query = {}, body = {};
    for (const p of endpoint.params || []) {
      const value = job.params[p.name] ?? p.default;
      if (value !== undefined) ((p.in || endpoint.paramsIn || (endpoint.method === 'GET' ? 'query' : 'body')) === 'query' ? query : body)[p.name] = value;
      if (p.required && value === undefined) throw Error('REQUIRED_PARAMETER_MISSING:' + p.name);
    }
    const request = {key, attempts: [], lastBytes: null};
    const receipt = {...job, startedAt: new Date().toISOString(), productionWrites: 0};
    try {
      const result = await context.run(request, () => gateways[endpoint.channel || 'social']({
        proxyPath: endpoint.proxyPath, method: endpoint.method, query,
        body: endpoint.method === 'GET' ? undefined : body,
        token: credentials.newApiToken || credentials.sessionToken, fallbackToken: credentials.sessionToken,
        retries: 1
      }));
      if (!request.lastBytes) throw Error('RAW_HTTP_RESPONSE_REQUIRED');
      const file = key + '.response.json';
      fs.writeFileSync(path.join(rawRoot, file), request.lastBytes);
      Object.assign(receipt, {ok: true, httpStatus: result.httpStatus, responseFile: file,
        sha256: sha(request.lastBytes), bytes: request.lastBytes.length});
    } catch (error) {
      Object.assign(receipt, {ok: false, httpStatus: error.httpStatus || null,
        errorKind: error.kind || 'request', reason: error.reason || null, retryable: error.retryable ?? false});
      if ([401, 402, 403].includes(error.httpStatus)) stopReason = {code: 'UPSTREAM_HTTP_' + error.httpStatus,
        endpoint: job.endpoint, topic: job.topic, stoppedAt: new Date().toISOString()};
    }
    Object.assign(receipt, {finishedAt: new Date().toISOString(), attempts: request.attempts});
    save(path.join(rawRoot, key + '.meta.json'), receipt);
    records.push(receipt);
    save(path.join(directory, 'collection-result.json'), {schema: 'ksf.five-topic.collection.v1',
      updatedAt: new Date().toISOString(), status: stopReason ? 'stopped-upstream' : 'running',
      plannedJobs: plan.jobs.length, completedJobs: records.length, stopReason, records, productionWrites: 0});
    console.log(JSON.stringify({topic: job.topic, platform: job.platform, endpoint: job.endpoint,
      ok: receipt.ok, httpStatus: receipt.httpStatus, completed: records.length, total: plan.jobs.length}));
    if (stopReason) break;
  }
  save(path.join(directory, 'collection-result.json'), {schema: 'ksf.five-topic.collection.v1',
    updatedAt: new Date().toISOString(), status: stopReason ? 'stopped-upstream' : 'completed',
    plannedJobs: plan.jobs.length, completedJobs: records.length, stopReason, records, productionWrites: 0});
  if (stopReason) process.exitCode = 2;
})().catch(error => {
  console.error(JSON.stringify({ok: false, errorCode: /^[A-Z][A-Z0-9_:.-]+$/.test(error.message) ? error.message : 'TOPIC_COLLECTION_FAILED'}));
  process.exitCode = 1;
}).finally(() => {
  if (fs.existsSync(lockFile) && JSON.parse(fs.readFileSync(lockFile, 'utf8')).pid === process.pid) fs.unlinkSync(lockFile);
});