xiaoshu-app/scripts/reconcile-legacy-lesson-pay.mjs

#!/usr/bin/env node
import { createHash, randomUUID } from 'node:crypto';
import { mkdir, writeFile } from 'node:fs/promises';
import { isAbsolute, join } from 'node:path';
import { pathToFileURL } from 'node:url';

function uniqueBy(rows, key) {
  const indexed = new Map();
  for (const row of rows) {
    const id = String(row[key] ?? '');
    indexed.set(id, indexed.has(id) ? null : row);
  }
  return indexed;
}

export function legacyLessonPayEvidence(lessons, appointments, coachId) {
  const orders = uniqueBy(appointments, 'GeneralID');
  const uniqueLessons = uniqueBy(lessons, 'yyds');
  const lessonIds = uniqueBy(lessons, 'GeneralID');
  const evidence = [];
  for (const [appointmentId, lesson] of uniqueLessons) {
    const order = orders.get(appointmentId);
    if (!lesson || !order || !lessonIds.get(String(lesson.GeneralID))) continue;
    const generalId = String(lesson.GeneralID), studentId = String(lesson.xymz ?? ''), classType = String(lesson.kclx ?? '');
    const amount = String(order.yongjin ?? '').trim(), completedAt = String(lesson.plpjsj ?? '').trim();
    if (![generalId, appointmentId, studentId].every(id => /^[1-9]\d*$/.test(id))) continue;
    if (String(lesson.jsmz) !== String(coachId) || String(order.pl) !== String(coachId) || String(order.szyh) !== studentId) continue;
    if (!['30', '40'].includes(String(order.dszt)) || !['1', '2', '3'].includes(classType) || String(order.dslx) !== classType) continue;
    if (!/^\d{1,6}(?:\.\d{1,2})?$/.test(amount) || !completedAt) continue;
    evidence.push({ generalId, appointmentId, studentId, coachId: String(coachId), classType, amount: Number(amount), completedAt });
  }
  return evidence.sort((a, b) => Number(a.generalId) - Number(b.generalId));
}

const eligible = `WITH evidence AS (
  SELECT * FROM jsonb_to_recordset($2::jsonb) AS e("generalId" text, "appointmentId" text,
    "studentId" text, "coachId" text, "classType" text, amount numeric, "completedAt" text, version text)
), matched AS (
  SELECT l."objectId", l."updatedAt"::text AS version, e.amount, e."generalId", e."appointmentId",
    count(*) OVER (PARTITION BY e."generalId") AS matches
  FROM evidence e
  JOIN "CommonModel" c ON c.company=$1 AND c."modelId"=59 AND c."generalId"::text=e."generalId"
    AND c.status=99 AND c."sourceKey"='legacy:model:59:general:'||e."generalId"
  JOIN "LessonRecord" l ON l.company=c.company AND l.id=c."itemId" AND l."sourceKey"=c."sourceKey"
  JOIN "CommonModel" ac ON ac.company=c.company AND ac."modelId"=54 AND ac."generalId"::text=e."appointmentId"
    AND ac.status=99 AND ac."sourceKey"='legacy:model:54:general:'||e."appointmentId"
  JOIN "CourseAppointment" a ON a.company=ac.company AND a.id=ac."itemId" AND a."sourceKey"=ac."sourceKey"
  WHERE l.xymz::text=e."studentId" AND l.jsmz::text=e."coachId" AND l.yyds::text=e."appointmentId"
    AND l.kclx::text=e."classType" AND l.plpjsj=e."completedAt" AND a.szyh::text=e."studentId" AND a.pl::text=e."coachId"
    AND l."teacherPaySnapshot" IS NULL AND a."teacherPaySnapshot" IS NULL
    AND (e.version IS NULL OR e.version=l."updatedAt"::text)
    AND NOT EXISTS (SELECT 1 FROM "PayrollLine" p WHERE p.company=l.company AND p.kind='lesson'
      AND (p."standardLessonKey"='legacy:'||e."generalId" OR p."lessonId"=l."objectId"))
)`;

export const legacyLessonPayPreviewSql = eligible + `
 SELECT "objectId", version, amount, "generalId", "appointmentId" FROM matched WHERE matches=1 ORDER BY "generalId"`;

