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

'use strict';

// 钉钉 AI 表格（原多维表）同步服务
// ----------------------------------------------------------------------------
// 双层架构的云端层：AI 表格 = 在线主档案 + 多人共享层；本地数据湖 = 计算缓存层。
//   · pushLake  — 把本地湖的新增/更新行（按 syncedAt 水位）upsert 到 AI 表格
//   · pullLake  — 把 AI 表格记录拉回本地湖（其他机器补数据用，云端为准）
// 业务键：日期|账号名|维度|门店ID（与本地 rowKey 语义对齐，幂等可重跑）。
//
// 钉钉侧契约（企业内部应用 access_token，域 api.dingtalk.com；2026-10-17 真机核对）：
//   token:   POST /v1.0/oauth2/accessToken  {appKey, appSecret} → {accessToken, expireIn}
//   列记录:  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 只影响显示，存储小数保真
// ⚠ 批量上限（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;

// 本地字段 → 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: '汇总' };

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.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;
}

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.sheetName || !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) {
    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, cfg.sheetName) + q, undefined, token);
      if (res.statusCode !== 200) throw err(`拉取记录失败 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 { /* 状态落盘失败不阻断推送 */ }
  }

  /** 推送一个主体的数据。force=true 时忽略水位推全量、并跳过冲突/全零保护。
   *  冲突策略：同键行「数据新者胜」——云端记录的 lastModifiedTime 比 本地行 syncedAt
   *  （取数时间）新则跳过（skippedStale），避免另一台机器后推的旧数据盖掉云端新数据。 */
  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 state = readPushState();
    const watermark = opts.force ? 0 : (state[key] || {}).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) || [];
    }
    if (!rows.length) return { pushed: 0, created: 0, updated: 0, skipped: '无新增数据' };

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

    const token = await getToken();
    // 现有记录索引：业务键 → {recordId, mtime}（分页拉全量）
    const existing = new Map();
    for (const rec of await listAllRecords(cfg)) {
      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.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, cfg.sheetName) + notableQuery(cfg), { records: part }, token);
      if (res.statusCode >= 300) throw err(`批量新增失败 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, cfg.sheetName) + notableQuery(cfg), { records: part }, token);
      if (res.statusCode >= 300) throw err(`批量更新失败 HTTP ${res.statusCode}: ${(res.json && res.json.message) || res.raw}`, 'AITABLE_UPDATE_FAIL');
      updated += part.length;
    }

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

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

  /**
   * 拉取 AI 表格记录回本地湖（云端为准，本地同键行被覆盖）。
   * entity 缺省 = 拉取云端全部账号：没有注册过的新账号会自动创建轻量主体
   * （fresh 机器拉一次即可用报表/图表/问数）。
   */
  async function pullLake(entity) {
    const cfg = requireConfig();
    const naposService = require('../../napos-service.js');
    const records = await listAllRecords(cfg);
    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 token = await getToken();
    const res = await transport('GET', RECORDS_PATH(cfg.baseId, cfg.sheetName) + notableQuery(cfg, { maxResults: 1 }), undefined, token);
    if (res.statusCode !== 200) {
      const hint = res.statusCode === 403 || res.statusCode === 404
        ? '——请确认该 AI 表格已分享给操作者账号（钉钉里把表格协作权限授给对应同事）'
        : '';
      throw err(`表读取失败 HTTP ${res.statusCode}: ${(res.json && res.json.message) || res.raw}${hint}`, 'AITABLE_TEST_FAIL');
    }
    return { ok: true, baseId: cfg.baseId, sheetName: cfg.sheetName, operatorId: cfg.operatorId };
  }

  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.sheetName || !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, rowToRecord, recordKey, recordToRow };