#!/usr/bin/env python3 # -*- coding: utf-8 -*- """把现场给的数据包按 A2 约定落位到 data/raw/<场站名称>/ 下 (2026-09-11)。 ## 为什么要有这个脚本 A2 定了四项数据源的落位: data/raw/<场站名称>/{scada_10min, 故障报警, 风机故障记录, 油样报告}。 现场拿到的却是几个 GBK 名的大压缩包 (10分钟数据.zip / 如东风场数据.zip), 内部目录名与 落位目录名并不一一对应 (例如报警在 `报警数据/`, 工单在 `工作/风机故障记录/`, 油样埋在 `数据收集/更新/(8)…/2025年油样/`)。手工拖拽容易拖错一层、也说不清依据什么。 本脚本把映射**写成表**, 于是"哪个包里的哪个目录去了哪"是可复核的, 重复跑也不会走样。 ## 落位映射 (target 相对 data/raw/<场站名称>/) scada_10min/ ← 10分钟数据.zip 全部 38 个 WTG*.csv (平铺; src/windscada/data.py 按 /.csv 读, 所以必须是文件本身不是再套一层目录) 故障报警/ ← 如东风场数据.zip: 报警数据/** (年度/季度 .xls, XML 报警导出) 如东风场数据.zip: 数据收集/更新/(2)故障记录(首发故障有标识)(2025.1-至今)/故障记录/** (2025年全年故障记录.xls / 2026年至今.xls — 正是页面写的 *年至今.xls) 风机故障记录/ ← 如东风场数据.zip: 工作/风机故障记录/** (2021…2026 年故障记录/ + 业主统计故障/ + 年度 .rar) 如东风场数据.zip: 数据收集/更新/(3)现场检修记录(2025.1-至今)/大部件维修记录.*.xlsx 油样报告/ ← 如东风场数据.zip: 数据收集/更新/(8)风机振动数据、油液分析记录(2025.1-至今)/2025年油样/** (去掉"2025年油样"这一层, 直接落成 <台号>/<部件>/, 与维护页说明"按台号/部件分目录"一致) ## 两组范围 (--scope) `--scope a2`(默认) 只落上面那四类, 即 A3 当时的范围, 语义不变。 `--scope mech` 落的是**"能重算的缺口"**(2026-09-11 用户令: 缺什么从现场包里抽)。它解决的是 "从零重算时, 某些产物因为源件不在 data/raw 而算不出来"这件事, 因此只挑**有消费者**的源: /西门子4.0技术资料/ ← 如东风场数据.zip: 西门子4.0技术资料/** (317 件 2.8 GB) + 如东风场数据/广核如海风电场西门子风机故障代码中英文对译表.xlsx 依据: `src/ontology/kb_ingest.py` 里 `TECH = P.RAW_ROOT/'西门子4.0技术资料'`, 且 `rd()` 要读 `故障处理/故障处理手册.xlsx`、`广核如海风电场西门子风机故障代码中英文对译表.xlsx`、 `如海故障代码表.xlsx`、`维护相关/维护作业指导书.xlsx` 四个文件。前三个里**对译表在 技术资料目录里没有**(包里它在 `scada数据(如东)/` 与包根各有一份), 所以单独补一条规则。 效果: 本体层 `objects.json` 从"包内没有源件(⛔)"变成**可重算**。 <场站>/scada_1min/ ← 1分钟数据.zip 全部 38 个 <台号>.csv (01E.csv…37B.csv) 依据: `src/windscada/config.py` 的 STATION_SUBDIRS 声明了这个子目录, `scan_stations.py` 报它"缺"; 表头带 `source_file=如海风机1分钟数据/如海测点_2025-01.csv`, 与场配置 `src_farm_names=['如海','如东']` 同源(如海=如东项目)。**包内没有消费者**: 它补的是 数据层完整性, 不改变任何页面数值 —— 落它是因为"扫描辨识"会把它报成缺口。 用法: python scripts/place_raw_data.py --dry-run # 只报要落什么, 不写盘 (默认 a2) python scripts/place_raw_data.py # 真落位 (已存在则覆盖) python scripts/place_raw_data.py --scope mech --dry-run # 看机理层/1min 会落什么 python scripts/place_raw_data.py --scope full # 两组一起 python scripts/place_raw_data.py --src D:\\别的现场数据目录 ## 故意不做的事 (两组范围共有的判断) · 不落 `(3)现场检修记录/2025年检修记录` 与 `2026年检修记录`: 与 工作/风机故障记录/2025年故障记录、 2026年故障记录 **逐件同名同大小**(19/19 件), 是同一批月度汇总表的副本。落两遍会让"按年目录" 摄入看到两份重复台账 (2026-08-31 长停台账虚高 35 倍那类事故的同款成因: 快照重复必须归并, 不能叠加)。 · 不落 `(8)…/振动分析报告/`(12 份月度用印版 PDF): 是振动线 (CMS) 的**成品牌报告**, 不是可再加工的 测量数据; 而 `windcms`/`m5_cms_tcm` 要的是 CMS 测点索引(handoff), 包里没有。 · 不落 `scada数据(如东)/**`(19 个月 × 12 个通道组 zip, 2.4 GB): 它是 `scada_10min/*.csv` 的**上游** 原始通道导出, 包内没有任何脚本读它(构建器读的是已经平铺好的 10min CSV) —— 落了也不参与重算。 · 不落 `fastlog数据/`(4 件 WTG0x.xls): 全库搜 `fastlog` 只有 2 处**注释**提到它("运行态见证"), 没有读取代码。 · 结论: 上面三项都是"有源件、无生成端/无消费者"。`genbearing_monthly`、`mblub_monthly`、 `yaw_dynamic_monthly`、`yaw1min_liveness`、`sector_power`、`duty_monthly`、`pc_monthly_bins`、 `thermal_monthly`、`structure.parquet`、`watch_channels_monthly` 这批产物同理 —— 全库只有读取方、**0 处写入方**, 属于 v0.2.0 未附构建脚本(见 docs §4)。 """ from __future__ import annotations import argparse import pathlib import os import shutil import sys import zipfile ROOT = pathlib.Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) # 现场数据包所在目录: 用 --src 指定, 或 env GUANLAN_PLACE_SRC; 默认取安装目录下 data/_incoming DEFAULT_SRC = pathlib.Path(os.environ.get('GUANLAN_PLACE_SRC') or (ROOT / 'data' / '_incoming')) ZIP_10MIN = '10分钟数据.zip' ZIP_FARM = '如东风场数据.zip' # (压缩包, 包内前缀, 目标, 是否去掉前缀这一层, 目标根: station= data/raw/<场站>/, raw= data/raw/) RULES = [ ('10分钟数据.zip', '', 'scada_10min', True, 'station'), ('如东风场数据.zip', '如东风场数据/报警数据/', '故障报警', True, 'station'), ('如东风场数据.zip', '如东风场数据/数据收集/更新/(2)故障记录(首发故障有标识)(2025.1-至今)/故障记录/', '故障报警', True, 'station'), ('如东风场数据.zip', '如东风场数据/工作/风机故障记录/', '风机故障记录', True, 'station'), ('如东风场数据.zip', '如东风场数据/数据收集/更新/(3)现场检修记录(2025.1-至今)/大部件维修记录.20240619143912557.xlsx', '风机故障记录', True, 'station'), ('如东风场数据.zip', '如东风场数据/数据收集/更新/(8)风机振动数据、油液分析记录(2025.1-至今)/2025年油样/', '油样报告', True, 'station'), ] # 机理层与可选层 (--scope mech): 只挑**有消费者**的源 —— 目的是让"从零重算"能覆盖到本体层。 RULES_MECH = [ # 本体层源件: kb_ingest.py 的 TECH = data/raw/西门子4.0技术资料 (317 件, 含故障处理/维护相关/各类图纸) ('如东风场数据.zip', '如东风场数据/西门子4.0技术资料/', '西门子4.0技术资料', True, 'raw'), # …但它要读的对译表不在技术资料目录里, 包里在别处 → 单独补一条 (见本文件开头"两组范围") ('如东风场数据.zip', '如东风场数据/广核如海风电场西门子风机故障代码中英文对译表.xlsx', '西门子4.0技术资料', True, 'raw'), # 数据层声明里的 scada_1min (config.STATION_SUBDIRS); 包内无消费者, 补的是数据层完整性 ('1分钟数据.zip', '', 'scada_1min', True, 'station'), ] # 说明性的"故意不落", 只在报告里列出来 SKIPPED = [ ('如东风场数据.zip', '如东风场数据/数据收集/更新/(3)现场检修记录(2025.1-至今)/2025年检修记录/', '与 工作/风机故障记录/2025年故障记录 逐件同名校验相同 (副本)'), ('如东风场数据.zip', '如东风场数据/数据收集/更新/(3)现场检修记录(2025.1-至今)/2026年检修记录/', '与 工作/风机故障记录/2026年故障记录 逐件同名校验相同 (副本)'), ('如东风场数据.zip', '如东风场数据/数据收集/更新/(8)风机振动数据、油液分析记录(2025.1-至今)/振动分析报告/', '振动线成品牌报告, 不是可再加工的测量数据; windcms 要的测点索引包里没有'), ('如东风场数据.zip', '如东风场数据/scada数据(如东)/', 'scada_10min/*.csv 的上游原始通道导出 (2.4 GB), 包内无脚本读它 → 不参与重算'), ('如东风场数据.zip', '如东风场数据/fastlog数据/', '全库搜 fastlog 只有 2 处注释提到, 无读取代码'), ] def gbk_name(info: zipfile.ZipInfo) -> str: """zip 条目名 → 真名。现场包用 GBK 存中文名, zipfile 按 cp437 解出来是乱码。""" raw = info.filename try: return raw.encode('cp437').decode('gbk') except Exception: try: return raw.encode('cp437').decode('utf-8') except Exception: return raw def human(n: float) -> str: for unit in ('B', 'KB', 'MB', 'GB'): if n < 1024 or unit == 'GB': return f'{n:.1f} {unit}' n /= 1024 return f'{n:.1f} GB' def plan(src: pathlib.Path, station: pathlib.Path, rules): """→ [(zip, 目标文件, 包内条目名, 解压后大小, 目标根, 目标名)]; 只读压缩包目录, 不解压。""" out = [] for zname, prefix, target, strip, root in rules: zp = src / zname if not zp.exists(): raise SystemExit(f'缺压缩包: {zp}') base = station if root == 'station' else station.parent # 'raw' = data/raw (机理层不在场站目录下) with zipfile.ZipFile(zp) as zf: for info in zf.infolist(): if info.is_dir(): continue name = gbk_name(info).replace('\\', '/') if not name.startswith(prefix): continue rest = name[len(prefix):] if strip else name if not rest: # 前缀恰是条目本身 = 单文件规则 (如 '…/大部件维修记录.20240619143912557.xlsx')。 # 旧写法在这里直接 continue, 于是**单文件规则从来没落过东西** —— 2026-09-11 逮到两例: # `大部件维修记录.20240619143912557.xlsx`(A2 规则, 从 A3 起就没落) 与 # `广核如海…对译表.xlsx`(mech 规则)。只有"目录条目"(prefix 以 / 结尾) 才该跳过。 if prefix.endswith('/'): continue rest = prefix.rstrip('/').rsplit('/', 1)[-1] out.append((zname, base / target / rest, name, info.file_size, root, target)) return out def main() -> int: ap = argparse.ArgumentParser() ap.add_argument('--src', default=str(DEFAULT_SRC), help='现场数据目录 (默认 %(default)s)') ap.add_argument('--dry-run', action='store_true', help='只报计划, 不写盘') ap.add_argument('--scope', choices=('a2', 'mech', 'full'), default='a2', help='a2=只落四项数据层(A3 语义, 默认) · mech=机理层源件+1min · full=两组一起') a = ap.parse_args() src = pathlib.Path(a.src) if not src.is_dir(): raise SystemExit(f'现场数据目录不存在: {src}') from src.windscada.config import farm, raw_station_dir station = pathlib.Path(raw_station_dir()) cfg = farm() print(f'场站目录: {station} (来自 src/windscada/config.py raw_station={cfg.get("raw_station")!r})') print(f'现场数据: {src} 范围: --scope {a.scope}\n') rules = {'a2': RULES, 'mech': RULES_MECH, 'full': RULES + RULES_MECH}[a.scope] items = plan(src, station, rules) by_target = {} for zname, dst, entry, size, root, target in items: d = by_target.setdefault((root, target), [0, 0]) d[0] += 1 d[1] += size print(f'== 计划落位 (scope={a.scope}) ==') for (root, target), (n, sz) in sorted(by_target.items(), key=lambda kv: kv[0][1]): where = (station / target) if root == 'station' else (station.parent / target) print(f' {target:18s} {n:5d} 件 {human(sz):>10s} → {where}') print(f' 合计 {len(items)} 件, {human(sum(i[3] for i in items))}') print('\n== 故意不落 ==') for zname, prefix, why in SKIPPED: print(f' {prefix}\n 理由: {why}') if a.dry_run: print('\n(dry-run, 未写盘)') return 0 print('\n== 落位 ==') done = 0 skipped = 0 for zname, dst, entry, size, root, target in items: # 已有同尺寸文件 = 已经是最新 → 跳过。重跑一次不该把 15 GB 原样再抄一遍 # (mech 范围 15.5 GB, a2 范围 14.7 GB; 2026-09-11 修单文件规则时就是靠这个避免整盘重写)。 if dst.exists() and dst.stat().st_size == size: skipped += 1 continue dst.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(src / zname) as zf: info = next(i for i in zf.infolist() if gbk_name(i).replace('\\', '/') == entry) with zf.open(info) as fsrc, open(dst, 'wb') as fdst: shutil.copyfileobj(fsrc, fdst, 1024 * 1024 * 4) got = dst.stat().st_size if got != size: raise SystemExit(f'写出大小不符: {dst} 期望 {size} 实得 {got}') done += 1 if done % 10 == 0 or size > 50 * 1024 * 1024: print(f' [{done}/{len(items)}] {human(size):>10s} {dst.relative_to(station.parent)}', flush=True) print(f'\n完成 (scope={a.scope}): 新写/更新 {done} 件, 已是最新跳过 {skipped} 件, 共 {len(items)} 件') return 0 if __name__ == '__main__': sys.exit(main())