reports-store.js 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192
  1. 'use strict';
  2. /**
  3. * 数据报表 · 本地数据湖
  4. * ----------------------------------------------------------------------------
  5. * 把从淘宝闪购商家后台取回的经营数据沉淀到本地 JSON 文件,支持:
  6. * · 多主体(每登录一个商家账号即一个主体 entity)
  7. * · 主体内分:全公司 / 连锁 / 单店 三种粒度的行
  8. * · 按「日期 × 门店」upsert 合并,可反复同步同一区间不产生重复
  9. *
  10. * 文件布局(项目根 data/reports/):
  11. * entities.json 主体注册表(含快照的连锁/门店结构)
  12. * <entityKey>/daily.json 该主体的数据行(唯一键 date|shopId)
  13. * <entityKey>/syncs.json 同步历史
  14. */
  15. const fs = require('fs');
  16. const path = require('path');
  17. const LAKE_DIR = path.join(__dirname, 'data', 'reports');
  18. // ---------- 通用 ----------
  19. function ensureDir(p) {
  20. fs.mkdirSync(p, { recursive: true });
  21. return p;
  22. }
  23. function readJsonSafe(file, fallback) {
  24. try {
  25. return JSON.parse(fs.readFileSync(file, 'utf8'));
  26. } catch {
  27. return fallback;
  28. }
  29. }
  30. function writeJsonAtomic(file, obj) {
  31. ensureDir(path.dirname(file));
  32. const temp = file + '.tmp';
  33. fs.writeFileSync(temp, JSON.stringify(obj), 'utf8');
  34. fs.renameSync(temp, file);
  35. }
  36. function sanitizeKey(v) {
  37. return String(v || '').replace(/[^a-zA-Z0-9_-]/g, '').slice(0, 64) || 'default';
  38. }
  39. // ---------- 主体注册表 ----------
  40. function registryFile() {
  41. return path.join(LAKE_DIR, 'entities.json');
  42. }
  43. function loadRegistry() {
  44. return readJsonSafe(registryFile(), { entities: [] });
  45. }
  46. function saveRegistry(reg) {
  47. writeJsonAtomic(registryFile(), reg);
  48. }
  49. /**
  50. * 注册 / 更新一个主体(来自 shopTree 快照)。
  51. * 返回 { entity, created }。
  52. */
  53. function upsertEntity({ merchantId, merchantName, chains = [], shops = [] }) {
  54. const reg = loadRegistry();
  55. const key = sanitizeKey(merchantId);
  56. const now = new Date().toISOString();
  57. let entity = reg.entities.find((e) => e.key === key);
  58. let created = false;
  59. if (!entity) {
  60. created = true;
  61. entity = { key, firstSeen: now, lastSeen: now };
  62. reg.entities.push(entity);
  63. }
  64. entity.lastSeen = now;
  65. entity.merchantId = merchantId || null;
  66. entity.name = merchantName || entity.name || key;
  67. entity.chainCount = chains.length;
  68. entity.shopCount = shops.length;
  69. entity.chains = chains;
  70. entity.shops = shops;
  71. saveRegistry(reg);
  72. return { entity, created };
  73. }
  74. function listEntities() {
  75. const reg = loadRegistry();
  76. return reg.entities
  77. .slice()
  78. .sort((a, b) => (a.lastSeen < b.lastSeen ? 1 : -1));
  79. }
  80. function getEntity(key) {
  81. return loadRegistry().entities.find((e) => e.key === sanitizeKey(key)) || null;
  82. }
  83. // ---------- 数据行 ----------
  84. function dailyFile(entityKey) {
  85. return path.join(LAKE_DIR, sanitizeKey(entityKey), 'daily.json');
  86. }
  87. function loadRows(entityKey) {
  88. const data = readJsonSafe(dailyFile(entityKey), { rows: [] });
  89. return Array.isArray(data.rows) ? data.rows : [];
  90. }
  91. function saveRows(entityKey, rows) {
  92. rows.sort((a, b) => (a.date === b.date
  93. ? String(a.shopId).localeCompare(String(b.shopId))
  94. : (a.date < b.date ? -1 : 1)));
  95. writeJsonAtomic(dailyFile(entityKey), { version: 1, updated: new Date().toISOString(), rows });
  96. }
  97. function rowKey(row) {
  98. // scope 必须参与键:全公司行与连锁行的 shopId 都为空串,缺 scope 会互相覆盖
  99. return `${row.date}|${row.scope || ''}|${row.shopId == null ? '' : row.shopId}`;
  100. }
  101. /** 合并写入:按 date|scope|shopId upsert,浅字段覆盖(raw 整体替换) */
  102. function upsertRows(entityKey, incoming) {
  103. if (!Array.isArray(incoming) || !incoming.length) return { added: 0, updated: 0, total: 0 };
  104. const rows = loadRows(entityKey);
  105. const index = new Map(rows.map((r, i) => [rowKey(r), i]));
  106. let added = 0;
  107. let updated = 0;
  108. for (const next of incoming) {
  109. if (!next || !next.date) continue;
  110. const k = rowKey(next);
  111. const prev = index.has(k) ? rows[index.get(k)] : null;
  112. if (prev) {
  113. Object.assign(prev, next, { raw: next.raw || prev.raw || null });
  114. updated += 1;
  115. } else {
  116. rows.push(next);
  117. index.set(k, rows.length - 1);
  118. added += 1;
  119. }
  120. }
  121. saveRows(entityKey, rows);
  122. return { added, updated, total: rows.length };
  123. }
  124. /** 行过滤:主体 + 区间 + 门店 + 粒度 */
  125. function queryRows(entityKey, { from, to, shopId, scope } = {}) {
  126. return loadRows(entityKey).filter((r) => {
  127. if (from && r.date < from) return false;
  128. if (to && r.date > to) return false;
  129. if (shopId && String(r.shopId) !== String(shopId)) return false;
  130. if (scope && r.scope !== scope) return false;
  131. return true;
  132. });
  133. }
  134. function datesInLake(entityKey) {
  135. const set = new Set();
  136. for (const r of loadRows(entityKey)) set.add(r.date);
  137. return [...set].sort();
  138. }
  139. function deleteEntityData(entityKey) {
  140. const dir = path.join(LAKE_DIR, sanitizeKey(entityKey));
  141. fs.rmSync(dir, { recursive: true, force: true });
  142. }
  143. // ---------- 同步历史 ----------
  144. function syncsFile(entityKey) {
  145. return path.join(LAKE_DIR, sanitizeKey(entityKey), 'syncs.json');
  146. }
  147. function appendSyncLog(entityKey, entry) {
  148. const file = syncsFile(entityKey);
  149. const data = readJsonSafe(file, { runs: [] });
  150. data.runs.unshift(entry);
  151. if (data.runs.length > 50) data.runs.length = 50;
  152. writeJsonAtomic(file, data);
  153. }
  154. function loadSyncLogs(entityKey) {
  155. return readJsonSafe(syncsFile(entityKey), { runs: [] }).runs;
  156. }
  157. module.exports = {
  158. LAKE_DIR,
  159. loadRegistry, saveRegistry, upsertEntity, listEntities, getEntity,
  160. loadRows, saveRows, upsertRows, queryRows, datesInLake, deleteEntityData, rowKey,
  161. appendSyncLog, loadSyncLogs,
  162. sanitizeKey,
  163. };