| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192 |
- 'use strict';
- /**
- * 数据报表 · 本地数据湖
- * ----------------------------------------------------------------------------
- * 把从淘宝闪购商家后台取回的经营数据沉淀到本地 JSON 文件,支持:
- * · 多主体(每登录一个商家账号即一个主体 entity)
- * · 主体内分:全公司 / 连锁 / 单店 三种粒度的行
- * · 按「日期 × 门店」upsert 合并,可反复同步同一区间不产生重复
- *
- * 文件布局(项目根 data/reports/):
- * entities.json 主体注册表(含快照的连锁/门店结构)
- * <entityKey>/daily.json 该主体的数据行(唯一键 date|shopId)
- * <entityKey>/syncs.json 同步历史
- */
- const fs = require('fs');
- const path = require('path');
- const 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;
- }
- // ---------- 数据行 ----------
- function dailyFile(entityKey) {
- return path.join(LAKE_DIR, sanitizeKey(entityKey), 'daily.json');
- }
- function loadRows(entityKey) {
- const data = readJsonSafe(dailyFile(entityKey), { rows: [] });
- return Array.isArray(data.rows) ? data.rows : [];
- }
- function saveRows(entityKey, rows) {
- rows.sort((a, b) => (a.date === b.date
- ? String(a.shopId).localeCompare(String(b.shopId))
- : (a.date < b.date ? -1 : 1)));
- writeJsonAtomic(dailyFile(entityKey), { version: 1, updated: new Date().toISOString(), rows });
- }
- function rowKey(row) {
- // scope 必须参与键:全公司行与连锁行的 shopId 都为空串,缺 scope 会互相覆盖
- return `${row.date}|${row.scope || ''}|${row.shopId == null ? '' : row.shopId}`;
- }
- /** 合并写入:按 date|scope|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;
- for (const next of incoming) {
- if (!next || !next.date) continue;
- 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;
- }
- }
- saveRows(entityKey, rows);
- 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));
- 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,
- };
|