'use strict'; /** * 数据报表 · 本地数据湖 * ---------------------------------------------------------------------------- * 把从淘宝闪购商家后台取回的经营数据沉淀到本地 JSON 文件,支持: * · 多主体(每登录一个商家账号即一个主体 entity) * · 主体内分:全公司 / 连锁 / 单店 三种粒度的行 * · 按「日期 × 门店」upsert 合并,可反复同步同一区间不产生重复 * * 文件布局(项目根 data/reports/): * entities.json 主体注册表(含快照的连锁/门店结构) * /daily-YYYYMM.json 按月分片的数据行(行键 date|scope|shopId) * /daily.json 旧版单文件;读到的首次写入时自动迁移为分片并删除 * /syncs.json 同步历史 * * 按月分片的意义:同步/查询只触碰命中月份的分片,重写体积有上界; * 过期修剪等价于删除旧月份文件。 */ const fs = require('fs'); const path = require('path'); // 湖根目录:GENPIC_LAKE_DIR 环境变量可覆盖(smoke 测试隔离用,生产不受影响) const LAKE_DIR = process.env.GENPIC_LAKE_DIR ? path.resolve(process.env.GENPIC_LAKE_DIR) : path.join(__dirname, 'data', 'reports'); // ---------- 通用 ---------- function ensureDir(p) { fs.mkdirSync(p, { recursive: true }); return p; } function readJsonSafe(file, fallback) { try { return JSON.parse(fs.readFileSync(file, 'utf8')); } catch { return fallback; } } function writeJsonAtomic(file, obj) { ensureDir(path.dirname(file)); const temp = file + '.tmp'; fs.writeFileSync(temp, JSON.stringify(obj), 'utf8'); fs.renameSync(temp, file); } function sanitizeKey(v) { return String(v || '').replace(/[^a-zA-Z0-9_-]/g, '').slice(0, 64) || 'default'; } // ---------- 主体注册表 ---------- function registryFile() { return path.join(LAKE_DIR, 'entities.json'); } function loadRegistry() { return readJsonSafe(registryFile(), { entities: [] }); } function saveRegistry(reg) { writeJsonAtomic(registryFile(), reg); } /** * 注册 / 更新一个主体(来自 shopTree 快照)。 * 返回 { entity, created }。 */ function upsertEntity({ merchantId, merchantName, chains = [], shops = [] }) { const reg = loadRegistry(); const key = sanitizeKey(merchantId); const now = new Date().toISOString(); let entity = reg.entities.find((e) => e.key === key); let created = false; if (!entity) { created = true; entity = { key, firstSeen: now, lastSeen: now }; reg.entities.push(entity); } entity.lastSeen = now; entity.merchantId = merchantId || null; entity.name = merchantName || entity.name || key; entity.chainCount = chains.length; entity.shopCount = shops.length; entity.chains = chains; entity.shops = shops; saveRegistry(reg); return { entity, created }; } function listEntities() { const reg = loadRegistry(); return reg.entities .slice() .sort((a, b) => (a.lastSeen < b.lastSeen ? 1 : -1)); } function getEntity(key) { return loadRegistry().entities.find((e) => e.key === sanitizeKey(key)) || null; } // ---------- 数据行(按月分片) ---------- const SHARD_RE = /^daily-(\d{6}|misc)\.json$/; function entityDir(entityKey) { return path.join(LAKE_DIR, sanitizeKey(entityKey)); } function legacyFile(entityKey) { return path.join(entityDir(entityKey), 'daily.json'); } function shardFile(entityKey, month) { return path.join(entityDir(entityKey), `daily-${month}.json`); } /** 行的月份桶:date 取 YYYYMMDD 前六位;异常格式进 misc 桶兜底 */ function monthOf(row) { const m = /^(\d{6})/.exec(String(row.date || '')); return m ? m[1] : 'misc'; } function sortRows(a, b) { return a.date === b.date ? String(a.shopId).localeCompare(String(b.shopId)) : (a.date < b.date ? -1 : 1); } // 分片解析缓存:数据中台/明细/湖全景每个请求都会 loadRows,mtime+size 未变的 // 文件直接复用上次解析结果,省掉重复 JSON.parse。写路径(upsert/save/delete) // 改动文件后 mtime 变化,缓存自然失效;缓存行对象为共享引用,调用方只读不改。 const shardCache = new Map(); // 绝对路径 -> { mtimeMs, size, rows } function readShardRows(file, into) { let stat = null; try { stat = fs.statSync(file); } catch { shardCache.delete(file); return; } const hit = shardCache.get(file); if (hit && hit.mtimeMs === stat.mtimeMs && hit.size === stat.size) { for (const r of hit.rows) into.set(rowKey(r), r); return; } const data = readJsonSafe(file, null); if (data && Array.isArray(data.rows)) { for (const r of data.rows) if (r && r.date) into.set(rowKey(r), r); shardCache.set(file, { mtimeMs: stat.mtimeMs, size: stat.size, rows: data.rows }); } else { shardCache.delete(file); } } /** * 全量视图:旧单文件 + 各月分片合并(分片优先于旧文件,覆盖迁移中断窗口)。 * 对上层仍是「一个 rows 数组」,与分片前语义一致。 */ function loadRows(entityKey) { const byKey = new Map(); readShardRows(legacyFile(entityKey), byKey); let shards = []; try { shards = fs.readdirSync(entityDir(entityKey)).filter((f) => SHARD_RE.test(f)).sort(); } catch { /* 目录不存在 */ } for (const f of shards) readShardRows(path.join(entityDir(entityKey), f), byKey); const rows = [...byKey.values()]; rows.sort(sortRows); return rows; } function writeShard(entityKey, month, rows) { writeJsonAtomic(shardFile(entityKey, month), { version: 2, updated: new Date().toISOString(), rows: rows.slice().sort(sortRows), }); } function removeLegacyFile(entityKey) { try { fs.unlinkSync(legacyFile(entityKey)); } catch { /* 已不存在 */ } } /** 整体替换某主体全部行(语义与旧 saveRows 一致),会清掉不再存在的月份分片 */ function saveRows(entityKey, rows) { const byMonth = new Map(); for (const r of rows) { if (!r || !r.date) continue; const m = monthOf(r); if (!byMonth.has(m)) byMonth.set(m, []); byMonth.get(m).push(r); } const dir = entityDir(entityKey); try { for (const f of fs.readdirSync(dir)) { const m = SHARD_RE.exec(f); if (m && !byMonth.has(m[1])) fs.unlinkSync(path.join(dir, f)); } } catch { /* 目录不存在则写时自建 */ } for (const [m, list] of byMonth) writeShard(entityKey, m, list); removeLegacyFile(entityKey); } function rowKey(row) { // scope 与 chainId 都必须参与键:全公司行、连锁行、区间汇总行的 shopId 都为空串, // 缺 chainId 时同日多个连锁/汇总行会互相覆盖(只留最后写入的一条) return `${row.date}|${row.scope || ''}|${row.chainId == null ? '' : String(row.chainId)}|${row.shopId == null ? '' : String(row.shopId)}`; } /** 合并写入:按 date|scope|chainId|shopId upsert,浅字段覆盖(raw 整体替换)。 * 只重写被本次数据触及的月份分片;旧单文件在首次写入时整体迁移。 */ function upsertRows(entityKey, incoming) { if (!Array.isArray(incoming) || !incoming.length) return { added: 0, updated: 0, total: 0 }; const rows = loadRows(entityKey); const index = new Map(rows.map((r, i) => [rowKey(r), i])); let added = 0; let updated = 0; const touched = new Set(); for (const next of incoming) { if (!next || !next.date) continue; touched.add(monthOf(next)); const k = rowKey(next); const prev = index.has(k) ? rows[index.get(k)] : null; if (prev) { Object.assign(prev, next, { raw: next.raw || prev.raw || null }); updated += 1; } else { rows.push(next); index.set(k, rows.length - 1); added += 1; } } const hasLegacy = fs.existsSync(legacyFile(entityKey)); const byMonth = new Map(); for (const r of rows) { const m = monthOf(r); // 未触及且无旧文件要迁移的月份保留原分片,不重写 if (!hasLegacy && !touched.has(m)) continue; if (!byMonth.has(m)) byMonth.set(m, []); byMonth.get(m).push(r); } for (const [m, list] of byMonth) writeShard(entityKey, m, list); if (hasLegacy) removeLegacyFile(entityKey); return { added, updated, total: rows.length }; } /** 行过滤:主体 + 区间 + 门店 + 粒度 */ function queryRows(entityKey, { from, to, shopId, scope } = {}) { return loadRows(entityKey).filter((r) => { if (from && r.date < from) return false; if (to && r.date > to) return false; if (shopId && String(r.shopId) !== String(shopId)) return false; if (scope && r.scope !== scope) return false; return true; }); } function datesInLake(entityKey) { const set = new Set(); for (const r of loadRows(entityKey)) set.add(r.date); return [...set].sort(); } function deleteEntityData(entityKey) { const dir = path.join(LAKE_DIR, sanitizeKey(entityKey)); // 先摘掉该主体的分片缓存,避免重建同名主体时短暂引用已删除的旧行对象 for (const f of [...shardCache.keys()]) { if (f.startsWith(dir + path.sep)) shardCache.delete(f); } fs.rmSync(dir, { recursive: true, force: true }); } // ---------- 同步历史 ---------- function syncsFile(entityKey) { return path.join(LAKE_DIR, sanitizeKey(entityKey), 'syncs.json'); } function appendSyncLog(entityKey, entry) { const file = syncsFile(entityKey); const data = readJsonSafe(file, { runs: [] }); data.runs.unshift(entry); if (data.runs.length > 50) data.runs.length = 50; writeJsonAtomic(file, data); } function loadSyncLogs(entityKey) { return readJsonSafe(syncsFile(entityKey), { runs: [] }).runs; } module.exports = { LAKE_DIR, loadRegistry, saveRegistry, upsertEntity, listEntities, getEntity, loadRows, saveRows, upsertRows, queryRows, datesInLake, deleteEntityData, rowKey, appendSyncLog, loadSyncLogs, sanitizeKey, };