guanlan-src/scripts/scan_stations.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""场站目录扫描辨识 —— 看观澜系统在 data/raw 下认出了什么 (2026-09-11 用户令)。

约定: **`data/raw/` 的下一级目录就是「场站名称」**, 其下五个约定子目录是摄入接口:

    data/raw/<场站名称>/scada_10min/      SCADA 10min 导出 (逐台 WTG*.csv)
    data/raw/<场站名称>/scada_1min/        (可选) 1min 导出
    data/raw/<场站名称>/故障报警/          报警事件导出 (SpreadsheetML *.xls)
    data/raw/<场站名称>/风机故障记录/      检修工单台账 (*.xls/xlsx)
    data/raw/<场站名称>/油样报告/          油液化验报告 (*.pdf)

辨识依据 (逐条命中即止): ① 场配置 raw_station 全等 → ② 与场名写法(srg_farm_names)互为子串 →
③ 只有一个场站目录 (单站部署) → ④ 都不过 = 不猜, 报"未识别"。

用法:
    python scripts/scan_stations.py                 # 扫 data/raw, 打印辨识结论与各子目录件数
    python scripts/scan_stations.py --json          # 机器可读
    python scripts/scan_stations.py --root D:\\别的raw
"""
from __future__ import annotations

import argparse
import json
import pathlib
import sys

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('--root', default=None, help='覆盖 data/raw 根 (默认取场配置)')
    ap.add_argument('--json', action='store_true')
    a = ap.parse_args()

    from src.windscada import config as C
    from src.windscada.config import farm, station_report, STATION_SUBDIRS

    cfg = farm()
    rep = station_report(cfg, a.root)
    if a.json:
        print(json.dumps({k: (str(v) if isinstance(v, pathlib.Path) else v)
                          for k, v in rep.items() if k != 'stations'}
                         | {'stations': {k: dict(n_files=v['n_files'],
                                                 dir=str(v['dir']),
                                                 subdirs=sorted(v['subdirs']))
                                         for k, v in rep['stations'].items()}},
                         ensure_ascii=False, indent=2))
        return 0

    print(f'原始件根 : {pathlib.Path(a.root) if a.root else C.RAW_ROOT}')
    print(f'当前场   : {cfg["name"]} ({C.current()})  raw_station={cfg.get("raw_station")!r} '
          f'场名写法={cfg.get("src_farm_names")}')
    print(f'\n扫描到 {len(rep["stations"])} 个场站目录:')
    for nm, info in rep['stations'].items():
        mark = '← 本场' if rep['matched'] and info['dir'] == rep['matched'] else ''
        print(f'  {nm}  ({info["n_files"]} 件)  {info["dir"]}  {mark}')
        for sd, p in info['subdirs'].items():
            n = sum(1 for x in p.rglob('*') if x.is_file())
            print(f'      {sd:14s} {n:6d} 件')
    print(f'\n辨识结论 : how={rep["how"]}  →  {rep["matched"] or "(未识别)"}')
    print(f'依据     : {rep["note"]}')
    if rep['matched']:
        print('\n各约定子目录件数:')
        for n in STATION_SUBDIRS:
            d = pathlib.Path(rep['matched']) / n
            print(f'  {n:14s} {rep["counts"].get(n, 0):6d} 件   {"存在" if d.is_dir() else "缺"}')
    print('\n下一步: 放入/更新数据后跑  python scripts/rebuild_from_raw.py  (等价验收加 --verify)')
    return 0


if __name__ == '__main__':
    # 控制台 GBK 编不出 ✔ 时降级为 '?' (见 src/console.py)。
    from src import console
    console.soft()
    sys.exit(main())