hyreal-gen-pic/reports-store.js

'use strict';

/**
 * 数据报表 · 本地数据湖
 * ----------------------------------------------------------------------------
 * 把从淘宝闪购商家后台取回的经营数据沉淀到本地 JSON 文件，支持：
 *   · 多主体（每登录一个商家账号即一个主体 entity）
 *   · 主体内分：全公司 / 连锁 / 单店 三种粒度的行
 *   · 按「日期 × 门店」upsert 合并，可反复同步同一区间不产生重复
 *
 * 文件布局（项目根 data/reports/）：
 *   entities.json                  主体注册表（含快照的连锁/门店结构）
 *   <entityKey>/daily-YYYYMM.json  按月分片的数据行（行键 date|scope|shopId）
 *   <entityKey>/daily.json         旧版单文件；读到的首次写入时自动迁移为分片并删除
 *   <entityKey>/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,
};