guanlan-src/scripts/extract_site_data.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""现场收资接入：从场站交来的年度 zip（内层是每月每类 `.mdb`）提取并转成观澜 `data/raw/<场站>/`。

## 场站交来的东西长什么样（2026-09-17 实测 `F:\temp\如东风场数据`）

```
25年.zip (1,022 MB) / 26年.zip (961 MB)
  └ 26年/1月/2026-01-cnt.zip → 2026-01-cnt.mdb   (52.7 MB)
            2026-01-std.zip → 2026-01-std.mdb   ( 2.3 MB)   ← tblGrid: 10 分钟电网/机组量测
            2026-01-tur.zip → 2026-01-tur.mdb   (47.6 MB)
            2026-01-scd.zip → 2026-01-scd.mdb   ( 5.6 MB)
            … 每月 12 类：cnt din dot flg grd int prs scd std sum tmp tur
```

即 **Access 数据库按月分类**（不是 CSV）。实测 `std.mdb` 里是单表 `tblGrid`：

```
TimeStamp | Station | WPSStatus | TimestampStation | CurrentL1..L3 | VoltageL1..L3 |
ActivePower | ReActivePower | ActivePowerExport | ReActivePowerExport | …
```
`TimeStamp` 是 10 分钟节拍，`Station` 是机组号（如 91/92 —— 与 WTG 号的对应关系见 §映射）。

## 读 MDB 的路线（离线、零 pip 依赖）

Windows 上用 **.NET `System.Data.Odbc` + 系统自带的 `Microsoft Access Driver (*.mdb, *.accdb)`**
（本机实测可用；驱动来自 Office/Access Runtime，纯服务器上可能没有 ⇒ 本器会**先探针、缺了就响亮报并给补救**）。
Linux 上用 `mdbtools` 的 `mdb-export`（本机没有）。两条路都不需要往包里塞新 pip 包。

## 用法

    python scripts/extract_site_data.py --src <年度zip所在目录> --list
    python scripts/extract_site_data.py --src <…> --extract [--year 26] [--month 1] [--out data/raw/如东/scada_mdb]
    python scripts/extract_site_data.py --src <…> --probe [--class std]
    python scripts/extract_site_data.py --src <…> --to-csv --class std [--year 26 --month 1] [--turbines WTG01,WTG02]

