| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182 |
- # -*- coding: utf-8 -*-
- """产物来源**自登记**: 构建脚本落盘后把"这一件是我从 data/raw 算出来的"记进 `outputs/<场>/_derived_manifest.json`。
- ## 为什么要有它
- `outputs/<场>/_provenance.json` 是**逐件来源台账**(raw-derived = 由 data/raw 重算 / shipped = 包内无生成端,
- 用随包件补齐)。它由 `scripts/products_restore_missing.py` 生成, 而那个脚本是按**随包快照**逐件走一遍的 ——
- 于是**新造的、快照里根本没有的产物**不会自动进台账 (既不算 raw-derived 也不算 shipped)。
- 早期的做法是在 `products_restore_missing.py` 里维护一张 `RAW_DERIVED` 精确路径表。对"件数少、名字固定"
- 的产物够用; 但振动侧的产物是 `<窗>/index.parquet` + `<窗>/spectra/*.npz`(分片名带序号) —— 窗名与分片数
- 都随数据变, 写不进精确表, 而**按名字通配**又会误伤同名旧件 (例如 `报告_CMS振动状态评估报告_*.md`
- 既有随包/自产的、也有厂家报告转录的, 名字形态一样)。
- 所以改成**自登记**: 谁算的谁登记, 台账只认这份登记。名字对不上不是问题, 因为登记的是**相对路径本身**。
- 用法 (构建脚本内):
- from src.derived_manifest import record
- record(P.out_root('rudong'), {rel: 'scripts/rudong_tcm_index.py (54 列, 与包内 tcm_index.parquet 同构)'},
- by='scripts/vib_raw_build.py')
- """
- from __future__ import annotations
- import json
- import pathlib
- import time
- FILENAME = '_derived_manifest.json'
- def path_of(store_root) -> pathlib.Path:
- return pathlib.Path(store_root) / FILENAME
- def load(store_root) -> dict:
- p = path_of(store_root)
- if not p.exists():
- return {}
- try:
- return json.loads(p.read_text(encoding='utf-8'))
- except Exception:
- return {}
- def prune(store_root) -> int:
- """删掉**登记了但盘上已不存在**的条目, 返回删除数。
- 为什么需要 (2026-09-16 实逮): 振动摄入对同一批数据重跑时会落 `<窗>_reimport_<时分>` 窗
- (设计如此, 该窗被 `data.EXCLUDE_DEFAULT` 排除在生产集外), 而登记是**追加式**的 ——
- 只补不删。重算几次后 `_derived_manifest.json` 里就攒了成百上千条指向已删目录的条目,
- `_provenance.json` 的 raw-derived 计数随之虚增 (实测 1,740 → 3,444, 而盘上并没有多出这些件)。
- 台账是本包的"来源正本", 虚高等于说假话 ⇒ 每次生成台账前先 prune。
- """
- store_root = pathlib.Path(store_root)
- cur = load(store_root)
- files = cur.get('files') or {}
- keep = {rel: v for rel, v in files.items() if (store_root / rel).exists()}
- gone = len(files) - len(keep)
- if gone:
- cur['files'] = keep
- cur['pruned'] = f'{time.strftime("%Y-%m-%d %H:%M")} 清理 {gone} 条不在盘的登记'
- path_of(store_root).write_text(json.dumps(cur, ensure_ascii=False, indent=1), encoding='utf-8')
- return gone
- def record(store_root, files: dict, by: str) -> pathlib.Path:
- """把 {相对产物仓的路径: 构建器说明} 合并进登记 (幂等: 同路径后写覆盖先写)。
- 幂等很关键 —— 重跑摄入不该让登记无限膨胀; 同时**不删**别的构建器登记的条目
- (振动摄入与厂家报告摄入是两个脚本, 各登各的)。"""
- store_root = pathlib.Path(store_root)
- cur = load(store_root)
- entries = cur.get('files') or {}
- for rel, builder in files.items():
- entries[pathlib.Path(rel).as_posix()] = dict(builder=builder, by=by,
- at=time.strftime('%Y-%m-%d %H:%M:%S'))
- cur = dict(note='产物来源自登记: 由构建脚本落盘后写入; _provenance.json 生成时把这些件记为 raw-derived',
- at=time.strftime('%Y-%m-%d %H:%M:%S'), files=entries)
- p = path_of(store_root)
- p.parent.mkdir(parents=True, exist_ok=True)
- p.write_text(json.dumps(cur, ensure_ascii=False, indent=1), encoding='utf-8')
- return p
|
