raw_scan.py 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452
  1. #!/usr/bin/env python3
  2. # -*- coding: utf-8 -*-
  3. r"""输入数据自动扫描识别 —— 发现 `<安装目录>/data/raw` 下的**新增/变化**,并给出"该重跑哪几步"。
  4. ## 为什么要有它(用户令 2026-09-19)
  5. 用户令:「对 `<安装目录>/data/raw` 目录下的接入数据处理,增加自动扫描识别机制,以能发现新增数据,并纳入重算。」
  6. 在此之前,"放了新数据要不要重算"全靠人记:`place_raw_data.py` 只管把现场包落位,`raw_data_check.py`
  7. 只管结构与命名对不对,`rebuild_all.py` 无脑全跑(或按人给的 `--skip-*` 跳过)——
  8. **没有任何一处回答"这回新来了什么、因此必须重跑哪几步"**。本脚本补这一环:
  9. 逐族指纹 (件数/体积/最新时间/清单摘要) → 与上次快照比 → 报【新增/变化】 → 映射到链上步骤
  10. ## 三档用法
  11. python scripts/raw_scan.py # 只报当前状态(不写快照)
  12. python scripts/raw_scan.py --write # 写/更新快照 outputs/<场>/_raw_scan.json
  13. python scripts/raw_scan.py --check # 与快照比: rc=0 无变化 · rc=4 有新增/变化 · rc=5 没有快照
  14. python scripts/raw_scan.py --check --write # 比完就把新快照记下(重算链第 ①b 步就是这条)
  15. python scripts/raw_scan.py --plan # 只打印"有变化时该跑哪些步"
  16. python scripts/raw_scan.py --json # 机器可读(rebuild_all 用它做"跳过了该跑的步"提醒)
  17. python scripts/raw_scan.py --deep # 变化判定加上**内容**哈希(默认只比 名字/大小/时间)
  18. python scripts/raw_scan.py --root <别的 raw> # 换个根扫(自测/多场)
  19. python scripts/raw_scan.py --selftest # 族表覆盖自检(与族表/链上步骤对账,防两处漂移)
  20. ## 口径与纪律
  21. · **指纹**默认 = 每族逐件 `相对路径|字节数|mtime(ns)` 排序后串起来取 sha1 —— 快、且"加了一件/改了一件/
  22. 删了一件"都会变;`--deep` 再叠一层内容 sha1(小文件才做,避免拿 153 GB 的 CMS 导出拼哈希)。
  23. · **未归类**:`data/raw/<场>/` 下出现族表里没有的子目录(新来的数据总会有第一次)→ 如实列进
  24. `unclassified` 并给出"没有消费者/需要认领"的话,**不猜**它是哪一族。
  25. · 这里只**发现与建议**,不改数据、不搬文件、不自动重算(要不要跑由 `rebuild_all.py` 或运维决定;
  26. 自动重算的开关见 `configs/serve.json` 的 `raw_watch`)。rc=4 是"有变化",不是失败。
  27. """
  28. from __future__ import annotations
  29. import argparse
  30. import hashlib
  31. import json
  32. import os
  33. import pathlib
  34. import re
  35. import sys
  36. import time
  37. ROOT = pathlib.Path(__file__).resolve().parents[1]
  38. sys.path.insert(0, str(ROOT))
  39. try: # GBK 控制台/日志重定向不再因 '²' 这类字符抛 UnicodeEncodeError
  40. sys.stdout.reconfigure(errors='replace')
  41. except Exception:
  42. pass
  43. SNAP_NAME = '_raw_scan.json'
  44. # ── 族表:raw 子目录 → (计件 glob, 链上步骤, 消费者说明) ──────────────────────────────
  45. # ★步骤标签与 `scripts/rebuild_all.py` 的计划名一一对应(人看到"③ SCADA 侧"就知道去跑哪一步)。
  46. FAMILIES: tuple[tuple[str, tuple[str, ...], tuple[str, ...], str], ...] = (
  47. ('故障报警', ('*.xls', '*.xlsx', '*.csv'), ('② 三门台账',),
  48. '报警事件导出 → windscada/alarms.parquet(报警集中/停机账)。缺 TimeOn/Alarmcode 的统计报表件会被摄入器跳过'),
  49. ('风机故障记录', ('*.xls', '*.xlsx', '*.csv'), ('② 三门台账',),
  50. '检修工单台账 → windscada/workorders.parquet(重复检修/停机损失)'),
  51. ('油样报告', ('*.pdf',), ('② 三门台账',),
  52. '油液化验报告 → windscada/oil_samples_index.parquet(油液面)'),
  53. ('scada_10min', ('*.csv',), ('③ SCADA 侧 10 个构建器', '④ 月度派生件', '④c 变桨面', '⑤b 事实契约 claim'),
  54. '10min 导出 → 功率曲线/损失/温度/停机/偏航/液压等 + 变桨面日粒度(温度/性能/变桨面)'),
  55. ('scada_1min', ('*.csv',), ('④c 变桨面',),
  56. '1min 导出(内部台号 01E…38B)→ 变桨面零位三口径与越线计数'),
  57. ('scada_mdb', ('*.mdb', '*.zip'), ('③ SCADA 侧 10 个构建器', '④ 月度派生件', '④c 变桨面'),
  58. '现场年度归档 Access 库(按月的 10min/1min)—— CSV 缺失时由 scada_source 回落到它'),
  59. ('windcms', ('*_decode.json',), ('④b 振动侧摄入 + CMS 报告 + 标量 z', '④d 振动在升/换件闭环', '④e 三层基线'),
  60. 'Brande TCM 导出 → 窗索引/谱库/CMS 报告/逐台页 + 在升闭环 + 三层基线(振动面)'),
  61. ('m5_cms_tcm', ('*.json', '*.docx', '*.pdf', '*.md'), ('④b 振动侧摄入 + CMS 报告 + 标量 z',),
  62. '现场给的正本(handoff_vibration_v2.json / component_history.json)与厂家报告——正本在位时优先于观澜自算'),
  63. )
  64. # 机理层资料不在场站目录下(A2 约定): data/raw/西门子4.0技术资料
  65. TECH_DIR, TECH_PATTERNS = '西门子4.0技术资料', ('*',) # 该族的消费者(kb_ingest)读整棵树 ⇒ 白名单就是"全部"
  66. TECH_STEPS = ('⑦ 本体: 码表/手册/文档',)
  67. TECH_NOTE = '厂商技术资料(故障处理手册/维护 WI/图纸/对译表)→ 本体对象库/检索索引/实机参数表'
  68. # raw 根下**随包发的说明件**(不是数据): 别把它们报成"未归类"。口径: 只有名字在白名单里且不参与摄入。
  69. IGNORE_FILES = {'README_把原始数据放这里.txt'}
  70. # ★2026-09-21 用户令「按月 10min 提取件就放在 <场站>/ 下」: 它是**给人核对/交接的导出件**, 不是摄入源
  71. # (摄入走 `<台号>.csv` + 同台补充件)。不加这条, 每次扫描都会把它报成"未归类 + 一处变化" ⇒
  72. # `--auto` / raw_watch 会永远以为"来了新数据", 每轮都触发一次全量重算。
  73. IGNORE_PATTERNS: tuple[tuple[str, str], ...] = (
  74. (r'^scada_\d+min_\d{6}\.csv$', '按月提取/核对件(非摄入源;摄入走 <台号>.csv + 同台补充件)'),
  75. )
  76. # ── 消费者取数口径(用户令 2026-09-20 步骤 C:让"放了没人读"当场可见)────────────────────────
  77. # ★为什么必须有它:族表原来只按**扩展名白名单**计件 —— 往 `scada_10min/` 里放 `WTG01-B2.csv`(856 列、
  78. # 2026-08 数据、与 `<台号>.csv` 命名不同)时,它命中 `*.csv` ⇒ 既算"进了摄入",又不会出现在任何
  79. # "没人读"的提示里;而取数层只认 `<台号>.csv` ⇒ **这批数据从来没被读过**,重算后时间窗自然不变。
  80. # 本表按**消费者真实取数口径**再判一次,并把结果分成两级:
  81. # · 缺口级 (gate):看着是数据(扩展名在 data_exts 里) 但没有任何消费者会读 ⇒ **计入变化(rc=4) + 大声报**
  82. # · 信息级 (info):附件/压缩包/图片/厂商软件这类,本来就不该被摄入 ⇒ 只列信息,不计变化
  83. CONSUME: dict[str, dict] = {
  84. '故障报警': dict(consume=('*.xls',), data_exts=('.xls', '.xlsx', '.csv'),
  85. how='摄入器读 *.xls 的 XML 事件导出(TimeOn/Alarmcode)'),
  86. '风机故障记录': dict(consume=('*.xls', '*.xlsx'), data_exts=('.xls', '.xlsx', '.csv'),
  87. how='台账读取 *.xls*(rar/jpg 是附件,不摄入)'),
  88. '油样报告': dict(consume=('*.pdf',), data_exts=('.pdf',), how='油样摄取读 *.pdf'),
  89. 'scada_10min': dict(consume=('WTG??.csv', 'WTG??-*.csv', 'WTG??_*.csv'), data_exts=('.csv', '.mdb', '.zip'),
  90. how='取数层读 `<台号>.csv` **+ 同台补充件** `<台号>-*.csv`(2026-09-20 起: 按列名对齐、'
  91. '按时间戳去重、主件优先)'),
  92. 'scada_1min': dict(consume=('???.csv', '???-*.csv', '???_*.csv'), data_exts=('.csv', '.mdb', '.zip'),
  93. how='取数层按内部台号取 `<台号>.csv`(01E.csv)+ 同台补充件'),
  94. 'scada_mdb': dict(consume=('*-10min.mdb', '*-1min.mdb', '*.zip'), data_exts=('.mdb', '.zip'),
  95. how='取数层读取按月的库(2026-08-10min.mdb);CSV 缺失时才回落',
  96. # ★2026-09-21: 现场按**类目**交付的月度通道组库(7月-8月交付 zip 里的
  97. # `2026-0X-<类>.mdb`)。它是 `scada_10min` 同台补充件的**上游**:取数层不直接读它,
  98. # 要先转成 `scada_10min/<台号>-<YYYYMM>.csv`。故列为"上游归档"(信息级),
  99. # 既不当缺口(不算变化/不触发重算),也不假装被摄入。
  100. upstream=tuple(f'*-{c}.mdb' for c in
  101. ('cnt', 'din', 'dot', 'flg', 'grd', 'int', 'prs', 'scd', 'std', 'sum', 'tmp', 'tur')),
  102. upstream_how='现场月度**通道组**归档(上游件):先转成同台 10min 补充件 '
  103. '`scada_10min/<台号>-<YYYYMM>.csv` 才被消费(口径见 docs/输入数据放置指导_v0.1.md §4.1)'),
  104. 'windcms': dict(consume=('*_decode.json', '*.pdf', '*.docx', '*.md', '*.json'),
  105. data_exts=('.json', '.zip', '.csv', '.mdb'),
  106. how='振动摄入读 *_decode.json;厂家报告 PDF 由报告转录侧读'),
  107. 'm5_cms_tcm': dict(consume=('*.json', '*.docx', '*.pdf', '*.md'), data_exts=('.json', '.zip'),
  108. how='现场正本 handoff/component_history 与厂家报告'),
  109. TECH_DIR: dict(consume=('*',), data_exts=(), how='本体读整棵树(白名单=全部)'),
  110. }
  111. def _digest(entries: list[tuple[str, int, int]], deep: bool, base: pathlib.Path, files: list[pathlib.Path],
  112. deep_max: int, dirs: list[str] | None = None) -> str:
  113. h = hashlib.sha1()
  114. for rel, size, mt in entries:
  115. h.update(f'{rel}|{size}|{mt}\n'.encode('utf-8', 'replace'))
  116. for d in dirs or []: # 子目录清单也进指纹: "新建了空目录(数据还没落)"也能发现
  117. h.update(f'D:{d}\n'.encode('utf-8', 'replace'))
  118. if deep:
  119. for p, (rel, size, _mt) in zip(files, entries):
  120. if size > deep_max:
  121. continue
  122. try:
  123. fh = hashlib.sha1()
  124. with open(p, 'rb') as f:
  125. for blk in iter(lambda: f.read(1 << 20), b''):
  126. fh.update(blk)
  127. h.update(rel.encode('utf-8', 'replace') + b'=' + fh.digest())
  128. except OSError:
  129. pass
  130. return h.hexdigest()
  131. def fingerprint(d: pathlib.Path, patterns: tuple[str, ...], deep: bool = False, deep_max: int = 8 << 20) -> dict:
  132. """一族目录的指纹。
  133. ★口径 (用户令 2026-09-19/20「对 data/raw 目录(含嵌套子目录)下,文件的增减做到监听」):
  134. · **递归**统计该族目录下**全部文件**(不限扩展名) —— 只按摄入白名单(patterns)统计的话,
  135. 往族目录里丢一个 `.zip`/`.txt` 就"看不见"了, 那不算监听; 白名单命中的另记 `matched`
  136. (才是真进摄入的件数), 差额用 `unmatched` 列出来给人看;
  137. · 指纹里**还含子目录清单**(只放路径名, 不放 mtime —— 目录 mtime 会随子文件变动而变, 那是噪声),
  138. 于是"新建了一个空目录(数据还没落进去)"也动指纹。
  139. """
  140. all_files: list[pathlib.Path] = []
  141. dirs: list[str] = []
  142. if d.is_dir():
  143. for p in d.rglob('*'):
  144. if p.is_file():
  145. all_files.append(p)
  146. elif p.is_dir():
  147. dirs.append(p.relative_to(d).as_posix())
  148. all_files = sorted(set(all_files))
  149. dirs.sort()
  150. matched: list[pathlib.Path] = []
  151. for pat in patterns:
  152. matched += [p for p in d.rglob(pat) if p.is_file()]
  153. mset = {p.relative_to(d).as_posix() for p in set(matched)}
  154. entries: list[tuple[str, int, int]] = []
  155. for p in all_files:
  156. try:
  157. st = p.stat()
  158. except OSError:
  159. continue
  160. entries.append((p.relative_to(d).as_posix(), st.st_size, st.st_mtime_ns))
  161. entries.sort()
  162. newest = ''
  163. if entries:
  164. newest = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(max(e[2] for e in entries) / 1e9))
  165. _unm = [e[0] for e in entries if e[0] not in mset]
  166. # ── 消费者口径再判一次(步骤 C): 扩展名"像数据"但没人读 ⇒ 缺口; 其余 ⇒ 信息 ──
  167. # ★判据必须落在"**全部文件里没被消费者读的**"上, 不能只看 unmatched(白名单没命中的那些) ——
  168. # 2026-09-20 实逮: `WTG01-B2.csv` 是 `*.csv`,白名单命中了它(所以 unmatched 里没有它),
  169. # 但取数层只认 `WTG01.csv` ⇒ 只看 unmatched 就永远发现不了这类"同名不同款"的静默漏读。
  170. spec = CONSUME.get(d.name) or {}
  171. cg = tuple(spec.get('consume') or ())
  172. dex = tuple(spec.get('data_exts') or ())
  173. consumed: set[str] = set()
  174. for g in cg:
  175. consumed |= {p.relative_to(d).as_posix() for p in d.rglob(g) if p.is_file()}
  176. # ── 上游归档件(客户口径里"已知但不由取数层直接读"的): 归信息级, 不算缺口、不计变化 ──
  177. upat = tuple(spec.get('upstream') or ())
  178. upstream: set[str] = set()
  179. for g in upat:
  180. upstream |= {p.relative_to(d).as_posix() for p in d.rglob(g) if p.is_file()}
  181. not_consumed = [r for r, _s, _m in entries if r not in consumed]
  182. gaps = [r for r in not_consumed
  183. if r not in upstream and pathlib.PurePosixPath(r).suffix.lower() in dex]
  184. others = [r for r in not_consumed if r not in gaps]
  185. return dict(files=len(entries), bytes=sum(e[1] for e in entries), newest=newest,
  186. matched=len(mset), dirs=len(dirs), unmatched=_unm[:10], unmatched_n=len(_unm),
  187. consumed=len(consumed), consume_how=spec.get('how', ''),
  188. upstream=len(upstream), upstream_how=spec.get('upstream_how', ''),
  189. gaps=gaps[:10], gaps_n=len(gaps), others=others[:10], others_n=len(others),
  190. digest=_digest(entries, deep, d, all_files, deep_max, dirs),
  191. top=[e[0] for e in sorted(entries, key=lambda x: -x[2])[:3]])
  192. def scan(raw_root: pathlib.Path, deep: bool = False, station_dir: pathlib.Path | None = None) -> dict:
  193. """扫一个 raw 根:逐族指纹 + 别处不认识的子目录(未归类)。
  194. `station_dir` 不给时按**场配置的辨识结果**定场站目录(`raw_station_dir()`,与摄入侧同一口径);
  195. 显式 `--root` 自测时可传入 `<root>/<场站名>` 之外的位置。
  196. """
  197. from src.windscada.config import raw_station_dir
  198. station = pathlib.Path(station_dir) if station_dir else raw_station_dir()
  199. fam: dict[str, dict] = {}
  200. for name, pats, steps, note in FAMILIES:
  201. fp = fingerprint(station / name, pats, deep)
  202. fp.update(steps=list(steps), note=note, where=str((station / name)))
  203. fam[name] = fp
  204. fp = fingerprint(raw_root / TECH_DIR, TECH_PATTERNS, deep)
  205. fp.update(steps=list(TECH_STEPS), note=TECH_NOTE, where=str(raw_root / TECH_DIR))
  206. fam[TECH_DIR] = fp
  207. known = {n for n, *_ in FAMILIES} | {TECH_DIR}
  208. unclassified = []
  209. known_dirs = {str(station / n) for n, *_ in FAMILIES} | {str(raw_root / TECH_DIR)}
  210. # ★未归类 = 任何**不在已知族目录里**的文件 (递归找, 不只看一层): raw 根下、场站目录下、以及
  211. # 场站目录里嵌套的新子目录(例如新来了 `CMS_202701` 这种整包目录) 都跑不掉。
  212. for p in (sorted(raw_root.rglob('*')) if raw_root.is_dir() else []):
  213. if not p.is_file():
  214. continue
  215. if any(str(p).startswith(k + os.sep) for k in known_dirs):
  216. continue
  217. if p.name in IGNORE_FILES:
  218. continue # 随包的放置说明件: 不是数据, 不报"未归类"
  219. if any(re.match(rx, p.name) for rx, _why in IGNORE_PATTERNS):
  220. continue # 交付核对用的导出件(位置由用户指定): 非摄入源, 不报"未归类"
  221. rel = p.relative_to(raw_root).as_posix()
  222. unclassified.append(dict(dir=rel, files=1,
  223. note='不在任何已知族的目录里 —— 没有消费者; 是新的输入族就得给它配摄入器'))
  224. # 已知族目录内**消费者不会读**的件 (步骤 C): 两级 —— 缺口级(gaps, 扩展名像数据) 与 信息级(others)
  225. stray, gaps, upstream = [], [], []
  226. for name, fp in fam.items():
  227. if fp.get('others_n'):
  228. stray.append(dict(family=name, n=fp['others_n'], example=fp.get('others')[:5]))
  229. if fp.get('gaps_n'):
  230. gaps.append(dict(family=name, n=fp['gaps_n'], example=fp.get('gaps')[:8],
  231. how=fp.get('consume_how', '')))
  232. if fp.get('upstream'):
  233. upstream.append(dict(family=name, n=fp['upstream'], how=fp.get('upstream_how', '')))
  234. # 场站目录下**新增的未知子目录**(整目录级; 其文件已在上面逐件登记, 这里给个汇总便于人读)
  235. extra_dirs = []
  236. if station.is_dir():
  237. for sub in sorted(station.iterdir()):
  238. if sub.is_dir() and sub.name not in known:
  239. extra_dirs.append(dict(dir=f'{station.name}/{sub.name}',
  240. files=sum(1 for p in sub.rglob('*') if p.is_file()),
  241. note='场站目录下的新子目录(族表里没有)'))
  242. return dict(at=time.strftime('%Y-%m-%d %H:%M:%S'), root=str(raw_root), station=str(station),
  243. deep=deep, families=fam, unclassified=unclassified + extra_dirs,
  244. stray_in_families=stray, gaps=gaps, upstream=upstream)
  245. def diff(old: dict, new: dict) -> list[dict]:
  246. """逐族比:新增/减少/内容变化(`files` 与 `digest` 分开说,便于人看懂)。"""
  247. out = []
  248. of = (old or {}).get('families') or {}
  249. for name, cur in (new.get('families') or {}).items():
  250. prev = of.get(name)
  251. if not prev:
  252. if cur['files']:
  253. out.append(dict(family=name, kind='首次记录', files=cur['files'], steps=cur['steps'],
  254. note=cur['note'], newest=cur['newest']))
  255. continue
  256. d_files = cur['files'] - prev['files']
  257. same_list = cur['digest'] == prev['digest']
  258. if d_files == 0 and same_list:
  259. continue
  260. kind = ('新增 %+d 件' % d_files) if d_files else '内容变化(件数不变)'
  261. _dd = cur.get('dirs', 0) - prev.get('dirs', 0)
  262. if d_files == 0 and _dd:
  263. kind = f'目录结构变化({_dd:+d} 个子目录)'
  264. out.append(dict(family=name, kind=kind, files=cur['files'], delta=d_files,
  265. bytes_delta=cur['bytes'] - prev['bytes'], newest=cur['newest'],
  266. top=cur.get('top'), steps=cur['steps'], note=cur['note'],
  267. matched=cur.get('matched'), unmatched_n=cur.get('unmatched_n', 0)))
  268. for u in new.get('unclassified') or []:
  269. out.append(dict(family=u['dir'], kind='未归类(不在任何已知族的目录里)', files=u['files'],
  270. steps=[], note=u.get('note') or '没有已知消费者 ⇒ 先认领'))
  271. # ★缺口级 (步骤 C, **计入变化**): 扩展名看着是数据、但按消费者口径没人会读 ⇒ 重算也不会纳入。
  272. # 典型: scada_10min/ 里的 `WTG01-B2.csv`(856 列 / 2026-08 数据) —— 取数层只认 `WTG01.csv`。
  273. for g in new.get('gaps') or []:
  274. out.append(dict(family=f'{g["family"]}(数据没人读·缺口)', kind=f'{g["n"]} 件数据不会被摄入',
  275. files=g['n'], steps=[],
  276. note=f'消费者口径: {g["how"]} ⇒ 例: ' + '、'.join(g['example'])
  277. + ' —— 这些件**不会进任何产物**(重算也不会变); 要么按消费者口径改名/转换, '
  278. '要么给取数层加"同台多件"的合并口径'))
  279. # ★附件类(信息级,**不计入变化**): 实测族目录里本来就有大量正常但不该被摄入的东西
  280. # (故障记录里的 .rar/.jpg、windcms 厂家报告 PDF、技术资料里的厂商软件 .exe/.lp …)。
  281. # 它们增删本身已经反映在 files/dirs/digest 上 ⇒ 该报的变化照报, 这里不再重复判一次。
  282. return out
  283. def _steps_of(changes: list[dict]) -> list[str]:
  284. seen, out = set(), []
  285. for c in changes:
  286. for s in c.get('steps') or []:
  287. if s not in seen:
  288. seen.add(s)
  289. out.append(s)
  290. return out
  291. def snap_path(raw_root: pathlib.Path) -> pathlib.Path:
  292. from src import paths as P
  293. return P.out_root() / SNAP_NAME
  294. def main() -> int:
  295. ap = argparse.ArgumentParser(description='输入数据自动扫描识别(发现新增 → 指明该重跑哪几步)')
  296. ap.add_argument('--farm', default=None)
  297. ap.add_argument('--root', default=None, help='覆盖 data/raw 根(自测/多场)')
  298. ap.add_argument('--write', action='store_true', help='把本次指纹写进快照')
  299. ap.add_argument('--check', action='store_true', help='与快照比:rc=0 无变化 · 4 有新增/变化 · 5 无快照')
  300. ap.add_argument('--plan', action='store_true', help='只打印"该跑哪些步"')
  301. ap.add_argument('--json', action='store_true', help='输出机器可读 JSON')
  302. ap.add_argument('--deep', action='store_true', help='变化判定叠内容哈希(小文件)')
  303. ap.add_argument('--selftest', action='store_true', help='族表/步骤覆盖自检(防与族表、链上步骤漂移)')
  304. a = ap.parse_args()
  305. from src import paths as P
  306. from src.windscada import config as C
  307. if a.farm:
  308. os.environ['WINDSCADA_FARM'] = a.farm # 与其它脚本同一口径(场由环境变量选)
  309. raw = pathlib.Path(a.root) if a.root else pathlib.Path(C.RAW_ROOT)
  310. # --root 换成别的 raw 根时,场站目录按"根下唯一/同名"识别,避免拿本场辨识结果去别处找
  311. st_dir = None
  312. if a.root:
  313. from src.windscada.config import station_scan
  314. sc = station_scan(root=raw)
  315. st_dir = (raw / sc['matched']) if sc.get('matched') else None
  316. now = scan(raw, deep=a.deep, station_dir=st_dir)
  317. sp = snap_path(raw)
  318. old = None
  319. if sp.is_file():
  320. try:
  321. old = json.loads(sp.read_text(encoding='utf-8'))
  322. except Exception:
  323. old = None
  324. changes = diff(old, now)
  325. if a.selftest:
  326. return selftest()
  327. if a.json:
  328. print(json.dumps(dict(scan=now, changes=changes, snapshot=str(sp),
  329. steps=_steps_of(changes), has_snapshot=bool(old)), ensure_ascii=False, indent=1))
  330. if a.write:
  331. sp.parent.mkdir(parents=True, exist_ok=True)
  332. sp.write_text(json.dumps(now, ensure_ascii=False, indent=1), encoding='utf-8')
  333. if a.check:
  334. return 0 if (old and not changes) else (4 if old else 5)
  335. return 0
  336. print(f'输入数据扫描 · {now["root"]} ({now["at"]}{" · deep" if a.deep else ""})')
  337. tot_f = sum(f['files'] for f in now['families'].values())
  338. tot_b = sum(f['bytes'] for f in now['families'].values())
  339. print(f' 共 {len(now["families"])} 族 · {tot_f} 件 · {tot_b / 1024 / 1024 / 1024:.1f} GB')
  340. for name, f in now['families'].items():
  341. flag = ''
  342. for c in changes:
  343. if c['family'] == name:
  344. flag = f" ← {c['kind']}"
  345. print(f' {name:16s} {f["files"]:6d} 件 {f["bytes"] / 1024 / 1024:9.1f} MB 最新 {f["newest"] or "—"}{flag}')
  346. if now['unclassified']:
  347. print(' 未归类(族表里没有,先认领):')
  348. for u in now['unclassified']:
  349. print(f' {u["dir"]:16s} {u["files"]:6d} 件 {u.get("note", "没有已知消费者")}')
  350. if now.get('gaps'):
  351. print(' ★[缺口] 族目录里有**看着是数据、但没有任何消费者会读**的件(重算不会纳入它们):')
  352. for g in now['gaps']:
  353. print(f' {g["family"]:16s} {g["n"]:5d} 件 例: ' + '、'.join(g['example'][:4])
  354. + (' …' if g['n'] > 4 else ''))
  355. print(f' 消费者口径: {g["how"]}')
  356. if now.get('upstream'):
  357. print(' (信息,不计入"变化")族目录里的**上游归档件**(要先转换才被消费):')
  358. for u in now['upstream']:
  359. print(f' {u["family"]:16s} {u["n"]:5d} 件')
  360. print(f' {u["how"]}')
  361. if now.get('stray_in_families'):
  362. print(' (信息,不计入"变化")族目录里**消费者不读的附件类**件:')
  363. for s in now['stray_in_families']:
  364. print(f' {s["family"]:16s} {s["n"]:5d} 件 例: ' + '、'.join(s['example'][:3]))
  365. if changes:
  366. print(f'\n== 本次发现 {len(changes)} 处新增/变化 ==')
  367. for c in changes:
  368. print(f' · {c["family"]}: {c["kind"]}'
  369. + (f'(体积 {c["bytes_delta"] / 1024 / 1024:+.1f} MB)' if c.get('bytes_delta') else '')
  370. + f' 最新 {c.get("newest") or "—"}')
  371. if c.get('top'):
  372. print(f' 最近落盘: ' + '、'.join(c['top']))
  373. if c.get('steps'):
  374. print(f' 该跑: ' + ' / '.join(c['steps']))
  375. _real = [c for c in changes if '缺口' not in c['family']]
  376. if _real:
  377. print('\n 建议: 跑一次重算把新数据纳入产物 —— python scripts/rebuild_all.py'
  378. '(默认全跑;若你用了 --skip-*,请确认没跳过上面列出的步)')
  379. if len(_real) < len(changes):
  380. print(' ★缺口类变化**重算也纳入不了**(消费者根本不读这些件): 先按上面写的口径改名/转换, '
  381. '或给取数层加"同台多件"的合并口径(见 docs/输入数据放置指导_v0.1.md §4)')
  382. else:
  383. print('\n== 与上次快照一致,没有新数据 ==' if old else '\n== 还没有快照(跑 --write 记一次基线)==')
  384. if a.write:
  385. sp.parent.mkdir(parents=True, exist_ok=True)
  386. sp.write_text(json.dumps(now, ensure_ascii=False, indent=1), encoding='utf-8')
  387. print(f'\n快照已更新 → {P.rel(sp)}')
  388. if a.check:
  389. if not old:
  390. print(f'[?] 没有快照可比({P.rel(sp)})—— 先跑 --write 记基线; rc=5')
  391. return 5
  392. return 4 if changes else 0
  393. return 0
  394. def selftest() -> int:
  395. """自检:① 族表覆盖 `products_reverse_audit.FAMILIES` 里所有 raw 输入目录 ② 步骤名都能在链上找到。"""
  396. import importlib.util
  397. import re
  398. spec = importlib.util.spec_from_file_location('_pra', ROOT / 'scripts' / 'products_reverse_audit.py')
  399. mod = importlib.util.module_from_spec(spec)
  400. spec.loader.exec_module(mod)
  401. inputs = {f['input'] for f in mod.FAMILIES if f.get('input')}
  402. mine = {n for n, *_ in FAMILIES} | {TECH_DIR}
  403. missing = sorted(inputs - mine)
  404. chain = (ROOT / 'scripts' / 'rebuild_all.py').read_text(encoding='utf-8')
  405. steps = {s for _n, _p, ss, _note in FAMILIES for s in ss} | set(TECH_STEPS)
  406. bad_steps = sorted(s for s in steps if s.split()[0] not in chain)
  407. print(f'族表覆盖: 反向呼应族表的 raw 输入 {len(inputs)} 个 · 本器覆盖 {len(mine)} 个 · 缺 {missing or "无"}')
  408. print(f'步骤标签: {len(steps)} 个 · 在 rebuild_all 里找不到的 {bad_steps or "无"}')
  409. ok = not missing and not bad_steps
  410. print('[OK] 自检通过' if ok else '[X] 自检不过 —— 族表或步骤标签与链上漂移了')
  411. return 0 if ok else 5
  412. if __name__ == '__main__':
  413. sys.exit(main())