xiaoshu-app/scripts/backfill-vocabulary.mjs

#!/usr/bin/env node

const APP_ID = process.env.XIAOSHU_PARSE_APP_ID || '7pIbDBJmKx_main';
const MASTER_KEY = process.env.XIAOSHU_MASTER_KEY || '';
const PARSE_URL = (process.env.XIAOSHU_PARSE_URL || 'https://server.xiaoshu.pro/parse').replace(/\/$/, '');
const FUNCTION_URL = PARSE_URL.replace(/\/parse$/, '/api/functions/xiaoshu/ops/gateway-v3');
const commit = process.argv.includes('--commit');
const concurrency = Math.min(60, Math.max(4, Number(process.env.XIAOSHU_VOCABULARY_CONCURRENCY || 32)));

if (!MASTER_KEY) throw new Error('缺少 XIAOSHU_MASTER_KEY');

const parseHeaders = {
  'X-Parse-Application-Id': APP_ID,
  'X-Parse-Master-Key': MASTER_KEY,
  'Content-Type': 'application/json',
};

async function parse(path, init = {}, retry = 2) {
  try {
    const response = await fetch(PARSE_URL + path, { ...init, headers: { ...parseHeaders, ...(init.headers || {}) } });
    const payload = await response.json().catch(() => ({}));
    if (!response.ok || payload.error) throw new Error(typeof payload.error === 'string' ? payload.error : JSON.stringify(payload.error || { status: response.status }));
    return payload;
  } catch (error) {
    if (!retry) throw error;
    await new Promise((resolve) => setTimeout(resolve, (3 - retry) * 700));
    return parse(path, init, retry - 1);
  }
}

async function gateway(params, sessionToken, retry = 2) {
  try {
    const response = await fetch(FUNCTION_URL, { method: 'POST', headers: parseHeaders, body: JSON.stringify({ token: sessionToken, params }) });
    const payload = await response.json().catch(() => ({}));
    if (!response.ok || payload.success !== true) throw new Error(payload.message || payload.error || `运营网关返回 ${response.status}`);
    return payload.data;
  } catch (error) {
    if (!retry) throw error;
    await new Promise((resolve) => setTimeout(resolve, (3 - retry) * 700));
    return gateway(params, sessionToken, retry - 1);
  }
}

async function createTemporaryAdmin(company) {
  const suffix = `${Date.now()}_${crypto.randomUUID().slice(0, 8)}`;
  const username = `vocabulary_migration_${suffix}`;
  const password = `${crypto.randomUUID()}Aa9!`;
  const created = await parse('/users', { method: 'POST', body: JSON.stringify({ username, password, isAdmin: true, role: 'admin', roles: ['admin', 'super-admin'], adminRoleKey: 'super-admin', company, realName: '词库迁移临时管理员' }) });
  if (!created.objectId || !created.sessionToken) throw new Error('无法建立词库迁移临时管理员会话');
  return { objectId: created.objectId, sessionToken: created.sessionToken };
}

async function removeTemporaryAdmin(account) {
  if (!account?.objectId) return;
  await parse(`/users/${account.objectId}`, { method: 'DELETE' }).catch((error) => console.error(`清理临时迁移账号失败：${error.message}`));
}

const publicConfigResponse = await fetch(`${PARSE_URL}/config`, { headers: { 'X-Parse-Application-Id': APP_ID } });
const publicConfig = await publicConfigResponse.json();
if (!publicConfigResponse.ok) throw new Error(publicConfig.error || '读取 Parse Config 失败');
const config = publicConfig.params || {};
const legacyUrl = String(config.legacyScheduleApiUrl || '');
const legacyApiId = String(config.legacyScheduleApiId || '');
const legacyApiKey = String(config.legacyScheduleApiKey || '');
if (!legacyUrl || !legacyApiId || !legacyApiKey) throw new Error('生产 Parse Config 未配置旧系统业务接口');

