guanlan-src/scripts/raw_scan.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""输入数据自动扫描识别 —— 发现 `<安装目录>/data/raw` 下的**新增/变化**，并给出"该重跑哪几步"。

## 为什么要有它（用户令 2026-09-19）

用户令：「对 `<安装目录>/data/raw` 目录下的接入数据处理，增加自动扫描识别机制，以能发现新增数据，并纳入重算。」

在此之前，"放了新数据要不要重算"全靠人记：`place_raw_data.py` 只管把现场包落位，`raw_data_check.py`
只管结构与命名对不对，`rebuild_all.py` 无脑全跑（或按人给的 `--skip-*` 跳过）——
**没有任何一处回答"这回新来了什么、因此必须重跑哪几步"**。本脚本补这一环：

    逐族指纹 (件数/体积/最新时间/清单摘要) → 与上次快照比 → 报【新增/变化】 → 映射到链上步骤

## 三档用法

    python scripts/raw_scan.py                     # 只报当前状态（不写快照）
    python scripts/raw_scan.py --write             # 写/更新快照 outputs/<场>/_raw_scan.json
    python scripts/raw_scan.py --check             # 与快照比: rc=0 无变化 · rc=4 有新增/变化 · rc=5 没有快照
    python scripts/raw_scan.py --check --write      # 比完就把新快照记下（重算链第 ①b 步就是这条）
    python scripts/raw_scan.py --plan               # 只打印"有变化时该跑哪些步"
    python scripts/raw_scan.py --json               # 机器可读（rebuild_all 用它做"跳过了该跑的步"提醒）
    python scripts/raw_scan.py --deep               # 变化判定加上**内容**哈希（默认只比 名字/大小/时间）
    python scripts/raw_scan.py --root <别的 raw>    # 换个根扫（自测/多场）
    python scripts/raw_scan.py --selftest           # 族表覆盖自检（与族表/链上步骤对账，防两处漂移）

## 口径与纪律

· **指纹**默认 = 每族逐件 `相对路径|字节数|mtime(ns)` 排序后串起来取 sha1 —— 快、且"加了一件/改了一件/
  删了一件"都会变；`--deep` 再叠一层内容 sha1（小文件才做，避免拿 153 GB 的 CMS 导出拼哈希）。
· **未归类**：`data/raw/<场>/` 下出现族表里没有的子目录（新来的数据总会有第一次）→ 如实列进
  `unclassified` 并给出"没有消费者/需要认领"的话，**不猜**它是哪一族。
· 这里只**发现与建议**，不改数据、不搬文件、不自动重算（要不要跑由 `rebuild_all.py` 或运维决定；
  自动重算的开关见 `configs/serve.json` 的 `raw_watch`）。rc=4 是"有变化"，不是失败。
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import pathlib
import sys
import time

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
try:                                     # GBK 控制台/日志重定向不再因 '²' 这类字符抛 UnicodeEncodeError
    sys.stdout.reconfigure(errors='replace')
except Exception:
    pass

SNAP_NAME = '_raw_scan.json'

