'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 表格(三表) ---------- let pushing = false; // 单飞锁:手动推送与自动同步的 fire-and-forget 推送不得并发 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 = {}) { if (pushing) throw err('已有推送在进行中,请稍后再试', 'AITABLE_PUSH_BUSY'); pushing = true; try { return await pushLakeLocked(entity, opts); } finally { pushing = false; } } async function pushLakeLocked(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; } /** 湖↔云只读比对快照(问数智能体 aitable_verify 工具用): * 三表记录按当前主体账号名过滤后整表拉回,不做任何写入。 */ async function snapshotForVerify(entity) { const cfg = requireConfig(); const naposService = require('../../napos-service.js'); const { sheets } = resolveSheets(cfg); const name = (naposService.store.getEntity(naposService.store.sanitizeKey(entity)) || {}).name || entity; const out = {}; for (const [scope, sheetName] of [['shop', sheets.detail], ['chain', sheets.chain], ['company', sheets.company]]) { const records = await listAllRecords(cfg, sheetName); out[scope] = records .map((rec) => recordToRow(rec, name)) .filter((r) => String(r.account || '') === name); } return out; } logs.pushSystemLog('钉钉 AI 表格同步服务已就绪(三表:门店明细/连锁汇总/公司汇总)', 'info'); return { pushLake, pullLake, testConnection, lastPushInfo, pendingCount, snapshotForVerify }; } module.exports = { createAitableService, FIELD_COLUMNS, DIMENSION_LABEL, SHEET_DEFAULTS, rowToRecord, recordKey, recordToRow, splitRowsBySheet, resolveSheets, pushSummary, };