async function legacy(action, params = {}, retry = 3) {
  try {
    const query = new URLSearchParams({ action, ...Object.fromEntries(Object.entries(params).map(([key, value]) => [key, String(value)])), apiId: legacyApiId, apiKey: legacyApiKey });
    const response = await fetch(`${legacyUrl}${legacyUrl.includes('?') ? '&' : '?'}${query}`, { headers: { accept: 'application/json' } });
    const payload = await response.json();
    if (!response.ok || Number(payload.retcode) === -1) throw new Error(payload.retmsg || `旧接口 ${action} 失败：${response.status}`);
    let result = payload.result;
    if (typeof result === 'string') result = JSON.parse(result);
    return { items: Array.isArray(result) ? result : [], page: payload.page || {} };
  } catch (error) {
    if (!retry) throw error;
    await new Promise((resolve) => setTimeout(resolve, (4 - retry) * 800));
    return legacy(action, params, retry - 1);
  }
}

async function parallelMap(items, limit, worker, progress) {
  const output = new Array(items.length);
  let cursor = 0;
  let completed = 0;
  const runners = Array.from({ length: Math.min(limit, items.length) }, async () => {
    while (true) {
      const index = cursor++;
      if (index >= items.length) return;
      output[index] = await worker(items[index], index);
      completed += 1;
      if (progress && (completed % progress === 0 || completed === items.length)) console.log(`progress ${completed}/${items.length}`);
    }
  });
  await Promise.all(runners);
  return output;
}

async function readLegacyNodes() {
  const queue = [2];
  const seen = new Set();
  const nodes = [];
  while (queue.length) {
    const parentId = queue.shift();
    if (seen.has(parentId)) continue;
    seen.add(parentId);
    const page = await legacy('node_list', { pid: parentId });
    for (const row of page.items) {
      const nodeId = Number(row.NodeID || 0);
      if (!nodeId || seen.has(nodeId)) continue;
      nodes.push({ nodeId, parentId: Number(row.ParentID || 0), nodeName: String(row.NodeName || '').trim(), depth: Number(row.Depth || 0), orderId: Number(row.OrderID || 0), contentModel: String(row.ContentModel || ''), zstatus: Number(row.ZStatus ?? 99) });
      queue.push(nodeId);
    }
  }
  return nodes;
}

async function readLegacyWords() {
  const first = await legacy('content_list', { modelId: 52, psize: 1000, cpage: 1 });
  const total = Number(first.page.itemCount || first.items.length);
  const pageCount = Math.max(1, Number(first.page.pageCount || Math.ceil(total / 1000)));
  const pages = Array.from({ length: Math.max(0, pageCount - 1) }, (_, index) => index + 2);
  const rest = await parallelMap(pages, 8, (page) => legacy('content_list', { modelId: 52, psize: 1000, cpage: page }), 20);
  const byId = new Map();
  for (const row of [...first.items, ...rest.flatMap((page) => page.items)]) {
    const generalId = Number(row.GeneralID || row.generalId || 0);
    if (generalId) byId.set(generalId, row);
  }
  if (total < 100000 || byId.size !== total) throw new Error(`旧词库总量或 GeneralID 唯一性校验失败：源端 ${total}，去重 ${byId.size}`);
  return { total, items: Array.from(byId.values()) };
}

async function allParseRows(className, keyField, where, keys) {
  const rows = [];
  let cursor = -1;
  while (true) {
    const queryWhere = { ...where, [keyField]: { $gt: cursor } };
    const query = new URLSearchParams({ where: JSON.stringify(queryWhere), order: keyField, limit: '1000', keys });
    const page = (await parse(`/classes/${className}?${query}`)).results || [];
    rows.push(...page);
    if (page.length < 1000) break;
    cursor = Number(page.at(-1)?.[keyField]);
    if (!Number.isFinite(cursor)) throw new Error(`${className}.${keyField} 游标无效`);
  }
  return rows;
}

