|
@@ -9,9 +9,13 @@
|
|
|
* · 按「日期 × 门店」upsert 合并,可反复同步同一区间不产生重复
|
|
* · 按「日期 × 门店」upsert 合并,可反复同步同一区间不产生重复
|
|
|
*
|
|
*
|
|
|
* 文件布局(项目根 data/reports/):
|
|
* 文件布局(项目根 data/reports/):
|
|
|
- * entities.json 主体注册表(含快照的连锁/门店结构)
|
|
|
|
|
- * <entityKey>/daily.json 该主体的数据行(唯一键 date|shopId)
|
|
|
|
|
- * <entityKey>/syncs.json 同步历史
|
|
|
|
|
|
|
+ * entities.json 主体注册表(含快照的连锁/门店结构)
|
|
|
|
|
+ * <entityKey>/daily-YYYYMM.json 按月分片的数据行(行键 date|scope|shopId)
|
|
|
|
|
+ * <entityKey>/daily.json 旧版单文件;读到的首次写入时自动迁移为分片并删除
|
|
|
|
|
+ * <entityKey>/syncs.json 同步历史
|
|
|
|
|
+ *
|
|
|
|
|
+ * 按月分片的意义:同步/查询只触碰命中月份的分片,重写体积有上界;
|
|
|
|
|
+ * 过期修剪等价于删除旧月份文件。
|
|
|
*/
|
|
*/
|
|
|
|
|
|
|
|
const fs = require('fs');
|
|
const fs = require('fs');
|
|
@@ -96,22 +100,86 @@ function getEntity(key) {
|
|
|
return loadRegistry().entities.find((e) => e.key === sanitizeKey(key)) || null;
|
|
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 dailyFile(entityKey) {
|
|
|
|
|
- return path.join(LAKE_DIR, sanitizeKey(entityKey), 'daily.json');
|
|
|
|
|
|
|
+function sortRows(a, b) {
|
|
|
|
|
+ return a.date === b.date
|
|
|
|
|
+ ? String(a.shopId).localeCompare(String(b.shopId))
|
|
|
|
|
+ : (a.date < b.date ? -1 : 1);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+function readShardRows(file, into) {
|
|
|
|
|
+ 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);
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+/**
|
|
|
|
|
+ * 全量视图:旧单文件 + 各月分片合并(分片优先于旧文件,覆盖迁移中断窗口)。
|
|
|
|
|
+ * 对上层仍是「一个 rows 数组」,与分片前语义一致。
|
|
|
|
|
+ */
|
|
|
function loadRows(entityKey) {
|
|
function loadRows(entityKey) {
|
|
|
- const data = readJsonSafe(dailyFile(entityKey), { rows: [] });
|
|
|
|
|
- return Array.isArray(data.rows) ? data.rows : [];
|
|
|
|
|
|
|
+ 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) {
|
|
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 });
|
|
|
|
|
|
|
+ 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) {
|
|
function rowKey(row) {
|
|
@@ -119,15 +187,18 @@ function rowKey(row) {
|
|
|
return `${row.date}|${row.scope || ''}|${row.shopId == null ? '' : row.shopId}`;
|
|
return `${row.date}|${row.scope || ''}|${row.shopId == null ? '' : row.shopId}`;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-/** 合并写入:按 date|scope|shopId upsert,浅字段覆盖(raw 整体替换) */
|
|
|
|
|
|
|
+/** 合并写入:按 date|scope|shopId upsert,浅字段覆盖(raw 整体替换)。
|
|
|
|
|
+ * 只重写被本次数据触及的月份分片;旧单文件在首次写入时整体迁移。 */
|
|
|
function upsertRows(entityKey, incoming) {
|
|
function upsertRows(entityKey, incoming) {
|
|
|
if (!Array.isArray(incoming) || !incoming.length) return { added: 0, updated: 0, total: 0 };
|
|
if (!Array.isArray(incoming) || !incoming.length) return { added: 0, updated: 0, total: 0 };
|
|
|
const rows = loadRows(entityKey);
|
|
const rows = loadRows(entityKey);
|
|
|
const index = new Map(rows.map((r, i) => [rowKey(r), i]));
|
|
const index = new Map(rows.map((r, i) => [rowKey(r), i]));
|
|
|
let added = 0;
|
|
let added = 0;
|
|
|
let updated = 0;
|
|
let updated = 0;
|
|
|
|
|
+ const touched = new Set();
|
|
|
for (const next of incoming) {
|
|
for (const next of incoming) {
|
|
|
if (!next || !next.date) continue;
|
|
if (!next || !next.date) continue;
|
|
|
|
|
+ touched.add(monthOf(next));
|
|
|
const k = rowKey(next);
|
|
const k = rowKey(next);
|
|
|
const prev = index.has(k) ? rows[index.get(k)] : null;
|
|
const prev = index.has(k) ? rows[index.get(k)] : null;
|
|
|
if (prev) {
|
|
if (prev) {
|
|
@@ -139,7 +210,17 @@ function upsertRows(entityKey, incoming) {
|
|
|
added += 1;
|
|
added += 1;
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
- saveRows(entityKey, rows);
|
|
|
|
|
|
|
+ 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 };
|
|
return { added, updated, total: rows.length };
|
|
|
}
|
|
}
|
|
|
|
|
|