| 12345678910111213141516171819202122232425262728293031323334353637383940414243 |
- # -*- 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()}
|