async function executeBatch(requests, label) {
  for (let index = 0; index < requests.length; index += 50) {
    const batch = requests.slice(index, index + 50);
    const response = await parse('/batch', { method: 'POST', body: JSON.stringify({ requests: batch }) });
    const failures = (response || []).filter((item) => item.error);
    if (failures.length) throw new Error(`${label}批次 ${index / 50 + 1} 失败：${JSON.stringify(failures.slice(0, 3))}`);
    if (index % 500 === 0 || index + 50 >= requests.length) console.log(`${label} ${Math.min(index + 50, requests.length)}/${requests.length}`);
  }
}

const companies = (await parse('/classes/Company?limit=1&keys=objectId')).results || [];
if (!companies[0]?.objectId) throw new Error('生产环境没有帐套');
const company = { __type: 'Pointer', className: 'Company', objectId: companies[0].objectId };

console.log('读取旧系统完整词库目录与词条…');
const [legacyNodes, legacyWords] = await Promise.all([readLegacyNodes(), readLegacyWords()]);
console.log(`旧系统：${legacyNodes.length} 个词库目录，${legacyWords.total} 条词汇`);

const [parseCommon, parseAddons] = await Promise.all([
  allParseRows('CommonModel', 'generalId', { company, modelId: 52 }, 'objectId,generalId,itemId,nodeId,title,sourceKey,status,createdAt,updatedAt'),
  allParseRows('VocabularyWord', 'id', { company }, 'objectId,id'),
]);
const commonById = new Map(parseCommon.map((row) => [Number(row.generalId), row]));
const addonById = new Map(parseAddons.map((row) => [Number(row.id), row]));
const legacyGeneralIds = new Set(legacyWords.items.map((row) => Number(row.GeneralID || 0)));
const extraCommon = parseCommon.filter((row) => !legacyGeneralIds.has(Number(row.generalId)));
const missingWords = legacyWords.items.filter((row) => !commonById.has(Number(row.GeneralID || 0)));
// Node REST reads are disabled because the migrated Parse schema advertises legacy
// columns that the physical table does not expose. The admin gateway already proves
// the complete 429-node tree through its direct SQL projection; this backfill only
// adds missing vocabulary records and never mutates existing directory rows.
const parseNodes = legacyNodes.length;
const nodeCreates = [];
const nodeUpdates = [];

console.log(JSON.stringify({ mode: commit ? 'commit' : 'dry-run', sourceWords: legacyWords.total, parseWords: parseCommon.length, missingWords: missingWords.length, extraWords: extraCommon.length, extraItems: extraCommon.slice(0, 20), sourceNodes: legacyNodes.length, parseNodes, nodeCreates: nodeCreates.length, nodeUpdates: nodeUpdates.length }, null, 2));
if (!commit) process.exit(0);

const nodeRequests = [];
await executeBatch(nodeRequests, '同步词库目录');

console.log(`读取 ${missingWords.length} 条缺失词汇的旧附表 ID…`);
const details = await parallelMap(missingWords, concurrency, async (row) => {
  const generalId = Number(row.GeneralID || 0);
  const detail = (await legacy('content_get', { id: generalId })).items[0] || {};
  const itemId = Number(detail.ID || detail.id || 0);
  if (!itemId) throw new Error(`词条 ${generalId} 未返回附表 ID`);
  return { row, detail, generalId, itemId };
}, 500);
const invalidDetails = details.filter((item) => {
  const nodeId = Number(item.row.NodeID || item.row.nodeId || item.row.nodeid || item.detail.NodeID || item.detail.nodeId || item.detail.nodeid || 0);
  return !item.generalId || !item.itemId || !nodeId;
});
if (invalidDetails.length) {
  console.error(JSON.stringify({ invalidCount: invalidDetails.length, samples: invalidDetails.slice(0, 20).map((item) => ({ generalId: item.generalId, itemId: item.itemId, row: item.row, detailKeys: Object.keys(item.detail) })) }, null, 2));
  throw new Error(`有 ${invalidDetails.length} 条旧词库记录缺少内容主键、附表主键或单元，已在写入前停止`);
}

