scada 接入兼容 MDB: 统一取数层 + 老库错位如实拒绝 + 验收逐值一致

用户令 2026-09-19「scada 数据接入兼容支持 CSV、MDB」。
此前 src/windscada/data.py::load_10min() 把路径写死成 <src_10min>/<台>.csv ⇒ 现场交付的年度归档库
(scada_mdb/<年>年/<月>月/*.mdb) 只能看不能算。

新增:
- src/windscada/scada_source.py: 统一取数层。CSV 在位就用 CSV(既有 10 个构建器的形态), 该台没有 CSV
  才回落 MDB; 用的哪种记在返回表 attrs['scada_source'] 并首次打印一行 —— 不静默换源; 两形态都取不到
  则响亮报错(不返回空表)。读 MDB 走 ACE + PowerShell, 逐分片表 SELECT ... WHERE turbine=? 再按
  (rid, turbine) 拼列。
- src/windscada/mdb_names.py: 列名规则单一实现(Access 不能含 . ! [ ], 建库要消毒; 读侧按同一规则把
  契约列名翻成库内列名、取回再改回, 否则从 MDB 取数会静默少列)。
- data.py: load_10min 改走取数层 + source_report() 形态盘点。

实测逮到并处置的四个坑:
1) 老库没有机组列(1min 源件更没有) ⇒ 1min 老库按台取不到数; 10min 老库用源件自带的 WTG 列兜底。
2) 老库分片表表头与数据行差一列(旧 build_chunks 表头 names[lo:hi+1] vs 行 vals[lo:hi]) ⇒ 第 250 列
   之后列名整体错位、相邻分片共用列名。读侧**响亮拒绝**并给出重建命令(拿错位库算数比缺数据更危险)。
3) csv_to_mdb 建库改为按**列名对齐**(1min 38 件实测 16 种表头, 位置堆法会静默串列) + 写 turbine 列。
4) rid 在库里是文本 ⇒ 合并后行序变字典序(1,10,100…), 与 CSV 对拍"对不上"其实数据没错; 现按数值排。

验收(可重复): csv_to_mdb --month 2026-07 --class 10min 重建后, 同台同月 CSV 取 vs MDB 取
**866 行 × 4 列(含 t2/t3 跨分片列)逐值一致**; 2025-03 老库被如实拒绝。
docs/现场收资接入_v0.1.md 新增 §8(形态/规则/四个坑/验收/重建进度)。

后台: 修好的脚本正重跑全部月份(10min 19 + 1min 18, 约 5 h)使归档库成为可摄入形态; 1min 包内暂无消费者。

docs/现场收资接入_v0.1.md

@@ -137,4 +137,48 @@ python scripts/rebuild_all.py               # 或门户「数据重算」
    按月过滤 → 按 250 列切片写临时分片 CSV（首列 `rid`）→ 建库 → 按列生成 DDL（`CREATE TABLE`）→
    按 §7.3 的链接语法 `INSERT … SELECT` → 打成同名 zip → **读回校验行数/首末行与源 CSV 一致**；
 2. 试点：`--month 2025-01 --turbines WTG01`（1 台 1 月）跑通并逐值对拍；
-3. 全量：38 台 × 21 个月（10min 3 张表/月、1min 1 张表/月），按 §7.2 按月分库。
+3. 全量：38 台 × 21 个月（10min 3 张表/月、1min 1 张表/月），按 §7.2 按月分库。
+
+---
+
+## 8. SCADA 接入兼容 CSV 与 MDB（用户令 2026-09-19）
+
+用户令: **「scada 数据接入兼容支持 CSV、MDB」**。此前 `src/windscada/data.py::load_10min()` 把路径写死成
+`<src_10min>/<台>.csv` ⇒ 只有"已导出 CSV"形态能被摄入, 现场交付的年度归档库只能看不能算。
+现在取数统一走 `src/windscada/scada_source.py`：
+
+| 形态 | 位置 | 取数方式 |
+|---|---|---|
+| CSV（优先） | `data/raw/<场站>/scada_10min/WTG01.csv…` | `pandas.read_csv(engine='pyarrow')`，按契约列取交集（既有行为不变） |
+| MDB（回落） | `data/raw/<场站>/scada_mdb/<年>年/<月>月/<年>-<月>-<类>.mdb` | ACE(`Microsoft.ACE.OLEDB.12.0`) + PowerShell，逐分片表 `SELECT … WHERE turbine=?`，按 `(rid, turbine)` 拼列 |
+
+**规则**: CSV 在位就用 CSV；该台没有 CSV 才回落 MDB；用的哪种记在返回表的 `attrs['scada_source']`
+并在首次使用时打一行 —— 不静默换源。两种形态都取不到 ⇒ **响亮报错**（不返回空表，空表会被下游当成"这台没数据"）。
+
+### 8.1 库这一侧的四个坑（2026-09-19 实测逮到）
+
+1. **老库没有机组列**：2026-09-19 之前建的库每行只有 `rid`，而 1min 源件**没有任何机组标识列**
+   （`source_file` 全行同值）⇒ 那份 1min 库**按台取不到数**。10min 老库还能救：源件自带 `WTG` 列，读侧兜底。
+2. **老库分片表表头与数据行差一列**：旧 `build_chunks` 表头写 `names[lo:hi+1]`、行写 `vals[lo:hi]`
+   ⇒ 第 250 列之后的**列名整体错位一格**，且相邻分片表共用同一列名（实测 `t1` 末列 == `t2` 首列）。
+   这种库读出来是"名字对不上数"的数据 —— 比缺数据危险 ⇒ 读侧**响亮拒绝**并给出重建命令。
+3. **列名消毒两边必须同一套**：Access 列名不能含 `.` `!` `[` `]`，建库时表头被消毒（非法字符→`_`、
+   截断 60、撞名加 `_1`）。规则单一实现在 `src/windscada/mdb_names.py`：读侧把契约列名翻成库内列名、
+   取回后改回契约名 —— 否则从 MDB 取数会**静默少列**。
+4. **行序必须按 `rid` 数值排**（`rid` 在库里是文本，直接合并会得到 `1,10,100…` 字典序 —— 首轮对拍
+   "对不上"就是这个原因，数据其实没错）。
+
+### 8.2 验收（可重复）
+
+```
+python scripts/csv_to_mdb.py --month 2026-07 --class 10min    # 用修好的脚本重建一个月
+# 逐值对拍: 同一台同一月, CSV 取数 vs MDB 取数 → 全列逐值一致（含 t2/t3 分片里的列）
+```
+2026-09-19 实测：**2026-07 库 866 行 × 4 列（含跨分片列）逐值一致**；2025-03 老库被如实拒绝。
+`data.load_10min()` 从 CSV 取与从 MDB 取，返回同一张表（`attrs['scada_source']` 分别是 `csv` / `mdb`）。
+
+### 8.3 重建进度
+
+修好的建库脚本**重跑全部月份**（10min 19 个月 + 1min 18 个月，约 5 h，后台跑；日志
+`F:\temp\csv2mdb_rebuild.log`）。1min 目前包内**没有消费者**，所以"1min 的 MDB 形态"属数据层完整性补齐；
+10min 才是摄入要用的那一类。

scripts/csv_to_mdb.py

@@ -42,46 +42,72 @@ CAD = {'10min': 144, '1min': 1440}          # 每天行数（10 分钟 / 1 分
 
 
 def safe_names(cols: list[str]) -> list[str]:
-    out, seen = [], {}
-    for i, c in enumerate(cols):
-        c = c.lstrip('\ufeff')          # 源表头带 UTF-8 BOM ⇒ 不清掉会变成 _TimeStamp（实测）
-        n = re.sub(r'[^0-9A-Za-z_\u4e00-\u9fff]', '_', c)[:60] or f'c{i}'
-        if n in seen:
-            seen[n] += 1
-            n = f'{n[:56]}_{seen[n]}'
-        else:
-            seen[n] = 0
-        out.append(n)
-    return out
+    """列名消毒 —— **单一实现**在 `src/windscada/mdb_names.py`（读侧 scada_source 要用同一套规则，
+    否则从 MDB 取数时按契约列名找不到列，静默少列）。这里只做转发，保持本脚本既有调用点不变。
+    """
+    from src.windscada.mdb_names import safe_names as _sn
+    return _sn(cols)
 
 
 def build_chunks(files, month: str, tmp: pathlib.Path, limit_rows: int):
-    nt, total, handles = 0, 0, {}
+    r"""逐台 CSV → 分片 CSV（供 TransferText 装入）。
+
+    ★2026-09-19 两处修正（都是为了"这份 MDB 能被**接回去当输入**用", 见 docs/现场收资接入_v0.1.md §8）:
+
+    ① **加 `turbine` 列**：原先每行只有 `rid` + 数据列, 逐台文件**堆在一起**, 而 1min 导出件里
+       **没有任何机组标识列**(实测: `source_file` 全行同值 "如海风机1分钟数据/如海测点_2025-01.csv")
+       ⇒ 那份 1min 库**按台取不到数**, 等于只能看不能算。现在首列写文件名的 stem(10min=`WTG01`,
+       1min=`01E`), 10min 另有的 `WTG` 列原样保留。
+    ② **按列名对齐, 不再按位置堆**：原先用**第一个文件**的表头当 schema, 其余文件整行按位置写入。
+       10min 实测 38 件表头**完全一致**(1 种) ⇒ 无害; 但 1min 实测 **16 种表头**(同 79 列, 名与序都不同)
+       ⇒ 位置堆法把 A 台的列写进 B 台的列位, **静默串列**(比缺列危险得多)。现在先扫一遍所有表头取并集,
+       逐行按列名落位, 缺列留空。
+    """
+    # ① 扫表头 → 并集(首见顺序) + 逐文件的列名→位置映射
+    heads: dict[pathlib.Path, list[str]] = {}
+    union: list[str] = []
+    for f in files:
+        with open(f, encoding='utf-8', errors='replace', newline='') as fh:
+            cols = fh.readline().rstrip('\n').split(',')
+        heads[f] = cols
+        for c in cols:
+            if c not in union:
+                union.append(c)
+    if len({tuple(v) for v in heads.values()}) > 1:
+        print(f'  [i] {len({tuple(v) for v in heads.values()})} 种表头 → 按**列名**对齐(并集 {len(union)} 列); '
+              f'缺列留空 (原先按位置堆会静默串列)')
+    names = safe_names(['rid', 'turbine'] + union)
+    nt = max(1, -(-len(union) // CHUNK))
+    total, handles = 0, {}
     try:
+        for i in range(nt):
+            lo, hi = i * CHUNK, min(len(union), (i + 1) * CHUNK)
+            p = tmp / f't{i+1}.csv'
+            p.write_text(','.join(names[0:2] + names[2 + lo:2 + hi]) + '\n', encoding='utf-8', newline='')
+            handles[i] = open(p, 'a', encoding='utf-8', newline='')
+        ini = []
+        for i in range(nt):
+            ini += [f'[t{i+1}.csv]', 'Format=CSVDelimited', 'ColNameHeader=True',
+                    'CharacterSet=65001', '']
+        (tmp / 'schema.ini').write_text('\n'.join(ini), encoding='utf-8')
         for f in files:
+            cols = heads[f]
+            idx = [union.index(c) for c in cols]
             with open(f, encoding='utf-8', errors='replace', newline='') as fh:
-                cols = fh.readline().rstrip('\n').split(',')
-                names = safe_names(['rid'] + cols)
-                if not nt:
-                    nt = max(1, -(-len(cols) // CHUNK))
-                    ini = []
-                    for i in range(nt):
-                        lo, hi = i * CHUNK, min(len(cols), (i + 1) * CHUNK)
-                        p = tmp / f't{i+1}.csv'
-                        p.write_text(','.join(names[lo:hi + 1]) + '\n', encoding='utf-8', newline='')
-                        handles[i] = open(p, 'a', encoding='utf-8', newline='')
-                        ini += [f'[{p.name}]', 'Format=CSVDelimited', 'ColNameHeader=True',
-                                'CharacterSet=65001', '']
-                    (tmp / 'schema.ini').write_text('\n'.join(ini), encoding='utf-8')
+                fh.readline()
                 n = 0
                 for line in fh:
                     if line[:7] != month:
                         continue
                     vals = line.rstrip('\n').split(',')
+                    row = [''] * len(union)
+                    for k, j in enumerate(idx):
+                        if k < len(vals):
+                            row[j] = vals[k]
                     n += 1
                     for i in range(nt):
-                        lo, hi = i * CHUNK, min(len(cols), (i + 1) * CHUNK)
-                        handles[i].write(','.join([str(n)] + vals[lo:hi]) + '\n')
+                        lo, hi = i * CHUNK, min(len(union), (i + 1) * CHUNK)
+                        handles[i].write(','.join([str(n), f.stem] + row[lo:hi]) + '\n')
                     if limit_rows and n >= limit_rows:
                         break
             total += n

src/windscada/data.py

@@ -27,15 +27,58 @@ def gate(cfg=None):
                 note=f"契约 {c['version']}: 活 {alive}/{total}; 未契约列 {c['n_cols_uncontracted']} 已登记不假装覆盖")
 
 def load_10min(turbine, cfg=None, groups=None):
+    """一台的 10min 数据（**CSV 与 MDB 两种形态都认**，用户令 2026-09-19）。
+
+    取数走 `scada_source`：`<src_10min>/<台>.csv` 在位就用 CSV；该台没有 CSV 才回落
+    `data/raw/<场站>/scada_mdb/<年>年/<月>月/*.mdb`（现场年度归档形态）。用的哪种记在
+    返回表的 `attrs['scada_source']`（并在首次使用时打一行，不静默换源）。
+    """
     cfg = cfg or farm()
     g = gate(cfg)
     if not g['ok']:
         raise RuntimeError(f"L0 校验门不过: {g['note']}")
     c = contract(cfg)
     want = [c['ts_col']] + contracted_cols(cfg, groups)
-    fp = f"{cfg['src_10min']}/{turbine}.csv"
-    have = set(pd.read_csv(fp, nrows=0).columns)
-    cols = [x for x in want if x in have]   # pyarrow 禁 callable usecols; 逐台交集防个别台缺列
-    d = pd.read_csv(fp, usecols=cols, engine='pyarrow')
-    d[c['ts_col']] = pd.to_datetime(d[c['ts_col']], errors='coerce')
-    return d.dropna(subset=[c['ts_col']]).rename(columns={c['ts_col']: 'ts'})
+    from . import scada_source as SS
+    d = SS.load('10min', turbine, columns=want, cfg=cfg)
+    src = str(d.attrs.get('scada_source') or '?')
+    _note_source(src)
+    # 契约外列不加载（两形态统一）—— CSV 侧原先在 read_csv 时用 usecols 做到, MDB 侧在这里收口
+    keep = [x for x in d.columns if x in want]
+    d = d.loc[:, keep].copy()
+    if not keep:
+        raise RuntimeError(f'{turbine}: 源件里没有契约列 (想要 {want[:4]}…; 实际列 {list(d.columns)[:6]}…)')
+    ts = c['ts_col']
+    d[ts] = pd.to_datetime(d[ts], errors='coerce')
+    out = d.dropna(subset=[ts]).rename(columns={ts: 'ts'})
+    out.attrs['scada_source'] = src
+    return out
+
+
+_NOTE_DONE: set = set()
+
+
+def _note_source(src: str) -> None:
+    """首次用到某种形态时说明一次（让人知道这次重算是从 CSV 还是从 MDB 取的）。"""
+    if src in _NOTE_DONE:
+        return
+    _NOTE_DONE.add(src)
+    print(f'  [i] SCADA 源形态: {src}'
+          + ('（<src_10min>/<台>.csv）' if src == 'csv' else '（scada_mdb 年度归档库）'), flush=True)
+
+
+def source_report(cfg=None) -> dict:
+    """本机 SCADA 收资形态盘点（页面/自检用）: 各形态件数 + 逐台会走哪种。"""
+    from . import scada_source as SS
+    cfg = cfg or farm()
+    out = {}
+    for cls in ('10min', '1min'):
+        s = SS.sources(cls, cfg)
+        turbs = SS.turbines(cls, cfg)
+        kinds = {}
+        for t in turbs:
+            k = SS.source_of(cls, t, cfg) or 'none'
+            kinds[k] = kinds.get(k, 0) + 1
+        out[cls] = dict(n_csv=s['n_csv'], n_mdb=s['n_mdb'], n_zip=s['n_zip'], n_turbines=len(turbs),
+                        by_source=kinds, csv_dir=str(s['csv_dir']), mdb_root=str(s['mdb_root']))
+    return out

src/windscada/mdb_names.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+"""MDB 列名规则（单一实现）：CSV 表头 → Access 可用的列名。
+
+为什么要有这个模块：Access 的列名不能含 `.` `!` `[` `]` 等字符，建库时要把表头"消毒"成合法名；
+**摄入侧按列名取数**（契约里的列名是 CSV 原样名）⇒ 两边必须用**同一套规则**，否则从 MDB 取数会
+静默少列（用户令 2026-09-19「scada 数据接入兼容支持 CSV、MDB」的必答题）。
+
+`scripts/csv_to_mdb.py`（写侧）与 `src/windscada/scada_source.py`（读侧）都从这里取同一实现。
+"""
+from __future__ import annotations
+
+import re
+
+KEEP = re.compile(r'[^0-9A-Za-z_\u4e00-\u9fff]')
+
+
+def safe_name(col: str, maxlen: int = 60) -> str:
+    """单个表头 → 合法列名（去掉非法字符、截断；长度上限 60 便于撞名时加后缀）。"""
+    return KEEP.sub('_', str(col).lstrip('\ufeff'))[:maxlen] or 'c'
+
+
+def safe_names(cols, keep: int = 60) -> list[str]:
+    """一串表头 → 合法且**互不重名**的列名（撞名加 `_1` `_2`…，与写侧逐一对齐）。"""
+    out, seen = [], {}
+    for i, c in enumerate(cols):
+        n = safe_name(c, keep) or f'c{i}'
+        if n in seen:
+            seen[n] += 1
+            n = f'{n[:keep - 4]}_{seen[n]}'
+        else:
+            seen[n] = 0
+        out.append(n)
+    return out
+
+
+def map_cols(union: list[str]) -> dict[str, str]:
+    """源表头 → 库内列名（按 safe_names 的顺序规则算，用于读侧把契约列名翻成库内列名）。"""
+    return dict(zip(union, safe_names(list(union))))
+
+
+def stored_to_source(union: list[str]) -> dict[str, str]:
+    """库内列名 → 源表头（读侧取回后把名字改回来，让下游看到契约里的列名）。"""
+    return {v: k for k, v in map_cols(union).items()}

src/windscada/scada_source.py

@@ -0,0 +1,349 @@
+# -*- coding: utf-8 -*-
+r"""SCADA 原始数据接入层：**CSV 与 MDB 两种形态都认**（用户令 2026-09-19）。
+
+## 为什么要有这一层
+
+现场交来的 SCADA 收资有两种形态，此前只有第一种能被摄入：
+
+```
+data/raw/<场站>/scada_10min/WTG01.csv … WTG38.csv          ← 已导出 CSV（38 件 / 14.4 GB）
+data/raw/<场站>/scada_mdb/<年>年/<月>月/<年>-<月>-<类>.mdb   ← 现场年度归档的 Access 库（25年/26年）
+```
+
+`src/windscada/data.py::load_10min()` 原来把路径**写死**成 `<src_10min>/<台>.csv`。用户令之后取数统一走本层：
+**CSV 在位就用 CSV**（既有 10 个构建器一直在用的形态，行数已核过），**该台没有 CSV 才回落 MDB**，
+并把"这次是从哪种形态取的"记在返回表的 `attrs['scada_source']` 里，供产物台账/页面说明来路 —— 不静默换源。
+
+## MDB 那一侧的两个硬事实（决定了本层怎么写）
+
+1. **单表 ≤255 列 ⇒ 库内按 250 列拆表**（`t1/t2/…`）。新库每行前两列是 `rid`（该台该月内的行号）与
+   `turbine`（机组标识）⇒ 取一台的多列数据要**跨分片表按 (rid, turbine) 拼**。
+2. **2026-09-19 之前建的老库缺 `turbine` 列**（那时按位置堆行）。10min 老库还能救：源件自带 `WTG` 列，
+   本层用它兜底；**1min 老库救不了**（源件没有任何机组标识列，且 38 件表头实测 16 种 ⇒ 位置堆法还会串列）。
+   处置：重跑 `scripts/csv_to_mdb.py`（现已写 `turbine` 列 + 按列名对齐），见 `docs/现场收资接入_v0.1.md` §8。
+
+读 MDB 走 ACE（`Microsoft.ACE.OLEDB.12.0`）+ PowerShell —— 与写库同一套（本机无 pyodbc/mdbtools；
+实测可用路线见 `docs/现场收资接入_v0.1.md` §7）。日期字段在 PowerShell 里显式格式化成
+`yyyy-MM-dd HH:mm:ss`（否则会按系统区域设置输出，下游时区/日序解析会踩雷）。
+
+## 用法
+
+    from src.windscada import scada_source as SS
+    SS.sources('10min', cfg)                     # 有哪些形态、各多少件（如实，不猜）
+    SS.turbines('10min', cfg)                    # 可用的机组名（CSV stem ∪ MDB 里的 turbine）
+    SS.load('10min', 'WTG01', columns=[...], cfg=cfg)   # → DataFrame（列名原样，ts 未解析）
+    SS.source_of('10min', 'WTG01', cfg)          # → 'csv' / 'mdb'
+
+纪律: 两种形态都取不到时**响亮 raise**（不返回空表 —— 空表会被下游当成"这台没数据"）。
+"""
+from __future__ import annotations
+
+import functools
+import pathlib
+import re
+import subprocess
+import tempfile
+
+CLS = {'10min': ('src_10min', 'scada_10min'), '1min': ('src_1min', 'scada_1min')}
+TURBINE_COL = 'turbine'          # 新库的机组列（csv_to_mdb.py 2026-09-19 起写）
+LEGACY_TURBINE_COL = 'WTG'       # 老库兜底：10min 源件自带 WTG 列
+
+_PS = r'''
+$ErrorActionPreference = "Continue"
+$c = New-Object -ComObject ADODB.Connection
+$c.Open("Provider=Microsoft.ACE.OLEDB.12.0;Data Source=@@MDB@@;")
+$turb = "@@TURB@@"
+$want = @(@@WANT@@)
+# 表名清单: 用 ADO 的 OpenSchema(20=adSchemaTables)。★`$c.GetSchema("Tables")` 会报
+# "Arguments are of the wrong type..."(那是 .NET 的方法名), 实测 2026-09-19 踩过。
+$tabs = @()
+$rs0 = $c.OpenSchema(20)
+while (-not $rs0.EOF) {
+  if ($rs0.Fields.Item("TABLE_TYPE").Value -eq "TABLE") { $tabs += $rs0.Fields.Item("TABLE_NAME").Value }
+  $rs0.MoveNext()
+}
+$rs0.Close()
+foreach ($t in $tabs) {
+  $rs = $c.Execute("SELECT TOP 1 * FROM [$t]")
+  $names = @()
+  foreach ($f in $rs.Fields) { $names += $f.Name }
+  $rs.Close()
+  # 机组列: 先试 turbine(新库), 再试 WTG(老 10min 库兜底); 两个都没有 ⇒ 这库按台取不到数
+  $tcol = $null
+  foreach ($p in @("turbine", "WTG")) {
+    try { $null = $c.Execute("SELECT TOP 1 [$p] FROM [$t]"); $tcol = $p; break } catch { }
+  }
+  $turbines = @()
+  if ($tcol) {
+    $rs1 = $c.Execute("SELECT DISTINCT [$tcol] FROM [$t] WHERE [$tcol] IS NOT NULL")
+    while (-not $rs1.EOF) { $turbines += ("" + $rs1.Fields.Item(0).Value); $rs1.MoveNext() }
+    $rs1.Close()
+  }
+  Write-Output ("#SCHEMA " + $t + "|" + ($names -join ",") + "|" + ($turbines -join ";"))
+  if (-not $tcol -or $turb -eq "") { continue }
+  $keep = @()
+  foreach ($w in $want) { if (($names -contains $w) -and $w -ne "rid" -and $w -ne $tcol) { $keep += $w } }
+  if ($want.Count -eq 0) {   # 不给白名单 = 全列（与 CSV 形态 columns=None 的语义一致）
+    foreach ($w in $names) { if ($w -ne "rid" -and $w -ne $tcol) { $keep += $w } }
+  }
+  $sel = @("rid", $tcol) + $keep
+  $collist = ($sel | ForEach-Object { "[$_]" }) -join ","
+  Write-Output ("#TABLE " + $t + "|" + ($sel -join ","))
+  $rs2 = $c.Execute("SELECT $collist FROM [$t] WHERE [$tcol] = '$turb'")
+  while (-not $rs2.EOF) {
+    $vals = @()
+    for ($i = 0; $i -lt $rs2.Fields.Count; $i++) {
+      $v = $rs2.Fields.Item($i).Value
+      if ($v -is [datetime]) { $vals += $v.ToString("yyyy-MM-dd HH:mm:ss") }
+      elseif ($null -eq $v) { $vals += "" }
+      else { $vals += ("" + $v) }
+    }
+    Write-Output ($vals -join "`t")
+    $rs2.MoveNext()
+  }
+  $rs2.Close()
+}
+$c.Close()
+'''
+
+
+def dirs_of(cls: str, cfg=None) -> tuple[pathlib.Path, pathlib.Path]:
+    """→ (CSV 目录, MDB 根)。两者都不必存在。"""
+    cfg = _cfg(cfg)
+    key, sub = CLS[cls]
+    csv_dir = pathlib.Path(str(cfg.get(key) or (pathlib.Path(str(cfg['raw_station_dir'])) / sub)))
+    mdb_root = pathlib.Path(str(cfg['raw_station_dir'])) / 'scada_mdb'
+    return csv_dir, mdb_root
+
+
+def _cfg(cfg=None):
+    if cfg is not None:
+        return cfg
+    from .config import farm
+    return farm()
+
+
+def sources(cls: str = '10min', cfg=None) -> dict:
+    """盘上有哪些形态（件数）—— 供页面/自检如实说明。"""
+    csv_dir, mdb_root = dirs_of(cls, cfg)
+    csvs = sorted(csv_dir.glob('*.csv')) if csv_dir.is_dir() else []
+    mdbs = sorted(mdb_root.rglob(f'*-{cls}.mdb')) if mdb_root.is_dir() else []
+    zips = sorted(mdb_root.rglob(f'*-{cls}.zip')) if mdb_root.is_dir() else []
+    return dict(csv_dir=csv_dir, mdb_root=mdb_root, n_csv=len(csvs), n_mdb=len(mdbs), n_zip=len(zips),
+                csvs=csvs, mdbs=mdbs, zips=zips,
+                kind=('csv' if csvs else ('mdb' if (mdbs or zips) else None)))
+
+
+@functools.lru_cache(maxsize=8)
+def schema_and_turbines(mdb: str) -> tuple:
+    """一个 MDB 的 (分片表 → 列名, 分片表 → 该列里的机组值)。老库缺 turbine 列时退 WTG 列。"""
+    lines = _ps(_PS.replace('@@MDB@@', str(mdb)).replace('@@TURB@@', '').replace('@@WANT@@', ''))
+    cols, turbs = {}, {}
+    for ln in lines:
+        if ln.startswith('#SCHEMA '):
+            body = ln[8:]
+            t, rest = body.split('|', 1)
+            names, tv = rest.split('|', 1)
+            cols[t] = [x for x in names.split(',') if x]
+            turbs[t] = tuple(sorted({x for x in tv.split(';') if x}))
+    return cols, turbs
+
+
+def _ps(script: str) -> list[str]:
+    """跑一段 PowerShell（ACE 只能经 COM 用）→ stdout 行。
+
+    ★退出码不可尽信（2026-09-19 实测）: ACE 在**进程退出时清理 COM** 偶发 `0xC0000005`（访问违例），
+    此时 stderr 为空、stdout 完整（脚本里每一行都打出来了）。原先 `rc != 0 ⇒ raise` 会把这种"数据都拿到了"
+    的情况报成读库失败。判据改成**看输出**: 有输出且退出码是那个崩溃码 ⇒ 收下并提示一次; 否则才 raise。
+    """
+    p = pathlib.Path(tempfile.gettempdir()) / '_scada_source.ps1'
+    p.write_text(script + '\nexit 0\n', encoding='utf-8-sig')
+    r = subprocess.run(['powershell', '-NoProfile', '-ExecutionPolicy', 'Bypass', '-File', str(p)],
+                       capture_output=True, text=True, errors='replace', timeout=3600)
+    out = [x for x in (r.stdout or '').splitlines() if x.strip()]
+    crash = r.returncode in (-1073741819, 3221225477)
+    if r.returncode != 0 and not (crash and out):
+        raise RuntimeError((r.stderr or r.stdout or '').strip()[:400] or 'PowerShell 读 MDB 失败')
+    if crash and out and 'ace_exit_crash' not in _WARNED:
+        _WARNED.add('ace_exit_crash')
+        print('  [i] ACE 退出时 COM 清理崩溃 (0xC0000005) —— 输出完整, 已按成功处理', flush=True)
+    return out
+
+
+_WARNED: set = set()
+
+
+def mdb_turbines(cls: str = '10min', cfg=None, limit: int = 3) -> set:
+    """MDB 侧的可用机组名（取前几个库的并集 —— 逐库扫全量太慢，够用）。"""
+    s = sources(cls, cfg)
+    got = set()
+    for m in s['mdbs'][:limit]:
+        _cols, turbs = schema_and_turbines(str(m))
+        for t in turbs.values():
+            got |= set(t)
+    return got
+
+
+def turbines(cls: str = '10min', cfg=None) -> list[str]:
+    """可用机组名 = CSV 文件名 stem ∪ MDB 里的 turbine 值。"""
+    s = sources(cls, cfg)
+    got = {f.stem for f in s['csvs']} | mdb_turbines(cls, cfg)
+    return sorted(got)
+
+
+def source_of(cls: str, turbine: str, cfg=None) -> str | None:
+    """这一台这次会从哪种形态取（csv 优先）。"""
+    csv_dir, _ = dirs_of(cls, cfg)
+    if (csv_dir / f'{turbine}.csv').is_file():
+        return 'csv'
+    return 'mdb' if turbine in mdb_turbines(cls, cfg, limit=1) else None
+
+
+def load(cls: str, turbine: str, columns=None, cfg=None, month: str | None = None):
+    """取一台的数据（**CSV 优先**；该台没有 CSV 时回落 MDB）。
+
+    columns = 白名单（两形态都按"实际有的列"取交集，缺列不报错 —— 与 `data.load_10min` 的既有口径一致）。
+    month   = 'YYYY-MM'，只取该月（MDB 形态天然是按月的；CSV 形态按首列前缀过滤）。
+    """
+    import pandas as pd
+    csv_dir, _ = dirs_of(cls, cfg)
+    fp = csv_dir / f'{turbine}.csv'
+    if fp.is_file():
+        if columns:
+            have = set(pd.read_csv(fp, nrows=0).columns)
+            cols = [c for c in columns if c in have]
+        else:
+            cols = None
+        d = pd.read_csv(fp, usecols=cols, engine='pyarrow') if cols else pd.read_csv(fp, engine='pyarrow')
+        if month:
+            ts = d.columns[0]
+            d = d[d[ts].astype(str).str.startswith(month)]
+        d.attrs['scada_source'] = 'csv'
+        return d.reset_index(drop=True)
+    if source_of(cls, turbine, cfg) != 'mdb':
+        raise RuntimeError(
+            f'{cls} 取数失败: {turbine} 既没有 CSV ({fp})，也没有含该台的 MDB '
+            f'({dirs_of(cls, cfg)[1]})。放原始件到 data/raw/<场站>/ 后重跑 scripts/rebuild_all.py。')
+    d = read_mdb(cls, turbine, columns, cfg, month)
+    d.attrs['scada_source'] = 'mdb'
+    return d
+
+
+def read_mdb(cls: str, turbine: str, columns=None, cfg=None, month: str | None = None):
+    """从月库里取一台：逐分片表按 (rid, turbine) 拼列。"""
+    import pandas as pd
+    from .mdb_names import safe_name
+    s = sources(cls, cfg)
+    mdbs = [m for m in s['mdbs'] if not month or m.stem.startswith(month)]
+    if not mdbs:
+        raise RuntimeError(f'{cls} 取数失败: scada_mdb 下没有 {month or "任何"} 的 .mdb')
+    # 契约列名 → 库内列名（Access 列名不能含 `.` `!` `[]` 等, 建库时消毒过; 两边同一套规则）
+    cand = {c: safe_name(c) for c in (columns or [])}
+    want_stored = sorted(set(cand.values()))
+    frames, missing_col = [], None
+    for m in mdbs:
+        want = want_stored
+        script = (_PS.replace('@@MDB@@', str(m)).replace('@@TURB@@', str(turbine))
+                  .replace('@@WANT@@', ','.join('"' + w + '"' for w in want)))
+        cur_t, cur_cols, buf, seen_cols = None, [], [], set()
+
+        def _flush():
+            if cur_t and buf:
+                frames.append(_frame(cur_cols, buf))
+        for ln in _ps(script):
+            if ln.startswith('#SCHEMA '):
+                continue
+            if ln.startswith('#TABLE '):
+                _flush()
+                body = ln[7:]
+                cur_t, cols = body.split('|', 1)
+                if not re.fullmatch(r't\d+', cur_t):        # Access 自建的杂表(如"名称自动更正保存失败")
+                    cur_t, cur_cols, buf = None, [], []
+                    continue
+                cur_cols, buf = cols.split(','), []
+                seen_cols |= set(cur_cols)
+            else:
+                buf.append(ln)
+        _flush()
+        if columns and not any(c in seen_cols for c in columns):
+            missing_col = [c for c in columns if c not in seen_cols][:3]
+    # 列位错位检测: 用 **#SCHEMA**(每个分片表的列清单, 与本次取了哪些列无关)
+    dup = set()
+    for m2 in mdbs:
+        sc = {}
+        for ln in _ps(_PS.replace('@@MDB@@', str(m2)).replace('@@TURB@@', '').replace('@@WANT@@', '')):
+            if ln.startswith('#SCHEMA '):
+                body = ln[8:]
+                t2, rest = body.split('|', 1)
+                if re.fullmatch(r't\d+', t2):
+                    sc[t2] = set(x for x in rest.split('|', 1)[0].split(',') if x)
+        ks = list(sc)
+        for i in range(len(ks)):
+            for j in range(i + 1, len(ks)):
+                dup |= (sc[ks[i]] & sc[ks[j]]) - {'rid', TURBINE_COL}
+        if dup:
+            break
+    if not frames:
+        raise RuntimeError(
+            f'{cls} 取数失败: {turbine} 在 {[m.name for m in mdbs]} 里取不到行 —— 老库可能没有 turbine 列'
+            f'（源件也没有 WTG 列）⇒ 重跑 scripts/csv_to_mdb.py 建新库（现已写 turbine 列）')
+    frames = [f.rename(columns={LEGACY_TURBINE_COL: TURBINE_COL}) if
+              (LEGACY_TURBINE_COL in f.columns and TURBINE_COL not in f.columns) else f for f in frames]
+    # 库内列名 → 契约列名（撞名时建库侧写过 `_1` 后缀, 这里按前缀认回来）
+    if cand:
+        for f in frames:
+            back, used = {}, set()
+            for orig, st in cand.items():
+                if st in f.columns and st not in used:
+                    back[st], _ = orig, used.add(st)
+                    continue
+                stem = st[:56]
+                for c2 in f.columns:
+                    if c2 not in used and c2.startswith(stem) and c2[len(stem):].startswith('_'):
+                        back[c2], _ = orig, used.add(c2)
+                        break
+            f.rename(columns=back, inplace=True)
+    # ★2026-09-19 逮到的**老库列位错位**: 2026-09-19 之前建的 10min 库, 分片表表头与数据行差一列
+    #   (旧 build_chunks 表头写 names[lo:hi+1] 而行写 vals[lo:hi]) ⇒ 第 250 列之后的**列名整体错位一格**,
+    #   且相邻分片表会共用同一个列名(实测 t1 末列 == t2 首列 == flg_wtc_ScReToOp_endvalue)。这种库读出来的
+    #   是"名字对不上数"的数据 —— 比缺数据危险, 故**响亮拒绝**, 让人重建而不是将就着算。
+    if dup:
+        raise RuntimeError(
+            f'{cls} 取数失败: {[m.name for m in mdbs]} 是**列位错位的老库**(分片表共用列名 {sorted(dup)[:3]}…) '
+            f'—— 2026-09-19 前的建库脚本表头与数据行差一列, 第 250 列之后的列名整体错位。'
+            f'处置: python scripts/csv_to_mdb.py --month <YYYY-MM> --class {cls} 重建该月库; '
+            f'或放 CSV 到 data/raw/<场站>/{CLS[cls][1]}/ 直接用 CSV 形态。')
+    d = frames[0]
+    for f in frames[1:]:
+        d = d.merge(f, on=['rid', TURBINE_COL], how='outer')
+    # 行序回到**源件顺序**: rid 在库里是文本 ⇒ 直接合并得到的是 '1','10','100' 这种字典序
+    # (2026-09-19 实逮: 与 CSV 逐值对拍对不上, 数据没错、是行序被按字符串排了)
+    d['rid'] = pd.to_numeric(d['rid'], errors='coerce')
+    d = d.sort_values([TURBINE_COL, 'rid'], kind='stable').reset_index(drop=True)
+    if missing_col:
+        d.attrs['mdb_missing'] = missing_col
+    return d
+
+
+def _frame(cols: list[str], rows: list[str]):
+    """PowerShell 落的制表符分隔文本 → DataFrame。
+
+    数值列转成数值: MDB 里存的是**数值类型**, ACE 读出来 `100.0` 会变成 `100` —— 若原样留成字符串,
+    下游按 CSV 口径写的老构建器会拿到 object 列 (`+`/`mean` 全崩)。这里逐列试探: 非空值全能解析成数字
+    就转数值, 否则留字符串 (台号/状态字这类)。
+    """
+    import pandas as pd
+    ncol = len(cols)
+    recs = []
+    for ln in rows:
+        vals = ln.split('\t')
+        recs.append((vals + [''] * ncol)[:ncol])
+    d = pd.DataFrame(recs, columns=cols)
+    for c in d.columns:
+        if c in ('rid', TURBINE_COL, LEGACY_TURBINE_COL):
+            continue
+        s = d[c].replace('', None)
+        num = pd.to_numeric(s, errors='coerce')
+        if num.notna().sum() == s.notna().sum() and s.notna().any():
+            d[c] = num
+    return d