guanlan-src/app_ontology/app_ontology_guanlan/sop/xlsx_read.py

# -*- coding: utf-8 -*-
"""xlsx_read.py — SCADA xlsx 按表头名读 (解"位置读列串列"陷阱, SOP §1 铁律).

SOP §1: 异构 .xls position-based 读列 → 列序漂移=串列 (新艾里偏航 2676° 假值自逮 3 次);
必**中文表头 name-based 映射**。本模块 = 该铁律的机器件:
  - detect_header_row: 扫前几行自动定位"含必需名的表头行" (双表头如太阳山 英文码行0/中文名行1 → 定到行1)
  - read_scada_xlsx: 按该行当列名读 + **必需列 name-check** (缺 → raise, 不静默读错位置当数据)

为什么 name-check 是核心: 若表头行猜错(如对着英文码行), 中文必需列 name 找不到 → 响亮 raise;
而 iloc 位置读会静默把英文码行当数据/串列 → 假值下游。响亮失败 > 静默串列。
"""
import pandas as pd


def detect_header_row(src, must_contain, max_scan=6, engine='openpyxl'):
    """扫前 max_scan 行, 找**同时含 must_contain 全部名**的行 → 返回其 0基行号 (表头行)。无→None。
    src 若为 BytesIO/文件流, 调用方须可重读 (本函数读后流位置会动; 传路径或每次新建 BytesIO)。"""
    raw = pd.read_excel(src, header=None, nrows=max_scan, engine=engine)
    for i in range(len(raw)):
        cells = {str(x).strip() for x in raw.iloc[i].tolist()}
        if all(m in cells for m in must_contain):
            return i
    return None


def read_scada_xlsx(src, header_row=0, required=None, usecols=None, engine='openpyxl'):
    """按表头名读 SCADA xlsx (name-based, 禁 iloc 位置读)。

    header_row: 用作列名的表头行 (0基)。双表头(英文码+中文名)读中文 → header_row=1;
                header_row=None → 用 detect_header_row(required) 自动定位 (required 必给)。
    required: 必需列名清单 → name-check: 缺任一 → raise (防表头行猜错静默串列)。
    usecols: 只保留这些列 (name-based)。返回列名 strip 的 name-based df。
    """
    if header_row is None:
        if not required:
            raise ValueError('read_scada_xlsx: header_row=None 时 required 必给 (供自动定位表头行)')
        header_row = detect_header_row(src, required[:2] if len(required) >= 2 else required, engine=engine)
        if header_row is None:
            raise ValueError(f'read_scada_xlsx: 自动定位表头行失败 (前几行无含 {required[:2]} 的行); '
                             '表头名对否? 该场是否转置格式(变量在行)?')
    df = pd.read_excel(src, header=header_row, engine=engine)
    df.columns = [str(c).strip() for c in df.columns]
    if required:
        missing = [c for c in required if c not in df.columns]
        if missing:
            raise ValueError(f'read_scada_xlsx 缺必需列 {missing} (name-based 防串列; header_row={header_row} 对否?). '
                             f'实际列前12: {list(df.columns)[:12]}')
    if usecols:
        df = df[[c for c in usecols if c in df.columns]]
    return df