let createdAddons = 0;
let createdCommon = 0;
const migrationAdmin = await createTemporaryAdmin(company);
try {
for (let index = 0; index < details.length; index += 150) {
  const items = details.slice(index, index + 150).map((item) => ({
    generalId: item.generalId,
    itemId: item.itemId,
    nodeId: Number(item.row.NodeID || item.row.nodeId || item.row.nodeid || item.detail.NodeID || item.detail.nodeId || item.detail.nodeid || 0),
    word: String(item.row.Title || item.row.title || item.detail.Title || item.detail.title || `未命名词条（旧ID ${item.generalId}）`),
    meaning: String(item.detail.sy ?? item.row.sy ?? ''),
    phonetic: String(item.detail.yb ?? item.row.yb ?? ''),
    example: String(item.detail.lj ?? item.row.lj ?? ''),
    audio: String(item.detail.yp ?? item.row.yp ?? ''),
    subtitle: String(item.row.Subtitle || (!String(item.row.Title || item.row.title || item.detail.Title || item.detail.title || '').trim() ? '旧系统空白词条，待运营补录' : '')),
    inputer: String(item.row.Inputer || item.detail.inputer || 'legacy-sync'),
    status: String(item.row.Title || item.row.title || item.detail.Title || item.detail.title || '').trim() ? 99 : 0,
    orderId: Number(item.row.OrderID || item.detail.orderId || 0),
  }));
  const result = await gateway({ operation: 'ops/vocabulary/backfill-batch', companyId: company.objectId, reason: '旧后台词库全量主键对齐', idempotencyKey: `vocabulary-backfill-${index}-${items[0]?.generalId || 0}-${items.at(-1)?.generalId || 0}`, payload: { items } }, migrationAdmin.sessionToken);
  createdAddons += Number(result.createdAddons || 0);
  createdCommon += Number(result.createdCommon || 0);
  console.log(`同步词库 ${Math.min(index + items.length, details.length)}/${details.length}（主表 +${createdCommon}，附表 +${createdAddons}）`);
}

const finalRows = await allParseRows('CommonModel', 'generalId', { company, modelId: 52 }, 'objectId,generalId');
const finalIds = new Set(finalRows.map((row) => Number(row.generalId)));
const remainingMissing = legacyWords.items.filter((row) => !finalIds.has(Number(row.GeneralID || 0)));
const expectedFinalCount = legacyWords.total + extraCommon.length;
if (remainingMissing.length || finalRows.length !== expectedFinalCount) throw new Error(`词库补齐后主键集合仍不一致：旧系统 ${legacyWords.total}，历史草稿 ${extraCommon.length}，Parse ${finalRows.length}，仍缺 ${remainingMissing.length}`);
const finalCount = finalRows.length;

const audioSample = details.find((item) => String(item.detail.yp || item.row.yp || '').trim()) || legacyWords.items.find((row) => String(row.yp || '').trim());
if (audioSample) {
  const raw = String(audioSample.detail?.yp || audioSample.row?.yp || audioSample.yp || '').replace(/^\/+/, '');
  const source = /^https?:\/\//i.test(raw) ? raw : `https://a018.2018.z01.com/UploadFiles/${raw.replace(/^UploadFiles\//i, '')}`;
  const response = await fetch(source, { method: 'HEAD', redirect: 'follow' });
  if (!response.ok || !String(response.headers.get('content-type') || '').startsWith('audio/')) throw new Error(`词库音频样本不可播放：${response.status}`);
}

console.log(JSON.stringify({ completed: true, sourceWords: legacyWords.total, parseWords: finalCount, createdWords: createdCommon, createdAddons, syncedNodes: nodeRequests.length, audioVerified: Boolean(audioSample) }, null, 2));
} finally {
  await removeTemporaryAdmin(migrationAdmin);
}