hyreal-gen-pic/server/services/aitable.js

'use strict';

// 钉钉 AI 表格（原多维表）同步服务 · 三表单粒度架构
// ----------------------------------------------------------------------------
// 双层架构的云端层：AI 表格 = 在线主档案 + 多人共享层；本地数据湖 = 计算缓存层。
// 云端按维度拆三张表（每张表内粒度唯一，AI/人工任意聚合都不会跨维度算重）：
//   · 门店明细（scope=shop）  · 连锁汇总（scope=chain）  · 公司汇总（scope=company）
//   · scope=summary 的区间汇总行不上云（可由门店行推导）
//   · pushLake — 把本地湖的新增/更新行（按 syncedAt 水位）分流 upsert 到三张表
//   · pullLake — 把三张表记录拉回本地湖（其他机器补数据用，云端为准）
// 业务键：日期|账号名|维度|连锁ID|门店ID（连锁ID 参与键：连锁/汇总行 shopId 为空，
// 缺它会互相覆盖）。与本地 rowKey（date|scope|chainId|shopId）语义对齐，幂等可重跑。
//
// 钉钉侧契约（企业内部应用 access_token，域 api.dingtalk.com；2026-10-17 真机核对）：
//   token:   POST /v1.0/oauth2/accessToken  {appKey, appSecret} → {accessToken, expireIn}
//   表清单:  GET  /v1.0/notable/bases/{baseId}/sheets?operatorId → {value:[{id,name}]}
//   建表:    POST /v1.0/notable/bases/{baseId}/sheets?operatorId  {name, columns:[{name,type}]}
//   建列:    POST …/sheets/{sheetIdOrName}/fields?operatorId  {name, type:'text'|'number'}
//   列清单:  GET  …/sheets/{sheetIdOrName}/fields?operatorId → {value:[{name,type,property}]}
//   列记录:  GET  …/records?operatorId&maxResults&nextToken → {records:[{id,fields}],nextToken,hasMore}
//   新增:    POST …/records {records:[{fields}]} → {value:[{id}]}（响应键是 value）
//   更新:    PUT  …/records {records:[{id,fields}]}（记录标识键是 id，不是 record_id）
//   删除:    POST …/records/delete {recordIds:[…]}（同步语义只 upsert，用不到）
//   · 所有 notable 接口 query 必带 operatorId=操作人 unionId（scripts/aitable-discover.js 发现）
//   · 数值字段读回是字符串（"12345.67"），recordToRow 已做 Number() 归一
//   · number 列 formatter=INT 只影响显示，存储小数保真
//   · 建表/建列接口有应用级限流（突发 ~5 次后 500 internalError，慢速补列即可），
//     记录读写接口不受影响；记录写入遇到不存在的字段名报 404 notFound
// ⚠ 批量上限（100）是保守值，首推大批量时如遇限流再调。

const fs = require('fs');
const path = require('path');
const https = require('https');

const API_BASE = 'https://api.dingtalk.com';
const TOKEN_PATH = '/v1.0/oauth2/accessToken';
const RECORDS_PATH = (baseId, sheetName) => `/v1.0/notable/bases/${encodeURIComponent(baseId)}/sheets/${encodeURIComponent(sheetName)}/records`;
const INSERT_CHUNK = 100;      // 单批新增上限（保守值，真机核对后可调）
const UPDATE_CHUNK = 100;
const LIST_PAGE = 100;

// 三张表的配置键 → 默认表名（读取处兜底，老 config.json 缺键零迁移）
const SHEET_DEFAULTS = { sheetDetail: '门店明细', sheetChain: '连锁汇总', sheetCompany: '公司汇总' };

