guanlan-src/scripts/place_raw_data.py

#!/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 按
                    <src_10min>/<turbine>.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年油样"这一层, 直接落成 <台号>/<部件>/<pdf>,
                                       与维护页说明"按台号/部件分目录"一致)

## 两组范围 (--scope)

`--scope a2`(默认) 只落上面那四类, 即 A3 当时的范围, 语义不变。

`--scope mech` 落的是**"能重算的缺口"**(2026-09-11 用户令: 缺什么从现场包里抽)。它解决的是
"从零重算时, 某些产物因为源件不在 data/raw 而算不出来"这件事, 因此只挑**有消费者**的源:

  <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=['如海','如东']` 同源(如海=如东项目)。**包内没有消费者**: 它补的是
            数据层完整性, 不改变任何页面数值 —— 落它是因为"扫描辨识"会把它报成缺口。

## 振动侧 (--scope vib, 2026-09-12 用户令)

用户令: 「数据层里的「CMS 振动评估报告」应遵循 `<安装目录>\\data\\raw\\如东\\windcms`、
「振动线 handoff」应遵循 `<安装目录>\\data\\raw\\如东\\m5_cms_tcm` 存放; 修改系统支持振动数据参与
运行、重算; 自现场包提取相应振动数据存放至上述目录」。

  <场站>/windcms/CMS_RuDong_CGN_202603-04/measurement/<年>/<月>/<WTGxx>/*_decode.json
      ← CMS_RuDong_CGN_202603-04.zip 的 measurement/** 全部成员 (25,679 件, 解压 153.7 GB)
      依据: 这就是 `src/windcms/pipeline.py` 的 `detect_input()` 认的 **`tcm_decoded_json`**
            (Brande TCM Enterprise 导出, SiteName=CGN Rudong, LocationName=WTGxx)。原样保留
            包内 `measurement/` 这一层, 于是"哪个包来的哪一层"仍可追溯; 摄入按 rglob 找
            `*_decode.json`, 套不套这层都能吃。**这是振动侧唯一能重算的源**: 六层链全部
            输入都是它 (索引/谱 → 扫线 → 能量占比 → 模型 → 融合 → 报告)。
            ★ 2026-09-11 版的 SKIPPED 里写着"振动分析报告是成品牌报告, windcms 要的测点索引
              包里没有" —— 那句在当时成立; 2026-09-12 用户把 CMS 原始导出补进现场包后不再成立,
              故本范围落地, 并在下面改注。

  <场站>/windcms/厂家报告/上海电气_月度/     ← 如东风场数据.zip:
      `数据收集/更新/（8）风机振动数据、油液分析记录（2025.1-至今）/振动分析报告/**`
      (12 份月度用印版 PDF, 2025-06…2026-05) + 现场目录散装件
      `中广核如东海上风电场2026年07月振动分析报告_上海电气.docx`
      依据: 数据层「CMS 振动评估报告」的**厂商评估报告**侧 (VDI3834 / NB/T 31129-2018 判级)。
            12 份 PDF 是纯扫描件 (无文本层, 实测 /Font=0、pdftotext 类抽取为空) → 只作归档证据,
            数值不参与判级; docx 有文本层, 由 scripts/vib_reports_build.py 摄入。

  <场站>/m5_cms_tcm/厂家报告/                ← 现场目录散装件
      `中广核如东海上风电场传动链振动分析报告_大生科技_20260311(1).docx`
      依据: 数据层「振动线 handoff」的**TCM 侧深度分析报告** (依托机组自带 TCM M-system 数据)。
            若干现场给了 handoff 正本 (`handoff_vibration_v2.json` / `component_history.json`),
            也放本目录 → 摄入优先采用现场正本, 不再用包内 shipped 快照。

用法:
    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 vib               # 落振动侧 (CMS 原始导出 153.7 GB + 厂家报告)
    python scripts/place_raw_data.py --scope vib --limit 300   # 冒烟: 只落 300 件 (试跑/校验用)
    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）…/振动分析报告/` 落一遍 (不在 a2/mech 范围): 包内另有两处**同件副本** ——
    根目录 `中广核如东海上风电场2026年5月振动分析报告用印版(2).pdf`(与 2026年05月 件同尺寸)
    与嵌套包 `如东海上振动报告11份.zip`(其 11 份与 `振动分析报告/` 的 11 份逐件同尺寸)。
    落三遍会得到 3 份同名报告, 摄入时会互相覆盖或重复计数 —— 故只在 vib 范围落目录里那一份。
  · 不落 `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'),
]

# 振动侧 (--scope vib): 数据层两条振动行的源件 —— CMS 原始测量导出 + 厂商评估报告
ZIP_CMS = 'CMS_RuDong_CGN_202603-04.zip'
DIR_SA_REPORTS = '如东风场数据/数据收集/更新/（8）风机振动数据、油液分析记录（2025.1-至今）/振动分析报告/'
RULES_VIB = [
    # ① CMS 原始测量导出 (Brande TCM Enterprise, SiteName=CGN Rudong): 25,679 件 *_decode.json。
    #    strip=False 保留包内 measurement/ 层 → <包名>/measurement/2026/03/WTGxx/*.json, 来源可追溯。
    (ZIP_CMS, 'measurement/', f'windcms/{ZIP_CMS[:-4]}', False, 'station'),
    # ② 上海电气月度振动分析报告 (12 份扫描件 PDF): 数据层「CMS 振动评估报告」的厂商报告侧
    ('如东风场数据.zip', DIR_SA_REPORTS, 'windcms/厂家报告/上海电气_月度', True, 'station'),
]

# 现场目录下的散装件 (不是压缩包成员): (源文件名, 目标相对 data/raw/<场站>/ 的路径)
RULES_VIB_LOOSE = [
    # 上海电气 2026年07月报告: 有文本层, 由 scripts/vib_reports_build.py 摄入成评估报告
    ('中广核如东海上风电场2026年07月振动分析报告_上海电气.docx',
     'windcms/厂家报告/上海电气_月度'),
    # 大生科技传动链振动分析报告 (TCM M-system 数据): 数据层「振动线 handoff」的 TCM 侧
    ('中广核如东海上风电场传动链振动分析报告_大生科技_20260311(1).docx',
     'm5_cms_tcm/厂家报告'),
]

# 说明性的"故意不落", 只在报告里列出来
SKIPPED = [
    ('如东风场数据.zip', '如东风场数据/数据收集/更新/（3）现场检修记录（2025.1-至今）/2025年检修记录/',
     '与 工作/风机故障记录/2025年故障记录 逐件同名校验相同 (副本)'),
    ('如东风场数据.zip', '如东风场数据/数据收集/更新/（3）现场检修记录（2025.1-至今）/2026年检修记录/',
     '与 工作/风机故障记录/2026年故障记录 逐件同名校验相同 (副本)'),
    ('如东风场数据.zip', '如东风场数据/数据收集/更新/（8）风机振动数据、油液分析记录（2025.1-至今）/振动分析报告/',
     '※ 2026-09-12 起**已改为落位** (vib 范围, → windcms/厂家报告/上海电气_月度)。原判断"成品牌报告+'
     '包内没有测点索引"在当时成立; 用户补入 CMS_RuDong_CGN_202603-04.zip 后, 索引源件已具备, '
     '报告作为数据层证据一并落位。此处保留记录以免后人以为漏了'),
    ('如东风场数据.zip', '如东风场数据/中广核如东海上风电场2026年5月振动分析报告用印版(2).pdf',
     '与 振动分析报告/…2026年05月…用印版.pdf 同尺寸 (副本, 只落目录里那一份)'),
    ('如东风场数据.zip', '如东风场数据/如东海上振动报告11份.zip',
     '其 11 份与 振动分析报告/ 的 11 份逐件同尺寸 (副本包, 只落目录里那一份)'),
    ('如东风场数据.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, loose_rules=(), limit: int = 0):
    """→ [(zip, 目标文件, 包内条目名, 解压后大小, 目标根, 目标名)]; 只读压缩包目录, 不解压。

    zip=None 表示散装件 (现场目录下的单个文件, 如两份 docx 振动报告不在任何压缩包里)。
    limit>0 时只取前 limit 件 (冒烟/试跑用; 大包 2.5 万件跑一遍要几十分钟, 得能小样验证)。"""
    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))
    for fname, target in loose_rules:
        fp = src / fname
        if not fp.exists():
            raise SystemExit(f'缺散装源件: {fp}')
        out.append((None, station / target / fname, fname, fp.stat().st_size, 'station', target))
    if limit:
        out = out[:limit]
    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', 'vib', 'full'), default='a2',
                    help='a2=四项数据层(A3 语义, 默认) · mech=机理层源件+1min · '
                         'vib=振动侧(CMS 原始导出+厂家报告) · full=三组一起')
    ap.add_argument('--limit', type=int, default=0, help='只落前 N 件 (冒烟/试跑; 默认 0=全落)')
    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, loose = {
        'a2': (RULES, ()),
        'mech': (RULES_MECH, ()),
        'vib': (RULES_VIB, RULES_VIB_LOOSE),
        'full': (RULES + RULES_MECH + RULES_VIB, RULES_VIB_LOOSE),
    }[a.scope]
    items = plan(src, station, rules, loose, limit=a.limit)
    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:44s} {n:6d} 件  {human(sz):>10s}   → {where}')
    print(f'  合计 {len(items)} 件, {human(sum(i[3] for i in items))}')

    if a.scope in ('vib', 'full'):
        print('\n== 故意不落 (振动侧同件副本) ==')
        for zname, prefix, why in SKIPPED:
            if '振动' in prefix or '报告11份' in prefix:
                print(f'  {prefix}\n      理由: {why}')

    if a.dry_run:
        print('\n(dry-run, 未写盘)')
        return 0

    print('\n== 落位 ==')
    done = 0
    skipped = 0
    written = 0
    # 按包分组 + **每包只开一次**。旧写法在循环体里 with ZipFile(...) 且用 next(...) 线性找条目 ——
    # 对 2.5 万件的大包是 25,679 次开包 × 25,759 次名字比较 ≈ 6.6 亿次比较, 实测不可接受。
    # 改成: 每包一次性建 {条目名: ZipInfo} 映射, 顺序流式写出。
    by_zip = {}
    for it in items:
        by_zip.setdefault(it[0], []).append(it)
    for zname, group in by_zip.items():
        zf = zipfile.ZipFile(src / zname) if zname else None
        try:
            imap = {gbk_name(i).replace('\\', '/'): i for i in zf.infolist()} if zf else {}
            for _, dst, entry, size, root, target in group:
                # 已有同尺寸文件 = 已经是最新 → 跳过。重跑一次不该把 150 GB 原样再抄一遍
                # (vib 范围解压 153.7 GB; a2 14.7 GB / mech 15.5 GB; 2026-09-11 修单文件规则时
                #  就是靠这个避免整盘重写)。
                if dst.exists() and dst.stat().st_size == size:
                    skipped += 1
                    continue
                dst.parent.mkdir(parents=True, exist_ok=True)
                if zf is None:
                    with open(src / entry, 'rb') as fsrc, open(dst, 'wb') as fdst:
                        shutil.copyfileobj(fsrc, fdst, 1024 * 1024 * 4)
                else:
                    with zf.open(imap[entry]) 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
                written += size
                # 大包进度: 每 500 件报一次 (2.5 万件逐件打印会把日志刷爆); 单件 >200 MB 也报
                if done % 500 == 0 or size > 200 * 1024 * 1024:
                    print(f'  [{done}/{len(items)}] 已写 {human(written)}  {dst.name}', flush=True)
        finally:
            if zf is not None:
                zf.close()
    print(f'\n完成 (scope={a.scope}): 新写/更新 {done} 件 ({human(written)}), 已是最新跳过 {skipped} 件, 共 {len(items)} 件')
    return 0


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