# ── 族表：raw 子目录 → (计件 glob, 链上步骤, 消费者说明) ──────────────────────────────
# ★步骤标签与 `scripts/rebuild_all.py` 的计划名一一对应（人看到"③ SCADA 侧"就知道去跑哪一步）。
FAMILIES: tuple[tuple[str, tuple[str, ...], tuple[str, ...], str], ...] = (
    ('故障报警', ('*.xls', '*.xlsx', '*.csv'), ('② 三门台账',),
     '报警事件导出 → windscada/alarms.parquet（报警集中/停机账）。缺 TimeOn/Alarmcode 的统计报表件会被摄入器跳过'),
    ('风机故障记录', ('*.xls', '*.xlsx', '*.csv'), ('② 三门台账',),
     '检修工单台账 → windscada/workorders.parquet（重复检修/停机损失）'),
    ('油样报告', ('*.pdf',), ('② 三门台账',),
     '油液化验报告 → windscada/oil_samples_index.parquet（油液面）'),
    ('scada_10min', ('*.csv',), ('③ SCADA 侧 10 个构建器', '④ 月度派生件', '④c 变桨面', '⑤b 事实契约 claim'),
     '10min 导出 → 功率曲线/损失/温度/停机/偏航/液压等 + 变桨面日粒度（温度/性能/变桨面）'),
    ('scada_1min', ('*.csv',), ('④c 变桨面',),
     '1min 导出（内部台号 01E…38B）→ 变桨面零位三口径与越线计数'),
    ('scada_mdb', ('*.mdb', '*.zip'), ('③ SCADA 侧 10 个构建器', '④ 月度派生件', '④c 变桨面'),
     '现场年度归档 Access 库（按月的 10min/1min）—— CSV 缺失时由 scada_source 回落到它'),
    ('windcms', ('*_decode.json',), ('④b 振动侧摄入 + CMS 报告 + 标量 z', '④d 振动在升/换件闭环', '④e 三层基线'),
     'Brande TCM 导出 → 窗索引/谱库/CMS 报告/逐台页 + 在升闭环 + 三层基线（振动面）'),
    ('m5_cms_tcm', ('*.json', '*.docx', '*.pdf', '*.md'), ('④b 振动侧摄入 + CMS 报告 + 标量 z',),
     '现场给的正本（handoff_vibration_v2.json / component_history.json）与厂家报告——正本在位时优先于观澜自算'),
)
# 机理层资料不在场站目录下（A2 约定）: data/raw/西门子4.0技术资料
TECH_DIR, TECH_PATTERNS = '西门子4.0技术资料', ('*',)   # 该族的消费者(kb_ingest)读整棵树 ⇒ 白名单就是"全部"
TECH_STEPS = ('⑦ 本体: 码表/手册/文档',)
TECH_NOTE = '厂商技术资料（故障处理手册/维护 WI/图纸/对译表）→ 本体对象库/检索索引/实机参数表'
# raw 根下**随包发的说明件**（不是数据）: 别把它们报成"未归类"。口径: 只有名字在白名单里且不参与摄入。
IGNORE_FILES = {'README_把原始数据放这里.txt'}


def _digest(entries: list[tuple[str, int, int]], deep: bool, base: pathlib.Path, files: list[pathlib.Path],
            deep_max: int, dirs: list[str] | None = None) -> str:
    h = hashlib.sha1()
    for rel, size, mt in entries:
        h.update(f'{rel}|{size}|{mt}\n'.encode('utf-8', 'replace'))
    for d in dirs or []:                       # 子目录清单也进指纹: "新建了空目录(数据还没落)"也能发现
        h.update(f'D:{d}\n'.encode('utf-8', 'replace'))
    if deep:
        for p, (rel, size, _mt) in zip(files, entries):
            if size > deep_max:
                continue
            try:
                fh = hashlib.sha1()
                with open(p, 'rb') as f:
                    for blk in iter(lambda: f.read(1 << 20), b''):
                        fh.update(blk)
                h.update(rel.encode('utf-8', 'replace') + b'=' + fh.digest())
            except OSError:
                pass
    return h.hexdigest()


def fingerprint(d: pathlib.Path, patterns: tuple[str, ...], deep: bool = False, deep_max: int = 8 << 20) -> dict:
    """一族目录的指纹。

    ★口径 (用户令 2026-09-19/20「对 data/raw 目录（含嵌套子目录）下，文件的增减做到监听」):
      · **递归**统计该族目录下**全部文件**(不限扩展名) —— 只按摄入白名单(patterns)统计的话,
        往族目录里丢一个 `.zip`/`.txt` 就"看不见"了, 那不算监听; 白名单命中的另记 `matched`
        (才是真进摄入的件数), 差额用 `unmatched` 列出来给人看;
      · 指纹里**还含子目录清单**(只放路径名, 不放 mtime —— 目录 mtime 会随子文件变动而变, 那是噪声),
        于是"新建了一个空目录(数据还没落进去)"也动指纹。
    """
    all_files: list[pathlib.Path] = []
    dirs: list[str] = []
    if d.is_dir():
        for p in d.rglob('*'):
            if p.is_file():
                all_files.append(p)
            elif p.is_dir():
                dirs.append(p.relative_to(d).as_posix())
    all_files = sorted(set(all_files))
    dirs.sort()
    matched: list[pathlib.Path] = []
    for pat in patterns:
        matched += [p for p in d.rglob(pat) if p.is_file()]
    mset = {p.relative_to(d).as_posix() for p in set(matched)}
    entries: list[tuple[str, int, int]] = []
    for p in all_files:
        try:
            st = p.stat()
        except OSError:
            continue
        entries.append((p.relative_to(d).as_posix(), st.st_size, st.st_mtime_ns))
    entries.sort()
    newest = ''
    if entries:
        newest = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(max(e[2] for e in entries) / 1e9))
    _unm = [e[0] for e in entries if e[0] not in mset]
    return dict(files=len(entries), bytes=sum(e[1] for e in entries), newest=newest,
                matched=len(mset), dirs=len(dirs), unmatched=_unm[:10], unmatched_n=len(_unm),
                digest=_digest(entries, deep, d, all_files, deep_max, dirs),
                top=[e[0] for e in sorted(entries, key=lambda x: -x[2])[:3]])


