217 lines
6.8 KiB
Python
217 lines
6.8 KiB
Python
"""解析数据管理导入用的 Excel / CSV（中文表头、【示例】行跳过）。
|
||
|
||
实现 ``replaces: import`` 时请复用本模块，勿只认「后缀 *」表头。
|
||
|
||
宿主契约（须兼容）：
|
||
|
||
- 必填红星写在显示名前：``*长尾词`` / ``* 名称``（旧技能文档可能写 ``名称*``）
|
||
- lookup 单元格可能是 ``157 · 波兰小包专线``（``{value} · {label}``），不只是纯数字
|
||
- 以 ``【示例】`` 开头的单元格所在行应跳过
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any, Iterable, Optional
|
||
|
||
EXAMPLE_MARKER = "【示例】"
|
||
DATA_SHEET_NAMES = ("数据", "Sheet1", "sheet1")
|
||
|
||
|
||
def normalize_on_error(raw: Optional[str]) -> str:
|
||
text = str(raw or "").strip().lower()
|
||
if text in {
|
||
"abort",
|
||
"stop",
|
||
"立即停止，不再导入后续行",
|
||
"立即停止",
|
||
"停止",
|
||
}:
|
||
return "abort"
|
||
# Chinese default / skip variants
|
||
return "skip"
|
||
|
||
|
||
def normalize_header(header: str) -> str:
|
||
"""归一化导入表头。
|
||
|
||
宿主模板把必填红星写在名称前（``*长尾词`` / ``* 长尾词``）；
|
||
旧技能文档常用后缀（``长尾词*``）。两侧都要认。
|
||
"""
|
||
text = str(header or "").strip().replace("\ufeff", "")
|
||
for _ in range(4):
|
||
prev = text
|
||
text = text.strip()
|
||
for star in ("*", "＊"):
|
||
text = text.removeprefix(star).removesuffix(star).strip()
|
||
for marker in ("（必填）", "(required)"):
|
||
text = text.removeprefix(marker).removesuffix(marker).strip()
|
||
if text == prev:
|
||
break
|
||
return text
|
||
|
||
|
||
def _is_example_row(cells: Iterable[Any]) -> bool:
|
||
for cell in cells:
|
||
text = str(cell or "").strip()
|
||
if text.startswith(EXAMPLE_MARKER):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _cell_str(value: Any) -> str:
|
||
if value is None:
|
||
return ""
|
||
if isinstance(value, float) and value.is_integer():
|
||
return str(int(value))
|
||
return str(value).strip()
|
||
|
||
|
||
def map_headers(
|
||
headers: list[str],
|
||
aliases: dict[str, tuple[str, ...]],
|
||
) -> dict[str, int]:
|
||
"""aliases: field_name -> accepted Chinese/English header names."""
|
||
normalized = [normalize_header(h) for h in headers]
|
||
mapping: dict[str, int] = {}
|
||
for field, names in aliases.items():
|
||
accepted = {normalize_header(n) for n in names}
|
||
accepted.add(normalize_header(field))
|
||
for index, header in enumerate(normalized):
|
||
if header in accepted:
|
||
mapping[field] = index
|
||
break
|
||
return mapping
|
||
|
||
|
||
def rows_as_dicts(
|
||
headers: list[str],
|
||
data_rows: list[list[Any]],
|
||
*,
|
||
aliases: dict[str, tuple[str, ...]],
|
||
) -> list[dict[str, str]]:
|
||
mapping = map_headers(headers, aliases)
|
||
if not mapping:
|
||
raise ValueError("未识别到有效表头，请使用技能模板或中文列名")
|
||
out: list[dict[str, str]] = []
|
||
for row_index, raw in enumerate(data_rows, start=2):
|
||
if _is_example_row(raw):
|
||
continue
|
||
if not any(_cell_str(c) for c in raw):
|
||
continue
|
||
item: dict[str, str] = {"_row": str(row_index)}
|
||
for field, col in mapping.items():
|
||
item[field] = _cell_str(raw[col] if col < len(raw) else "")
|
||
out.append(item)
|
||
return out
|
||
|
||
|
||
def read_csv_matrix(path: Path) -> tuple[list[str], list[list[str]]]:
|
||
raw = path.read_bytes()
|
||
text = raw.decode("utf-8-sig")
|
||
try:
|
||
dialect = csv.Sniffer().sniff(text[:4096], delimiters=",\t;")
|
||
except csv.Error:
|
||
dialect = csv.excel
|
||
reader = csv.reader(text.splitlines(), dialect)
|
||
rows = [list(r) for r in reader]
|
||
if not rows:
|
||
raise ValueError("CSV 为空")
|
||
return [str(c) for c in rows[0]], rows[1:]
|
||
|
||
|
||
def read_xlsx_matrix(path: Path) -> tuple[list[str], list[list[Any]]]:
|
||
try:
|
||
from openpyxl import load_workbook
|
||
except ImportError as exc:
|
||
raise ValueError("读取 Excel 需要 openpyxl，请安装依赖或另存为 CSV 后再导入") from exc
|
||
|
||
wb = load_workbook(path, read_only=True, data_only=True)
|
||
try:
|
||
sheet = None
|
||
for name in DATA_SHEET_NAMES:
|
||
if name in wb.sheetnames:
|
||
sheet = wb[name]
|
||
break
|
||
if sheet is None:
|
||
# Prefer first non-guide sheet
|
||
for name in wb.sheetnames:
|
||
if name not in ("填写说明",):
|
||
sheet = wb[name]
|
||
break
|
||
if sheet is None:
|
||
raise ValueError("Excel 中没有可用工作表")
|
||
matrix: list[list[Any]] = []
|
||
for row in sheet.iter_rows(values_only=True):
|
||
matrix.append(list(row))
|
||
if not matrix:
|
||
raise ValueError("Excel 数据表为空")
|
||
headers = [_cell_str(c) for c in matrix[0]]
|
||
return headers, matrix[1:]
|
||
finally:
|
||
wb.close()
|
||
|
||
|
||
def load_table_rows(
|
||
file_path: str,
|
||
*,
|
||
aliases: dict[str, tuple[str, ...]],
|
||
) -> list[dict[str, str]]:
|
||
path = Path(file_path)
|
||
if not path.is_file():
|
||
raise FileNotFoundError(f"文件不存在：{path}")
|
||
suffix = path.suffix.lower()
|
||
if suffix in {".xlsx", ".xlsm"}:
|
||
headers, data = read_xlsx_matrix(path)
|
||
elif suffix in {".csv", ".txt"}:
|
||
headers, data = read_csv_matrix(path)
|
||
else:
|
||
raise ValueError("仅支持 .xlsx / .csv 文件")
|
||
return rows_as_dicts(headers, data, aliases=aliases)
|
||
|
||
|
||
_STATUS_ALIASES = {
|
||
"active": "active",
|
||
"启用": "active",
|
||
"archived": "archived",
|
||
"停用": "archived",
|
||
"pending_review": "pending_review",
|
||
"待确认": "pending_review",
|
||
"approved": "approved",
|
||
"已采用": "approved",
|
||
"skipped": "skipped",
|
||
"已跳过": "skipped",
|
||
"pending": "pending",
|
||
"待生成": "pending",
|
||
"generated": "generated",
|
||
"已生成": "generated",
|
||
"failed": "failed",
|
||
"失败": "failed",
|
||
}
|
||
|
||
|
||
def map_status(raw: str, *, allowed: set[str], default: str) -> str:
|
||
text = str(raw or "").strip()
|
||
if not text:
|
||
return default
|
||
key = _STATUS_ALIASES.get(text) or _STATUS_ALIASES.get(text.lower()) or text
|
||
return key if key in allowed else default
|
||
|
||
|
||
def optional_int(raw: str) -> Optional[int]:
|
||
"""解析编号；兼容宿主 lookup 展示值 ``157 · 波兰小包专线``。"""
|
||
text = str(raw or "").strip()
|
||
if not text or text.lower() in {"null", "none", "undefined"}:
|
||
return None
|
||
if re.fullmatch(r"-?\d+", text):
|
||
value = int(text)
|
||
return None if value <= 0 else value
|
||
# Host lookup labels: "157 · 名称" / "157·名称" / "157 - 名称"
|
||
matched = re.match(r"^(-?\d+)\s*[·•\-—|｜]\s*\S", text)
|
||
if matched:
|
||
value = int(matched.group(1))
|
||
return None if value <= 0 else value
|
||
raise ValueError(f"编号必须是整数：{text}")
|