// 本地字段 → AI 表格列名（中文列头，人工可读可筛；三张表同一套列，缺维度的列留空）
const FIELD_COLUMNS = {
  date: '日期',
  account: '账号名',
  dimension: '维度',
  shopId: '门店ID',
  shopName: '门店名称',
  chainId: '连锁ID',
  chainName: '连锁名称',
  validOrderCnt: '有效订单',
  incomeAmt: '收入(元)',
  grossAmt: '营业额(元)',
  expenseAmt: '支出(元)',
  expUv: '曝光人数',
  orderUserCnt: '下单人数',
  clkExpRate: '进店转化率',
  ordClkRate: '下单转化率',
  perActualPay: '客单价(元)',
  actOrdCnt: '活动订单',
  actSubsidyAmt: '活动补贴(元)',
  shopRoi: '推广ROI',
  rebuyRate30d: '30日复购率',
  bizHours: '营业时长(h)',
  abnormalCloseHours: '异常关店时长(h)',
  goodsSaleAmt: '商品销售额(元)',
};
const METRIC_FIELDS = ['validOrderCnt', 'incomeAmt', 'grossAmt', 'expenseAmt', 'expUv', 'orderUserCnt',
  'clkExpRate', 'ordClkRate', 'perActualPay', 'actOrdCnt', 'actSubsidyAmt', 'shopRoi', 'rebuyRate30d',
  'bizHours', 'abnormalCloseHours', 'goodsSaleAmt'];
const DIMENSION_LABEL = { company: '公司', chain: '连锁', shop: '门店', summary: '汇总' };
// scope → 目标表配置键（summary 不上云）
const SCOPE_SHEET_KEY = { shop: 'sheetDetail', chain: 'sheetChain', company: 'sheetCompany' };

function rowToRecord(row, accountName) {
  const fields = {
    [FIELD_COLUMNS.date]: row.date || '',
    [FIELD_COLUMNS.account]: accountName,
    [FIELD_COLUMNS.dimension]: DIMENSION_LABEL[row.scope] || row.scope || '',
    [FIELD_COLUMNS.shopId]: String(row.shopId || ''),
    [FIELD_COLUMNS.shopName]: row.shopName || '',
    [FIELD_COLUMNS.chainId]: String(row.chainId || ''),
    [FIELD_COLUMNS.chainName]: row.chainName || '',
  };
  for (const f of METRIC_FIELDS) {
    const v = row[f];
    fields[FIELD_COLUMNS[f]] = v == null ? null : Number(v);
  }
  return { fields };
}

function recordKey(fields) {
  return [fields[FIELD_COLUMNS.date], fields[FIELD_COLUMNS.account], fields[FIELD_COLUMNS.dimension],
    String(fields[FIELD_COLUMNS.chainId] || ''), String(fields[FIELD_COLUMNS.shopId] || '')].join('|');
}

function recordToRow(record, defaultAccount) {
  const f = record.fields || {};
  const byName = (col) => {
    if (f[col] != null) return f[col];
    // 兼容按 fieldId 返回的形态：取 record.fields 中无法按名列出的场景交由真机联调修正
    return null;
  };
  const rev = {};
  for (const [k, col] of Object.entries(FIELD_COLUMNS)) rev[col] = k;
  const row = {
    date: String(byName(FIELD_COLUMNS.date) || '').replace(/-/g, ''),
    scope: Object.entries(DIMENSION_LABEL).find(([, label]) => label === byName(FIELD_COLUMNS.dimension))?.[0] || 'shop',
    shopId: String(byName(FIELD_COLUMNS.shopId) || ''),
    shopName: byName(FIELD_COLUMNS.shopName) || '',
    chainId: String(byName(FIELD_COLUMNS.chainId) || ''),
    chainName: byName(FIELD_COLUMNS.chainName) || '',
    account: byName(FIELD_COLUMNS.account) || defaultAccount || '',
  };
  for (const f2 of METRIC_FIELDS) {
    const v = byName(FIELD_COLUMNS[f2]);
    row[f2] = v == null || v === '' ? 0 : Number(v);
  }
  row.syncedAt = new Date().toISOString();
  return row;
}

/** 本地湖行分流到三张云端表（纯函数，smoke 直接回归）。
 *  summary 行不上云（可由门店行推导，且混入任一表都会造成表内算重）。 */
function splitRowsBySheet(rows) {
  const out = { detail: [], chain: [], company: [], skippedSummary: 0 };
  for (const row of Array.isArray(rows) ? rows : []) {
    const key = SCOPE_SHEET_KEY[row && row.scope];
    if (key === 'sheetDetail') out.detail.push(row);
    else if (key === 'sheetChain') out.chain.push(row);
    else if (key === 'sheetCompany') out.company.push(row);
    else out.skippedSummary += 1;
  }
  return out;
}