def scan(raw_root: pathlib.Path, deep: bool = False, station_dir: pathlib.Path | None = None) -> dict:
    """扫一个 raw 根：逐族指纹 + 别处不认识的子目录（未归类）。

    `station_dir` 不给时按**场配置的辨识结果**定场站目录（`raw_station_dir()`，与摄入侧同一口径）；
    显式 `--root` 自测时可传入 `<root>/<场站名>` 之外的位置。
    """
    from src.windscada.config import raw_station_dir
    station = pathlib.Path(station_dir) if station_dir else raw_station_dir()
    fam: dict[str, dict] = {}
    for name, pats, steps, note in FAMILIES:
        fp = fingerprint(station / name, pats, deep)
        fp.update(steps=list(steps), note=note, where=str((station / name)))
        fam[name] = fp
    fp = fingerprint(raw_root / TECH_DIR, TECH_PATTERNS, deep)
    fp.update(steps=list(TECH_STEPS), note=TECH_NOTE, where=str(raw_root / TECH_DIR))
    fam[TECH_DIR] = fp
    known = {n for n, *_ in FAMILIES} | {TECH_DIR}
    unclassified = []
    known_dirs = {str(station / n) for n, *_ in FAMILIES} | {str(raw_root / TECH_DIR)}
    # ★未归类 = 任何**不在已知族目录里**的文件 (递归找, 不只看一层): raw 根下、场站目录下、以及
    #   场站目录里嵌套的新子目录(例如新来了 `CMS_202701` 这种整包目录) 都跑不掉。
    for p in (sorted(raw_root.rglob('*')) if raw_root.is_dir() else []):
        if not p.is_file():
            continue
        if any(str(p).startswith(k + os.sep) for k in known_dirs):
            continue
        if p.name in IGNORE_FILES:
            continue                       # 随包的放置说明件: 不是数据, 不报"未归类"
        rel = p.relative_to(raw_root).as_posix()
        unclassified.append(dict(dir=rel, files=1,
                                 note='不在任何已知族的目录里 —— 没有消费者; 是新的输入族就得给它配摄入器'))
    # 已知族目录内**白名单没命中**的文件: 能进摄入的才计数, 这些是"放了但没人读"的
    stray = []
    for name, fp in fam.items():
        if fp.get('unmatched_n'):
            stray.append(dict(family=name, n=fp['unmatched_n'], example=fp.get('unmatched')[:5]))
    # 场站目录下**新增的未知子目录**(整目录级; 其文件已在上面逐件登记, 这里给个汇总便于人读)
    extra_dirs = []
    if station.is_dir():
        for sub in sorted(station.iterdir()):
            if sub.is_dir() and sub.name not in known:
                extra_dirs.append(dict(dir=f'{station.name}/{sub.name}',
                                       files=sum(1 for p in sub.rglob('*') if p.is_file()),
                                       note='场站目录下的新子目录（族表里没有）'))
    return dict(at=time.strftime('%Y-%m-%d %H:%M:%S'), root=str(raw_root), station=str(station),
                deep=deep, families=fam, unclassified=unclassified + extra_dirs, stray_in_families=stray)


