'use strict'; // 钉钉 AI 表格(原多维表)同步服务 // ---------------------------------------------------------------------------- // 双层架构的云端层:AI 表格 = 在线主档案 + 多人共享层;本地数据湖 = 计算缓存层。 // · pushLake — 把本地湖的新增/更新行(按 syncedAt 水位)upsert 到 AI 表格 // · pullLake — 把 AI 表格记录拉回本地湖(其他机器补数据用,云端为准) // 业务键:日期|账号名|维度|门店ID(与本地 rowKey 语义对齐,幂等可重跑)。 // // 钉钉侧契约(企业内部应用 access_token,域 api.dingtalk.com): // token: POST /v1.0/oauth2/accessToken {appKey, appSecret} → {accessToken, expireIn} // 记录: /v1.0/notable/bases/{baseId}/sheets/{sheetName}/records // POST 批量新增 / PUT 批量更新 / GET 分页列表 // ⚠ 端点形态与批量上限在首次真机联调时核对——所有钉钉细节都收敛在本文件, // 如有偏差只需改这里的常量与两个 transport 方法。 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; } 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'); return cfg; } async function listAllRecords(cfg) { const token = await getToken(); const out = []; let nextToken = ''; for (;;) { const q = `?maxResults=${LIST_PAGE}${nextToken ? `&nextToken=${encodeURIComponent(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 时忽略水位推全量。返回统计。 */ 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: '无新增数据' }; const token = await getToken(); // 现有记录索引:业务键 → recordId(分页拉全量) 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, rec.id); } let created = 0; let updated = 0; const toCreate = []; const toUpdate = []; // {recordId, fields} for (const row of rows) { const rec = rowToRecord(row, accountName); const k = recordKey(rec.fields); const rid = existing.get(k); if (rid) toUpdate.push({ record_id: rid, 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), { 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), { 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, 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) + `?maxResults=1`, undefined, token); if (res.statusCode !== 200) throw err(`表读取失败 HTTP ${res.statusCode}: ${(res.json && res.json.message) || res.raw}`, 'AITABLE_TEST_FAIL'); return { ok: true, baseId: cfg.baseId, sheetName: cfg.sheetName }; } function lastPushInfo(entity) { const key = require('../../napos-service.js').store.sanitizeKey(entity); const s = readPushState()[key] || null; return s; } logs.pushSystemLog('钉钉 AI 表格同步服务已就绪', 'info'); return { pushLake, pullLake, testConnection, lastPushInfo }; } module.exports = { createAitableService, FIELD_COLUMNS, DIMENSION_LABEL, rowToRecord, recordKey, recordToRow };