export const legacyLessonPayApplySql = eligible + `
 UPDATE "LessonRecord" l SET "teacherPaySnapshot"=m.amount, "updatedAt"=now()
 FROM matched m WHERE m.matches=1 AND l.company=$1 AND l."objectId"=m."objectId"
   AND l."updatedAt"::text=m.version AND l."teacherPaySnapshot" IS NULL
 RETURNING l."objectId", l."teacherPaySnapshot" AS amount, m."generalId", m."appointmentId"`;

async function main() {
  const opt = key => process.argv.find(arg => arg.startsWith(key + '='))?.slice(key.length + 1);
  const coachId = Number(opt('--coach-id')), out = opt('--output');
  const applyHash = opt('--apply-plan'), token = process.env.XIAOSHU_SESSION_TOKEN, master = process.env.XIAOSHU_MASTER_KEY;
  if (!Number.isSafeInteger(coachId) || coachId <= 0 || !out || !isAbsolute(out) || !token) {
    throw Error('需要 --coach-id、--output=私有绝对路径及 XIAOSHU_SESSION_TOKEN');
  }
  if (applyHash && !master) throw Error('补记课酬需要 XIAOSHU_MASTER_KEY');
  const base = (process.env.XIAOSHU_PARSE_URL || 'https://server.xiaoshu.pro/parse').replace(/\/$/, '');
  const headers = { 'Content-Type': 'application/json', 'X-Parse-Application-Id': process.env.XIAOSHU_PARSE_APP_ID || '7pIbDBJmKx_main', 'X-Parse-Session-Token': token };
  async function parse(path, method = 'GET', body, privileged = false) {
    const r = await fetch(base + path, { method, headers: privileged ? { ...headers, 'X-Parse-Master-Key': master } : headers,
      body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(30000) });
    if (!r.ok) throw Error('Parse 请求失败 ' + r.status);
    return r.json();
  }
  const user = await parse('/users/me');
  const company = user.company?.objectId;
  if (!company || user.isDisabled || user.isDeleted || !['super-admin', 'ops-manager'].includes(user.adminRoleKey)) throw Error('需要有效的帐套管理员');
  const config = (await parse('/config')).params;
  async function source(params) {
    const rows = [];
    let expected;
    for (let cpage = 1; ; cpage++) {
      const url = new URL(config.legacyScheduleApiUrl);
      for (const [key, value] of Object.entries({ ...params, action: 'content_list', cpage, psize: 1000, apiId: config.legacyScheduleApiId, apiKey: config.legacyScheduleApiKey })) url.searchParams.set(key, value);
      const r = await fetch(url, { signal: AbortSignal.timeout(30000) });
      if (!r.ok) throw Error('旧系统课酬来源不可用');
      const j = await r.json(), items = typeof j.result === 'string' ? JSON.parse(j.result) : j.result;
      if (j.retcode < 0 || !Array.isArray(items) || !Number.isSafeInteger(Number(j.page?.itemCount))) throw Error('旧系统返回不完整');
      expected ??= Number(j.page.itemCount);
      if (expected !== Number(j.page.itemCount)) throw Error('旧系统分页过程中数据已变动');
      rows.push(...items);
      if (cpage >= Number(j.page.pageCount || 1)) break;
    }
    if (rows.length !== expected) throw Error('旧系统分页不完整');
    return rows;
  }
  const [lessons, appointments] = await Promise.all([
    source({ modelId: 59, myfield2: 'jsmz=' + coachId }),
    source({ uid: coachId, nid: 29, myfield2: 'pl=' + coachId, orders: 'B.yysj=DESC', dszt: 30, xiaokejilu: 1 }),
  ]);
  const evidence = legacyLessonPayEvidence(lessons, appointments, coachId);
  const plan = { company, coachId, evidence };
  const sha256 = createHash('sha256').update(JSON.stringify(plan)).digest('hex');
  await mkdir(out, { recursive: true, mode: 0o700 });
  const save = (name, value) => writeFile(join(out, name), JSON.stringify(value, null, 2), { mode: 0o600, flag: 'wx' });
  await save('source-plan-' + randomUUID() + '.json', { ...plan, sha256 });
  if (!applyHash) {
    console.log(JSON.stringify({ sha256, sourceLessons: lessons.length, sourceAppointments: appointments.length, matched: evidence.length, applied: false }));
    return;
  }
  if (applyHash !== sha256 || !evidence.length) throw Error('旧课酬来源已变化，请重新预览');
  const name = 'xiaoshu/system/legacy-lesson-pay-' + randomUUID(), expires = Date.now() + 5 * 60000;
  const code = `async function handler(request,response){try{
    const current=request.user;
    if(!current||current.id!==${JSON.stringify(user.objectId)}||Date.now()>${expires})return response.status(403).json({success:false});
    await current.fetch({useMasterKey:true});
    const co=${JSON.stringify(company)};
    if(current.get('company')?.id!==co||current.get('isDisabled')||current.get('isDeleted')||!['super-admin','ops-manager'].includes(current.get('adminRoleKey')))return response.status(403).json({success:false});
    const evidence=${JSON.stringify(evidence)};
    const p=request.body.params||{};
    const data=await Psql.tx(async t=>{
      await t.none("SET LOCAL lock_timeout='3s'");
      await t.none("SET LOCAL statement_timeout='45s'");
      await t.one('SELECT pg_advisory_xact_lock(hashtext($1))',['direct-sync:'+co]);
      await t.one('SELECT pg_advisory_xact_lock(hashtext($1))',['xs-payroll:'+co]);
      const active=await t.one('SELECT count(*)::int n FROM "LegacyDirectRun" WHERE company=$1 AND status=$2 AND "leaseUntil">now()',[co,'running']);
      if(active.n)throw Error('夜间同步正在运行');
      const found=await t.query(${JSON.stringify(legacyLessonPayPreviewSql)},[co,JSON.stringify(evidence)]);
      if(p.mode==='preview')return found;
      if(p.mode!=='apply'||JSON.stringify(p.expected)!==JSON.stringify(found))throw Error('待补课次或版本已改变，请重新预览');
      const versions=new Map(found.map(row=>[row.generalId,row.version]));
      const ready=evidence.filter(row=>versions.has(row.generalId)).map(row=>({...row,version:versions.get(row.generalId)}));
      const changed=await t.query(${JSON.stringify(legacyLessonPayApplySql)},[co,JSON.stringify(ready)]);
      if(changed.length!==found.length)throw Error('课酬补记数量变化，事务已回滚');
      for(const row of changed)await t.one('SELECT xs_payroll_audit($1,$2,$3,$4,$5,$6::jsonb,$7::jsonb)',[co,current.id,'restore-legacy-lesson-pay',row.objectId,'核对旧销课接口原金额，来源计划 ${sha256}',JSON.stringify({teacherPaySnapshot:null}),JSON.stringify(row)]);
      return changed;
    });return response.json({success:true,data});
  }catch(error){return response.status(409).json({success:false,message:String(error.message)});}}`;
  let id;
  async function invoke(params) {
    const r = await fetch(base.replace(/\/parse$/, '/api/functions/') + name, { method: 'POST', headers, body: JSON.stringify({ token, params }), signal: AbortSignal.timeout(60000) });
    const j = await r.json();
    if (!r.ok || !j.success) throw Error(j.message || '课酬核对失败');
    return j.data;
  }
  try {
    id = (await parse('/classes/Function', 'POST', { name, path: name, code, type: 'standalone', enabled: true, params: [], paramList: [], respType: 'json' }, true)).objectId;
    const expected = await invoke({ mode: 'preview' });
    await save('before-' + sha256 + '.json', expected);
    const changed = await invoke({ mode: 'apply', expected });
    await save('receipt-' + sha256 + '.json', changed);
    if ((await invoke({ mode: 'preview' })).length !== 0) throw Error('应用后仍有符合条件的未补课次');
    console.log(JSON.stringify({ sha256, updated: changed.length, applied: true }));
  } finally {
    if (id) await parse('/classes/Function/' + id, 'DELETE', undefined, true);
  }
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
  main().catch(error => { console.error(error.message); process.exitCode = 1; });
}