def diff(old: dict, new: dict) -> list[dict]:
    """逐族比：新增/减少/内容变化（`files` 与 `digest` 分开说，便于人看懂）。"""
    out = []
    of = (old or {}).get('families') or {}
    for name, cur in (new.get('families') or {}).items():
        prev = of.get(name)
        if not prev:
            if cur['files']:
                out.append(dict(family=name, kind='首次记录', files=cur['files'], steps=cur['steps'],
                                note=cur['note'], newest=cur['newest']))
            continue
        d_files = cur['files'] - prev['files']
        same_list = cur['digest'] == prev['digest']
        if d_files == 0 and same_list:
            continue
        kind = ('新增 %+d 件' % d_files) if d_files else '内容变化(件数不变)'
        _dd = cur.get('dirs', 0) - prev.get('dirs', 0)
        if d_files == 0 and _dd:
            kind = f'目录结构变化({_dd:+d} 个子目录)'
        out.append(dict(family=name, kind=kind, files=cur['files'], delta=d_files,
                        bytes_delta=cur['bytes'] - prev['bytes'], newest=cur['newest'],
                        top=cur.get('top'), steps=cur['steps'], note=cur['note'],
                        matched=cur.get('matched'), unmatched_n=cur.get('unmatched_n', 0)))
    for u in new.get('unclassified') or []:
        out.append(dict(family=u['dir'], kind='未归类（不在任何已知族的目录里）', files=u['files'],
                        steps=[], note=u.get('note') or '没有已知消费者 ⇒ 先认领'))
    # ★族内"白名单外"的件**不计入变化**（信息而已）: 实测族目录里本来就有大量正常但不被摄入器读的东西
    #   （故障记录里的 .rar/.jpg、windcms 厂家报告 PDF、技术资料里的厂商软件 .exe/.lp …）。
    #   它们增删本身已经反映在 files/dirs/digest 上 ⇒ 该报的变化照报, 这里不再重复判一次。
    return out


def _steps_of(changes: list[dict]) -> list[str]:
    seen, out = set(), []
    for c in changes:
        for s in c.get('steps') or []:
            if s not in seen:
                seen.add(s)
                out.append(s)
    return out


def snap_path(raw_root: pathlib.Path) -> pathlib.Path:
    from src import paths as P
    return P.out_root() / SNAP_NAME