/** 解析三张表名（配置缺省用默认值）+ 表名指纹（水位失效判定用） */
function resolveSheets(cfg) {
  const sheets = {
    detail: cfg.sheetDetail || SHEET_DEFAULTS.sheetDetail,
    chain: cfg.sheetChain || SHEET_DEFAULTS.sheetChain,
    company: cfg.sheetCompany || SHEET_DEFAULTS.sheetCompany,
  };
  return { sheets, fingerprint: [sheets.detail, sheets.chain, sheets.company].join('|') };
}

/** 推送结果的一行摘要（日志与状态徽标共用）：合计 + 分表明细 */
function pushSummary(r) {
  const base = `新增 ${r.created} 行、更新 ${r.updated} 行`;
  const parts = Object.entries(r.perSheet || {})
    .filter(([, s]) => s.rows > 0)
    .map(([, s]) => `${s.sheet} 新增${s.created}/更新${s.updated}`)
    .join('，');
  return parts ? `${base}（${parts}）` : base;
}

function createAitableService(ctx) {
  const { config, workspace, logs } = ctx;

  // ---------- token（进程内缓存，提前 60s 过期） ----------

  let tokenCache = { token: null, expireAt: 0 };

  function transport(method, urlPath, bodyObj, token) {
    return new Promise((resolve, reject) => {
      const data = bodyObj !== undefined ? JSON.stringify(bodyObj) : null;
      const req = https.request({
        hostname: 'api.dingtalk.com', path: urlPath, method,
        headers: {
          'Content-Type': 'application/json',
          ...(token ? { 'x-acs-dingtalk-access-token': token } : {}),
          ...(data ? { 'Content-Length': Buffer.byteLength(data) } : {}),
        },
        timeout: 30000,
      }, (res) => {
        const chunks = [];
        res.on('data', (c) => chunks.push(c));
        res.on('end', () => {
          const raw = Buffer.concat(chunks).toString('utf8');
          let json = null;
          try { json = JSON.parse(raw); } catch { /* keep raw */ }
          resolve({ statusCode: res.statusCode, json, raw: raw.slice(0, 800) });
        });
      });
      req.on('error', reject);
      req.on('timeout', () => req.destroy(new Error('请求超时')));
      if (data) req.write(data);
      req.end();
    });
  }

  async function getToken() {
    const cfg = config.aitable || {};
    if (!cfg.appKey || !cfg.appSecret) throw err('未配置钉钉应用凭据（appKey/appSecret）', 'AITABLE_NO_CFG');
    const now = Date.now();
    if (tokenCache.token && now < tokenCache.expireAt) return tokenCache.token;
    const res = await transport('POST', TOKEN_PATH, { appKey: cfg.appKey, appSecret: cfg.appSecret });
    const token = res.json && (res.json.accessToken || (res.json.result && res.json.result.accessToken));
    const expireIn = Number((res.json && (res.json.expireIn ?? res.json.result?.expireIn)) || 7200);
    if (res.statusCode !== 200 || !token) {
      throw err(`获取 access_token 失败 HTTP ${res.statusCode}: ${(res.json && (res.json.message || res.json.errmsg)) || res.raw}`, 'AITABLE_AUTH_FAIL');
    }
    tokenCache = { token, expireAt: now + Math.max(60, expireIn - 60) * 1000 };
    return token;
  }

  function err(message, code) { const e = new Error(message); e.code = code; e.status = 400; return e; }

  // notable 接口统一拼 query：业务参数 + 必填 operatorId
  function notableQuery(cfg, params = {}) {
    const q = new URLSearchParams({ operatorId: cfg.operatorId || '', ...params });
    return `?${q.toString()}`;
  }

  function requireConfig() {
    const cfg = config.aitable || {};
    if (!cfg.enabled) throw err('钉钉 AI 表格同步未启用（系统设置中开启）', 'AITABLE_DISABLED');
    if (!cfg.baseId || !cfg.appKey) throw err('钉钉 AI 表格配置不完整（AppKey / Base ID）', 'AITABLE_NO_CFG');
    if (!cfg.operatorId) throw err('缺少操作者 unionId：先在钉钉开放平台给应用开通通讯录读权限，再运行 node scripts/aitable-discover.js 发现并填入', 'AITABLE_NO_OPERATOR');
    return cfg;
  }

  async function listAllRecords(cfg, sheetName) {
    const token = await getToken();
    const out = [];
    let nextToken = '';
    for (;;) {
      const q = notableQuery(cfg, { maxResults: LIST_PAGE, ...(nextToken ? { nextToken } : {}) });
      const res = await transport('GET', RECORDS_PATH(cfg.baseId, sheetName) + q, undefined, token);
      if (res.statusCode !== 200) throw err(`拉取「${sheetName}」记录失败 HTTP ${res.statusCode}: ${(res.json && res.json.message) || res.raw}`, 'AITABLE_LIST_FAIL');
      const d = res.json || {};
      const records = Array.isArray(d.records) ? d.records
        : Array.isArray(d.value) ? d.value
        : Array.isArray(d.data) ? d.data : [];
      out.push(...records);
      nextToken = d.nextToken || d.nextCursor || '';
      if (!nextToken) break;
    }
    return out;
  }

  function chunk(arr, n) {
    const out = [];
    for (let i = 0; i < arr.length; i += n) out.push(arr.slice(i, i + n));
    return out;
  }

  // ---------- 推送：本地湖 → AI 表格（三表） ----------

  const pushStateFile = () => path.join(workspace.getWorkspaceRoot(), '.genpic', 'aitable-push.json');

  function readPushState() {
    try { return JSON.parse(fs.readFileSync(pushStateFile(), 'utf8')); } catch { return {}; }
  }

  function writePushState(state) {
    try {
      const file = pushStateFile();
      fs.mkdirSync(path.dirname(file), { recursive: true });
      const temp = `${file}.${process.pid}.tmp`;
      fs.writeFileSync(temp, JSON.stringify(state, null, 2), 'utf8');
      fs.renameSync(temp, file);
    } catch { /* 状态落盘失败不阻断推送 */ }
  }

  /** 单张表的 upsert：diff 现有记录 → 分批新增/更新。返回 {created, updated, skippedStale}。
   *  冲突策略：同键行「数据新者胜」——云端记录的 lastModifiedTime 比 本地行 syncedAt
   *  （取数时间）新则跳过（skippedStale），避免另一台机器后推的旧数据盖掉云端新数据。 */
  async function pushSheet(cfg, token, sheetName, accountName, rows, opts) {
    if (!rows.length) return { created: 0, updated: 0, skippedStale: 0 };
    const existing = new Map();
    for (const rec of await listAllRecords(cfg, sheetName)) {
      const f = rec.fields || {};
      const k = recordKey({
        [FIELD_COLUMNS.date]: f[FIELD_COLUMNS.date],
        [FIELD_COLUMNS.account]: f[FIELD_COLUMNS.account],
        [FIELD_COLUMNS.dimension]: f[FIELD_COLUMNS.dimension],
        [FIELD_COLUMNS.chainId]: f[FIELD_COLUMNS.chainId],
        [FIELD_COLUMNS.shopId]: f[FIELD_COLUMNS.shopId],
      });
      if (rec.id) existing.set(k, { id: rec.id, mtime: Number(rec.lastModifiedTime) || 0 });
    }

    let created = 0;
    let updated = 0;
    let skippedStale = 0;
    const toCreate = [];
    const toUpdate = []; // {id, fields}（更新接口的记录标识键是 id，不是 record_id）
    for (const row of rows) {
      const rec = rowToRecord(row, accountName);
      const k = recordKey(rec.fields);
      const hit = existing.get(k);
      if (hit) {
        const localAt = new Date(row.syncedAt || 0).getTime();
        if (!opts.force && hit.mtime && localAt && hit.mtime > localAt) { skippedStale += 1; continue; }
        toUpdate.push({ id: hit.id, fields: rec.fields });
      } else {
        toCreate.push(rec);
      }
    }
    for (const part of chunk(toCreate, INSERT_CHUNK)) {
      const res = await transport('POST', RECORDS_PATH(cfg.baseId, sheetName) + notableQuery(cfg), { records: part }, token);
      if (res.statusCode >= 300) throw err(`「${sheetName}」批量新增失败 HTTP ${res.statusCode}: ${(res.json && res.json.message) || res.raw}`, 'AITABLE_CREATE_FAIL');
      created += part.length;
    }
    for (const part of chunk(toUpdate, UPDATE_CHUNK)) {
      const res = await transport('PUT', RECORDS_PATH(cfg.baseId, sheetName) + notableQuery(cfg), { records: part }, token);
      if (res.statusCode >= 300) throw err(`「${sheetName}」批量更新失败 HTTP ${res.statusCode}: ${(res.json && res.json.message) || res.raw}`, 'AITABLE_UPDATE_FAIL');
      updated += part.length;
    }
    return { created, updated, skippedStale };
  }

  /** 推送一个主体的数据到三张表。force=true 时忽略水位推全量、并跳过冲突/全零保护。
   *  水位按「主体 + 三表名指纹」记录：表名变化（如单表→三表迁移）自动降级为全量重推。 */
  async function pushLake(entity, opts = {}) {
    const cfg = requireConfig();
    const naposService = require('../../napos-service.js');
    const key = naposService.store.sanitizeKey(entity);
    const meta = naposService.store.getEntity(key);
    const accountName = meta ? meta.name : key;
    const { sheets, fingerprint } = resolveSheets(cfg);

    const state = readPushState();
    const prev = state[key] || {};
    const watermark = opts.force || prev.sheets !== fingerprint ? 0 : (prev.lastPushedAt || 0);
    let rows;
    if (watermark) {
      const all = naposService.store.loadRows(key) || [];
      rows = all.filter((r) => new Date(r.syncedAt || 0).getTime() > watermark);
    } else {
      rows = naposService.store.loadRows(key) || [];
    }
    const parts = splitRowsBySheet(rows);
    const pushed = parts.detail.length + parts.chain.length + parts.company.length;
    if (!pushed) return { pushed: 0, created: 0, updated: 0, skippedSummary: parts.skippedSummary, skipped: '无新增数据' };
    // 大批量（尤其首推全量）耗时较长，开始时先留痕，避免长时间静默
    logs.pushSystemLog(`推送云端开始：${accountName}（${watermark ? '增量' : '全量'} ${pushed} 行）`, 'info');

    // 全零保护：大批量门店行指标全零基本可断定取数异常（已知门店级 bug 的症状），
    // 防止把坏数据灌进云端主档案；确认要推（force）可绕过。
    const zeroRows = parts.detail.filter((r) => METRIC_FIELDS.every((f) => !Number(r[f]))).length;
    if (!opts.force && parts.detail.length >= 20 && zeroRows / parts.detail.length > 0.8) {
      throw err(`本批门店行 ${parts.detail.length} 行中 ${zeroRows} 行指标全零，疑似取数异常，已拦截推送（确认无误可在数据中台强制推送）`, 'AITABLE_ALL_ZERO');
    }

    const token = await getToken();
    const targets = [
      ['detail', sheets.detail, parts.detail],
      ['chain', sheets.chain, parts.chain],
      ['company', sheets.company, parts.company],
    ];
    const perSheet = {};
    let created = 0;
    let updated = 0;
    let skippedStale = 0;
    for (const [scope, sheetName, scopeRows] of targets) {
      const r = await pushSheet(cfg, token, sheetName, accountName, scopeRows, opts);
      perSheet[scope] = { sheet: sheetName, rows: scopeRows.length, ...r };
      created += r.created;
      updated += r.updated;
      skippedStale += r.skippedStale;
    }

    const now = Date.now();
    state[key] = { lastPushedAt: now, pushedRows: pushed, sheets: fingerprint, at: new Date().toISOString() };
    writePushState(state);
    return { pushed, created, updated, skippedStale, skippedSummary: parts.skippedSummary, perSheet, account: accountName };
  }

  // ---------- 拉取：AI 表格（三表） → 本地湖 ----------

  /**
   * 拉取三张 AI 表格记录回本地湖（云端为准，本地同键行被覆盖）。
   * entity 缺省 = 拉取云端全部账号：没有注册过的新账号会自动创建轻量主体
   * （fresh 机器拉一次即可用报表/图表/问数）。
   */
  async function pullLake(entity) {
    const cfg = requireConfig();
    const naposService = require('../../napos-service.js');
    logs.pushSystemLog(`云端拉取开始${entity ? `：${entity}` : '（全部账号）'}`, 'info');
    const { sheets } = resolveSheets(cfg);
    const records = [];
    for (const sheetName of [sheets.detail, sheets.chain, sheets.company]) {
      records.push(...await listAllRecords(cfg, sheetName));
    }
    const groups = new Map();
    const wantName = entity
      ? ((naposService.store.getEntity(naposService.store.sanitizeKey(entity)) || {}).name || entity)
      : null;
    for (const rec of records) {
      const f = rec.fields || {};
      const acc = String(f[FIELD_COLUMNS.account] || '').trim();
      if (!acc) continue;
      if (wantName && acc !== wantName) continue;
      if (!groups.has(acc)) groups.set(acc, []);
      groups.get(acc).push(recordToRow({ fields: f }, acc));
    }
    if (!groups.size) return { pulled: 0, accounts: [], message: '云端没有可拉取的数据' };

    const result = { pulled: 0, accounts: [] };
    for (const [acc, rows] of groups) {
      let key = naposService.store.sanitizeKey(acc);
      const meta = naposService.store.getEntity(key);
      if (!meta || !meta.key) {
        const created = naposService.store.upsertEntity({ merchantId: acc, merchantName: acc, chains: [], shops: [] });
        key = created.entity.key;
      }
      const res = naposService.store.upsertRows(key, rows);
      result.pulled += rows.length;
      result.accounts.push({ account: acc, rows: rows.length, added: res.added, updated: res.updated });
    }
    return result;
  }

  /** 连通性测试：token + 操作者身份 + 三张表逐表可读（缺哪张报哪张） */
  async function testConnection() {
    const cfg = requireConfig();
    const { sheets } = resolveSheets(cfg);
    const token = await getToken();
    const result = { ok: true, baseId: cfg.baseId, operatorId: cfg.operatorId, sheets: {} };
    for (const [scope, sheetName] of [['detail', sheets.detail], ['chain', sheets.chain], ['company', sheets.company]]) {
      const res = await transport('GET', RECORDS_PATH(cfg.baseId, sheetName) + notableQuery(cfg, { maxResults: 1 }), undefined, token);
      const ok = res.statusCode === 200;
      const hint = !ok && (res.statusCode === 403 || res.statusCode === 404)
        ? '——表不存在或不可读：请在钉钉该 AI 表格里确认已建同名数据表并分享给操作者账号'
        : ok ? '' : (res.json && res.json.message) || res.raw;
      result.sheets[scope] = { name: sheetName, ok };
      if (!ok) { result.ok = false; result.sheets[scope].hint = hint; }
    }
    return result;
  }

  function lastPushInfo(entity) {
    const key = require('../../napos-service.js').store.sanitizeKey(entity);
    const s = readPushState()[key] || null;
    return s;
  }

  /** 待推送行数：水位之后有变化（syncedAt 更新）的行；配置不齐返回 null（调用方显示配置态） */
  function pendingCount(entity) {
    const cfg = config.aitable || {};
    if (!cfg.enabled || !cfg.baseId || !cfg.appKey || !cfg.operatorId) return null;
    const naposService = require('../../napos-service.js');
    const key = naposService.store.sanitizeKey(entity);
    const watermark = (readPushState()[key] || {}).lastPushedAt || 0;
    const all = naposService.store.loadRows(key) || [];
    if (!watermark) return all.length;
    return all.filter((r) => new Date(r.syncedAt || 0).getTime() > watermark).length;
  }

  logs.pushSystemLog('钉钉 AI 表格同步服务已就绪（三表：门店明细/连锁汇总/公司汇总）', 'info');
  return { pushLake, pullLake, testConnection, lastPushInfo, pendingCount };
}

module.exports = {
  createAitableService, FIELD_COLUMNS, DIMENSION_LABEL, SHEET_DEFAULTS,
  rowToRecord, recordKey, recordToRow, splitRowsBySheet, resolveSheets, pushSummary,
};