退出码: 0 成功 · 2 来源/参数不齐 · 4 平台无 MDB 读取能力（已给补救办法） · 5 转换/校验失败
"""
from __future__ import annotations

import argparse
import hashlib
import json
import pathlib
import shutil
import subprocess
import sys
import tempfile
import zipfile

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from src import paths as P                                              # noqa: E402

WIN32 = sys.name == 'nt' if hasattr(sys, 'name') else False
WIN32 = sys.platform.startswith('win')
DRIVER = 'Microsoft Access Driver (*.mdb, *.accdb)'
# 类目 → 观澜侧的落点（先按已知的上线：只有 std 与 sum 是 10min 量测；其余待现场口径确认）
CLASS_TO_RAW = {
    'std': 'scada_10min',      # 实测: tblGrid = 10 分钟电网/机组量测（TimeStamp/Station/ActivePower/…）
    'sum': 'scada_10min',      # 待确认（体量最大, 疑为统计汇总）
    'cnt': 'scada_10min',      # 待确认
    'tur': 'scada_10min',      # 待确认（47 MB, 疑为机组状态）
}


# 现场交来的**已导出 CSV** zip → 观澜 raw 目录（2026-09-17 实测：与 data/raw 现有 38 件逐名对应）
ZIP_TO_RAW = {
    '10分钟数据': 'scada_10min',     # WTG01.csv … WTG38.csv（38 件 / 14.4 GB）
    '1分钟数据': 'scada_1min',       # 01E.csv … 38B.csv（38 件 / 12.8 GB，机组代号命名）
}


def verify_raw(src: pathlib.Path, farm: str | None = None) -> int:
    """对账：现场 zip 里的 CSV ↔ `data/raw/<场>/<类>/` 现有文件（逐件比大小）。

    这一步回答的是"**现有的输入数据到底是不是从这两个 zip 来的**" —— 不去解压十几 GB 就能验。
    """
    farm = farm or P.farm()
    zips = [z for z in find_year_zips(src) if z.stem in ZIP_TO_RAW]
    if not zips:
        print(f'[X] {src} 下没有 {"/".join(ZIP_TO_RAW)}.zip')
        return 2
    print(f'== 现场 zip ↔ data/raw/{farm}/ 对账 ==')
    bad = 0
    for z in zips:
        d = station_dir(farm) / ZIP_TO_RAW[z.stem]
        with zipfile.ZipFile(z) as zf:
            ents = [e for e in zf.infolist() if not e.is_dir()]
        same = miss = diff = 0
        miss_names, diff_names = [], []
        for e in ents:
            tgt = d / pathlib.Path(e.filename).name
            if not tgt.is_file():
                miss += 1
                miss_names.append(tgt.name)
            elif tgt.stat().st_size == e.file_size:
                same += 1
            else:
                diff += 1
                diff_names.append(f'{tgt.name}(zip {e.file_size:,} vs 盘 {tgt.stat().st_size:,})')
        bad += miss + diff
        print(f'   {z.name} → {P.rel(d)}: zip {len(ents)} 件 · 逐字节同大小 {same} · 缺 {miss} · 大小不同 {diff}')
        for n in miss_names[:4]:
            print(f'      缺: {n}')
        for n in diff_names[:4]:
            print(f'      异: {n}')
    print(f'   结论: {"现有 raw 输入与现场 zip 逐件一致（可追溯）" if bad == 0 else str(bad) + " 件需要重新提取"}  '
          f'（缺件提取: --extract-csv --only-missing）')
    return 0 if bad == 0 else 5


def extract_csv(src: pathlib.Path, farm: str | None = None, only_missing: bool = False) -> int:
    """把"已导出 CSV"的现场 zip 解到 `data/raw/<场>/<类>/`（幂等：同名同大小即跳过）。"""
    farm = farm or P.farm()
    zips = [z for z in find_year_zips(src) if z.stem in ZIP_TO_RAW]
    if not zips:
        print(f'[X] {src} 下没有 {"/".join(ZIP_TO_RAW)}.zip')
        return 2
    done = skipped = 0
    for z in zips:
        d = station_dir(farm) / ZIP_TO_RAW[z.stem]
        d.mkdir(parents=True, exist_ok=True)
        print(f'   {z.name} → {P.rel(d)}')
        with zipfile.ZipFile(z) as zf:
            for e in zf.infolist():
                if e.is_dir():
                    continue
                tgt = d / pathlib.Path(e.filename).name
                if tgt.is_file() and tgt.stat().st_size == e.file_size:
                    skipped += 1
                    continue
                if only_missing and tgt.is_file():
                    skipped += 1
                    continue
                print(f'      + {tgt.name}  {e.file_size / 1e6:.1f} MB')
                with zf.open(e) as fh, open(tgt, 'wb') as out:
                    shutil.copyfileobj(fh, out, 1024 * 1024 * 8)
                done += 1
    print(f'   完成: 新解 {done} 件 · 跳过（已同大小）{skipped} 件')
    print('   接下去: python scripts/raw_data_check.py（体检）→ python scripts/rebuild_all.py（重算）')
    return 0


def station_dir(farm: str | None = None) -> pathlib.Path:
    """`data/raw/<场站>/` —— ★场站**目录名**可与 farm 键不同（键 = rudong，目录 = 如东），
    所以必须走唯一取用口 `raw_station_dir()`；第一版按 `P.RAW_ROOT / farm` 拼，于是"38 件全缺"
    （对着 data/raw/rudong 找, 而东西在 data/raw/如东）。"""
    from src.windscada.config import raw_station_dir
    return pathlib.Path(raw_station_dir(farm or P.farm()))


def find_year_zips(src: pathlib.Path) -> list[pathlib.Path]:

    return sorted([p for p in src.glob('*.zip') if p.is_file()])


def list_structure(src: pathlib.Path) -> int:
    zips = find_year_zips(src)
    if not zips:
        print(f'[X] {src} 下没有年度 zip')
        return 2
    print(f'== 现场收资结构 · {src} ==')
    for z in zips:
        with zipfile.ZipFile(z) as zf:
            names = zf.namelist()
        months = sorted({n.split('/')[1] for n in names if n.count('/') >= 1 and n.split('/')[1]})
        classes = sorted({n.split('/')[-1].split('-')[-1].replace('.zip', '')
                          for n in names if n.lower().endswith('.zip')})
        print(f'   {z.name}  {z.stat().st_size / 1e6:.0f} MB · {len(names)} 条目')
        print(f'      月份 {len(months)} 个: {", ".join(months)}')
        print(f'      类目 {len(classes)} 个: {", ".join(classes)}')
    return 0


def extract(src: pathlib.Path, out: pathlib.Path, year: str | None, month: str | None) -> int:
    zips = [z for z in find_year_zips(src) if (not year or z.stem.startswith(year))]
    if not zips:
        print('[X] 没有匹配的年度 zip')
        return 2
    out.mkdir(parents=True, exist_ok=True)
    man_p = out / '_manifest.json'
    man = json.loads(man_p.read_text(encoding='utf-8')) if man_p.is_file() else dict(zips=[], files=[])
    done = 0
    for z in zips:
        with zipfile.ZipFile(z) as zf:
            inner = [n for n in zf.namelist() if n.lower().endswith('.zip')
                     and (not month or f'/{month}月/' in n)]
            print(f'   {z.name}: 内层 {len(inner)} 个分类 zip' + (f'（限 {month} 月）' if month else ''))
            for n in inner:
                parts = n.split('/')                       # 26年/1月/2026-01-std.zip
                rel = pathlib.Path(*parts[1:]) if len(parts) > 1 else pathlib.Path(parts[0])
                dst_dir = out / rel.parent
                dst_dir.mkdir(parents=True, exist_ok=True)
                with zf.open(n) as fh, zipfile.ZipFile(fh) as inner_zf:
                    for m in inner_zf.namelist():
                        tgt = dst_dir / pathlib.Path(m).name
                        if tgt.is_file() and tgt.stat().st_size == inner_zf.getinfo(m).file_size:
                            continue
                        tgt.write_bytes(inner_zf.read(m))
                        h = hashlib.sha256(tgt.read_bytes()).hexdigest()[:16]
                        man['files'] = [x for x in man['files'] if x['path'] != str(tgt.relative_to(out))]
                        man['files'].append(dict(path=str(tgt.relative_to(out)), bytes=tgt.stat().st_size,
                                                 sha256_16=h, from_zip=f'{z.name}:{n}'))
                        done += 1
                        print(f'      + {tgt.relative_to(out)}  {tgt.stat().st_size / 1e6:.1f} MB  sha256:{h}')
    man['zips'] = sorted({*man.get('zips', []), *[z.name for z in zips]})
    man_p.write_text(json.dumps(man, ensure_ascii=False, indent=1) + '\n', encoding='utf-8')
    print(f'   完成: 新解 {done} 个 MDB → {P.rel(out)}（清单 {P.rel(man_p)}）')
    return 0


def have_mdb_route() -> tuple[bool, str]:
    if WIN32:
        ps = ('$d=(Get-OdbcDriver | Where-Object { $_.Name -like "*Access Driver (*.mdb*" }).Count;'
              'if($d -gt 0){"ok"}else{"no"}')
        try:
            r = subprocess.run(['powershell', '-NoProfile', '-Command', ps], capture_output=True, text=True,
                               errors='replace', timeout=120)
            ok = 'ok' in (r.stdout or '')
            return ok, 'Windows + .NET System.Data.Odbc + 系统 Access ODBC 驱动' if ok else '缺 Access ODBC 驱动'
        except Exception as e:
            return False, f'探测失败: {type(e).__name__}'
    ok = bool(shutil.which('mdb-export') or shutil.which('mdb-tables'))
    return ok, 'mdbtools (mdb-export)' if ok else '缺 mdbtools'


def mdb_tables(mdb: pathlib.Path) -> list[str]:
    if WIN32:
        ps = (f'$cs="Driver={{{DRIVER}}};Dbq={mdb};";'
              '$cn=New-Object System.Data.Odbc.OdbcConnection($cs);$cn.Open();'
              '$cn.GetSchema("Tables") | Where-Object { $_.TABLE_TYPE -eq "TABLE" } | '
              'ForEach-Object { $_.TABLE_NAME };$cn.Close()')
        r = subprocess.run(['powershell', '-NoProfile', '-Command', ps], capture_output=True, text=True,
                           errors='replace', timeout=300)
        if r.returncode != 0:
            raise RuntimeError((r.stderr or r.stdout or '').strip()[:300])
        return [x.strip() for x in (r.stdout or '').splitlines() if x.strip()]
    r = subprocess.run(['mdb-tables', '-1', str(mdb)], capture_output=True, text=True, errors='replace', timeout=300)
    return [x.strip() for x in (r.stdout or '').splitlines() if x.strip()]


def mdb_to_csv(mdb: pathlib.Path, table: str, out_csv: pathlib.Path, where: str = '') -> int:
    """MDB 单表 → CSV（UTF-8）。Windows 走 PowerShell/.NET Odbc；Linux 走 mdb-export。"""
    out_csv.parent.mkdir(parents=True, exist_ok=True)
    if WIN32:
        q = f'SELECT * FROM [{table}]' + (f' WHERE {where}' if where else '')
        ps = f'''$cs="Driver={{{DRIVER}}};Dbq={mdb};"
$cn=New-Object System.Data.Odbc.OdbcConnection($cs);$cn.Open()
$cmd=$cn.CreateCommand();$cmd.CommandText="{q}"
$da=New-Object System.Data.Odbc.OdbcDataAdapter($cmd)
$dt=New-Object System.Data.DataTable;[void]$da.Fill($dt)
$dt | Export-Csv -Path "{out_csv}" -NoTypeInformation -Encoding UTF8
"rows=$($dt.Rows.Count)"
$cn.Close()'''
        tf = pathlib.Path(tempfile.gettempdir()) / '_mdb_dump.ps1'
        tf.write_text(ps, encoding='utf-8-sig')
        r = subprocess.run(['powershell', '-NoProfile', '-ExecutionPolicy', 'Bypass', '-File', str(tf)],
                           capture_output=True, text=True, errors='replace', timeout=3600)
        print('      ' + (r.stdout or '').strip().replace('\n', ' ')[:80])
        if r.returncode != 0:
            print('      [X] ' + (r.stderr or r.stdout or '').strip()[:200])
            return 5
        return 0
    cmd = ['mdb-export', str(mdb), table] + (['-q', where] if where else [])
    with open(out_csv, 'w', encoding='utf-8', newline='') as fh:
        r = subprocess.run(cmd, stdout=fh, stderr=subprocess.PIPE, text=True, errors='replace', timeout=3600)
    if r.returncode != 0:
        print('      [X] ' + (r.stderr or '')[:200])
        return 5
    return 0


def probe(out: pathlib.Path, cls: str | None) -> int:
    ok, route = have_mdb_route()
    print(f'== MDB 读取能力 ==\n   {route}: {"可用" if ok else "不可用"}')
    if not ok:
        print('   补救: Windows 装 Access Runtime（或 Office）后自带 Access ODBC 驱动; '
              'Linux 装 mdbtools（apt/yum install mdbtools）; 或让现场把 MDB 导出成 CSV。')
        return 4
    mdbs = sorted(out.rglob('*.mdb'))
    mdbs = [m for m in mdbs if (not cls or f'-{cls}.' in m.name)]
    if not mdbs:
        print(f'   [i] {P.rel(out)} 下还没有 MDB —— 先跑 --extract')
        return 0
    print(f'   已解出 MDB {len(mdbs)} 个（限类目 {cls or "全部"}）')
    for m in mdbs[:8]:
        try:
            t = mdb_tables(m)
            print(f'   {m.relative_to(out)}: 表 {t}')
        except Exception as e:
            print(f'   {m.relative_to(out)}: [X] {type(e).__name__}: {str(e)[:120]}')
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description='现场收资（年度 zip → 每月每类 MDB）提取与接入')
    ap.add_argument('--src', default=str(pathlib.Path('F:/temp/如东风场数据')),
                    help='年度 zip 所在目录（默认按现场交来的目录名）')
    ap.add_argument('--out', default=str(station_dir() / 'scada_mdb'),
                    help='MDB 落点（默认 data/raw/<场站>/scada_mdb）')
    ap.add_argument('--list', action='store_true', help='只列结构（年度 zip / 月份 / 类目）')
    ap.add_argument('--extract', action='store_true', help='解压出 MDB（幂等, 带清单）')
    ap.add_argument('--probe', action='store_true', help='探针: 读取能力 + 已解出 MDB 的表')
    ap.add_argument('--verify-raw', action='store_true', help='对账: 现场 CSV zip ↔ data/raw 现有文件(逐件比大小)')
    ap.add_argument('--extract-csv', action='store_true', help='把现场 CSV zip 解到 data/raw（幂等）')
    ap.add_argument('--only-missing', action='store_true', help='与 --extract-csv 合用: 只补缺件, 不覆盖已存在的')
    ap.add_argument('--to-csv', action='store_true', help='MDB → 观澜 data/raw 的 CSV（按类目映射）')
    ap.add_argument('--class', dest='cls', default=None, help='类目: cnt/sum/std/tur/…')
    ap.add_argument('--year', default=None, help='年度前缀（如 26）')
    ap.add_argument('--month', default=None, help='月份数字（如 1）')
    ap.add_argument('--table', default=None, help='指定表名（默认取该 MDB 的第一张表）')
    ap.add_argument('--turbines', default=None, help='只导这些机组（逗号分隔, 如 WTG01,WTG02）')
    a = ap.parse_args()
    src, out = pathlib.Path(a.src), pathlib.Path(a.out)
    if a.list:
        return list_structure(src)
    if a.extract:
        return extract(src, out, a.year, a.month)
    if a.verify_raw:
        return verify_raw(src)
    if a.extract_csv:
        return extract_csv(src, only_missing=a.only_missing)
    if a.probe:
        return probe(out, a.cls)
    if a.to_csv:
        return to_csv(out, a.cls, a.year, a.month, a.table, a.turbines)
    ap.print_help()
    return 2


def to_csv(out: pathlib.Path, cls: str | None, year: str | None, month: str | None,
           table: str | None, turbines: str | None) -> int:
    ok, route = have_mdb_route()
    if not ok:
        print(f'[X] 本机没有 MDB 读取能力（{route}）—— 见 --probe 的补救办法')
        return 4
    if not cls:
        print('[X] --to-csv 需要 --class（如 std）')
        return 2
    mdbs = [m for m in sorted(out.rglob(f'*-{cls}.mdb')) if (not year or f'{year}年' in str(m))
            and (not month or f'{month}月' in str(m))]
    if not mdbs:
        print(f'[X] 没找到 *-{cls}.mdb（先 --extract）')
        return 2
    raw_dir = station_dir() / CLASS_TO_RAW.get(cls, f'scada_{cls}')
    print(f'== MDB → CSV · 类目 {cls} · {len(mdbs)} 个月 · 落点 {P.rel(raw_dir)} ==')
    want = {t.strip().upper() for t in turbines.split(',')} if turbines else None
    for m in mdbs:
        try:
            tbls = mdb_tables(m)
        except Exception as e:
            print(f'   {m.name}: [X] {type(e).__name__}: {str(e)[:120]}')
            continue
        tbl = table or (tbls[0] if tbls else None)
        if not tbl:
            print(f'   {m.name}: 没有表')
            continue
        tmp_csv = pathlib.Path(tempfile.gettempdir()) / f'_{m.stem}_{tbl}.csv'
        rc = mdb_to_csv(m, tbl, tmp_csv)
        if rc:
            continue
        import pandas as pd
        df = pd.read_csv(tmp_csv, dtype=str)
        print(f'   {m.name} / {tbl}: {len(df)} 行 × {len(df.columns)} 列 → {list(df.columns)[:8]}')
        # 现场口径: Station = 机组号(91→WTG01), 每个机组一个文件、时间列名统一成 Time
        st = next((c for c in df.columns if c.lower() in ('station', 'stationid', 'wtg')), None)
        ts = next((c for c in df.columns if 'time' in c.lower() and 'station' not in c.lower()), None)
        if not st or not ts:
            print(f'      [i] 没识别出 Station/TimeStamp 列 —— 该月份先原样留存 {P.rel(tmp_csv)}')
            continue
        base = df[st].astype(str).str.extract(r'(\d+)')[0].astype('Int64')
        df['_wtg'] = 'WTG' + (base - (90 if base.min() and base.min() >= 90 else 0)).astype('Int64').astype(str).str.zfill(2)
        for w, g in df.groupby('_wtg'):
            if want and w.upper() not in want:
                continue
            g2 = g.drop(columns=['_wtg']).rename(columns={ts: 'Time'})
            dst = raw_dir / f'{w}.csv'
            head = not dst.is_file()
            g2.to_csv(dst, mode='a' if not head else 'w', header=head, index=False, encoding='utf-8')
            if head:
                print(f'      → {P.rel(dst)}  {len(g2)} 行（Time 列 = {ts}）')
    print('   完成。接下去: python scripts/raw_data_check.py（体检）→ python scripts/rebuild_all.py')
    return 0


if __name__ == '__main__':
    for _s in (sys.stdout, sys.stderr):
        try:
            _s.reconfigure(errors='replace')
        except Exception:
            pass
    sys.exit(main())