def main() -> int:
    ap = argparse.ArgumentParser(description='输入数据自动扫描识别（发现新增 → 指明该重跑哪几步）')
    ap.add_argument('--farm', default=None)
    ap.add_argument('--root', default=None, help='覆盖 data/raw 根（自测/多场）')
    ap.add_argument('--write', action='store_true', help='把本次指纹写进快照')
    ap.add_argument('--check', action='store_true', help='与快照比：rc=0 无变化 · 4 有新增/变化 · 5 无快照')
    ap.add_argument('--plan', action='store_true', help='只打印"该跑哪些步"')
    ap.add_argument('--json', action='store_true', help='输出机器可读 JSON')
    ap.add_argument('--deep', action='store_true', help='变化判定叠内容哈希（小文件）')
    ap.add_argument('--selftest', action='store_true', help='族表/步骤覆盖自检（防与族表、链上步骤漂移）')
    a = ap.parse_args()

    from src import paths as P
    from src.windscada import config as C
    if a.farm:
        os.environ['WINDSCADA_FARM'] = a.farm            # 与其它脚本同一口径（场由环境变量选）
    raw = pathlib.Path(a.root) if a.root else pathlib.Path(C.RAW_ROOT)
    # --root 换成别的 raw 根时，场站目录按"根下唯一/同名"识别，避免拿本场辨识结果去别处找
    st_dir = None
    if a.root:
        from src.windscada.config import station_scan
        sc = station_scan(root=raw)
        st_dir = (raw / sc['matched']) if sc.get('matched') else None
    now = scan(raw, deep=a.deep, station_dir=st_dir)
    sp = snap_path(raw)
    old = None
    if sp.is_file():
        try:
            old = json.loads(sp.read_text(encoding='utf-8'))
        except Exception:
            old = None
    changes = diff(old, now)

    if a.selftest:
        return selftest()

    if a.json:
        print(json.dumps(dict(scan=now, changes=changes, snapshot=str(sp),
                              steps=_steps_of(changes), has_snapshot=bool(old)), ensure_ascii=False, indent=1))
        if a.write:
            sp.parent.mkdir(parents=True, exist_ok=True)
            sp.write_text(json.dumps(now, ensure_ascii=False, indent=1), encoding='utf-8')
        if a.check:
            return 0 if (old and not changes) else (4 if old else 5)
        return 0

    print(f'输入数据扫描 · {now["root"]}  ({now["at"]}{" · deep" if a.deep else ""})')
    tot_f = sum(f['files'] for f in now['families'].values())
    tot_b = sum(f['bytes'] for f in now['families'].values())
    print(f'  共 {len(now["families"])} 族 · {tot_f} 件 · {tot_b / 1024 / 1024 / 1024:.1f} GB')
    for name, f in now['families'].items():
        flag = ''
        for c in changes:
            if c['family'] == name:
                flag = f"  ← {c['kind']}"
        print(f'    {name:16s} {f["files"]:6d} 件  {f["bytes"] / 1024 / 1024:9.1f} MB  最新 {f["newest"] or "—"}{flag}')
    if now['unclassified']:
        print('  未归类（族表里没有，先认领）:')
        for u in now['unclassified']:
            print(f'    {u["dir"]:16s} {u["files"]:6d} 件  {u.get("note", "没有已知消费者")}')
    if now.get('stray_in_families'):
        print('  （信息，不计入"变化"）族目录里**摄入白名单没命中**的件 —— 摄入器不会读它们:')
        for s in now['stray_in_families']:
            print(f'    {s["family"]:16s} {s["n"]:5d} 件  例: ' + '、'.join(s['example'][:3]))

    if changes:
        print(f'\n== 本次发现 {len(changes)} 处新增/变化 ==')
        for c in changes:
            print(f'  · {c["family"]}: {c["kind"]}'
                  + (f'（体积 {c["bytes_delta"] / 1024 / 1024:+.1f} MB）' if c.get('bytes_delta') else '')
                  + f'  最新 {c.get("newest") or "—"}')
            if c.get('top'):
                print(f'      最近落盘: ' + '、'.join(c['top']))
            if c.get('steps'):
                print(f'      该跑: ' + ' / '.join(c['steps']))
        print('\n  建议: 跑一次重算把新数据纳入产物 —— python scripts/rebuild_all.py'
              '（默认全跑；若你用了 --skip-*，请确认没跳过上面列出的步）')
    else:
        print('\n== 与上次快照一致，没有新数据 ==' if old else '\n== 还没有快照（跑 --write 记一次基线）==')

    if a.write:
        sp.parent.mkdir(parents=True, exist_ok=True)
        sp.write_text(json.dumps(now, ensure_ascii=False, indent=1), encoding='utf-8')
        print(f'\n快照已更新 → {P.rel(sp)}')
    if a.check:
        if not old:
            print(f'[?] 没有快照可比（{P.rel(sp)}）—— 先跑 --write 记基线; rc=5')
            return 5
        return 4 if changes else 0
    return 0


def selftest() -> int:
    """自检：① 族表覆盖 `products_reverse_audit.FAMILIES` 里所有 raw 输入目录 ② 步骤名都能在链上找到。"""
    import importlib.util
    import re
    spec = importlib.util.spec_from_file_location('_pra', ROOT / 'scripts' / 'products_reverse_audit.py')
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    inputs = {f['input'] for f in mod.FAMILIES if f.get('input')}
    mine = {n for n, *_ in FAMILIES} | {TECH_DIR}
    missing = sorted(inputs - mine)
    chain = (ROOT / 'scripts' / 'rebuild_all.py').read_text(encoding='utf-8')
    steps = {s for _n, _p, ss, _note in FAMILIES for s in ss} | set(TECH_STEPS)
    bad_steps = sorted(s for s in steps if s.split()[0] not in chain)
    print(f'族表覆盖: 反向呼应族表的 raw 输入 {len(inputs)} 个 · 本器覆盖 {len(mine)} 个 · 缺 {missing or "无"}')
    print(f'步骤标签: {len(steps)} 个 · 在 rebuild_all 里找不到的 {bad_steps or "无"}')
    ok = not missing and not bad_steps
    print('[OK] 自检通过' if ok else '[X] 自检不过 —— 族表或步骤标签与链上漂移了')
    return 0 if ok else 5


if __name__ == '__main__':
    sys.exit(main())