"""flatness-suite 批量测量结果的模板关联与旧报表解析。""" from __future__ import annotations import argparse import json import math import posixpath from pathlib import Path from typing import Any from zipfile import BadZipFile, ZipFile from xml.etree import ElementTree as ET _SHEET_NS = {"a": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"} _REL_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships" def _column_index(reference: str) -> int: letters = "".join(character for character in reference if character.isalpha()) if not letters: raise ValueError(f"无效的 Excel 单元格引用: {reference}") result = 0 for character in letters.upper(): result = result * 26 + ord(character) - ord("A") + 1 return result def _cell_value(cell: ET.Element, shared_strings: list[str]) -> Any: cell_type = cell.attrib.get("t") if cell_type == "inlineStr": return "".join(text.text or "" for text in cell.findall(".//a:t", _SHEET_NS)) value_node = cell.find("a:v", _SHEET_NS) if value_node is None or value_node.text is None: return "" raw = value_node.text if cell_type == "s": try: return shared_strings[int(raw)] except (IndexError, ValueError) as exc: raise ValueError("Excel 共享字符串索引无效") from exc if cell_type in {"str", "e"}: return raw if cell_type == "b": return raw == "1" try: number = float(raw) return int(number) if number.is_integer() else number except ValueError: return raw def read_xlsx_rows(path: str | Path) -> tuple[str, list[list[Any]]]: """只用标准库读取 xlsx 的第一个工作表。""" workbook_path = Path(path) if not workbook_path.is_file(): raise FileNotFoundError(f"Excel 文件不存在: {workbook_path}") try: with ZipFile(workbook_path) as xlsx: names = set(xlsx.namelist()) shared_strings: list[str] = [] if "xl/sharedStrings.xml" in names: shared_root = ET.fromstring(xlsx.read("xl/sharedStrings.xml")) shared_strings = [ "".join(text.text or "" for text in item.findall(".//a:t", _SHEET_NS)) for item in shared_root.findall("a:si", _SHEET_NS) ] workbook_root = ET.fromstring(xlsx.read("xl/workbook.xml")) first_sheet = workbook_root.find("a:sheets/a:sheet", _SHEET_NS) if first_sheet is None: raise ValueError(f"Excel 中没有工作表: {workbook_path}") sheet_name = first_sheet.attrib.get("name", "Sheet1") relation_id = first_sheet.attrib.get(f"{{{_REL_NS}}}id") if not relation_id: raise ValueError(f"Excel 工作表关系缺失: {workbook_path}") rels_root = ET.fromstring(xlsx.read("xl/_rels/workbook.xml.rels")) relation_targets = { relation.attrib["Id"]: relation.attrib["Target"] for relation in rels_root } target = relation_targets.get(relation_id) if not target: raise ValueError(f"Excel 工作表文件关系缺失: {workbook_path}") sheet_path = target.lstrip("/") if target.startswith("/") else posixpath.normpath(posixpath.join("xl", target)) if sheet_path not in names: raise ValueError(f"Excel 工作表文件不存在: {sheet_path}") sheet_root = ET.fromstring(xlsx.read(sheet_path)) sparse_rows: list[dict[int, Any]] = [] max_column = 0 for row in sheet_root.findall("a:sheetData/a:row", _SHEET_NS): values: dict[int, Any] = {} for cell in row.findall("a:c", _SHEET_NS): column = _column_index(cell.attrib.get("r", "")) values[column] = _cell_value(cell, shared_strings) max_column = max(max_column, column) if values: sparse_rows.append(values) except (BadZipFile, KeyError, ET.ParseError) as exc: raise ValueError(f"无法解析 Excel 文件 {workbook_path}: {exc}") from exc rows = [ [row.get(column, "") for column in range(1, max_column + 1)] for row in sparse_rows ] if not rows: raise ValueError(f"Excel 工作表为空: {workbook_path}") return sheet_name, rows def _header_index(headers: list[Any], candidates: tuple[str, ...]) -> int: normalized = [str(header).strip() for header in headers] for candidate in candidates: if candidate in normalized: return normalized.index(candidate) raise ValueError(f"Excel 缺少列: {'/'.join(candidates)}") def normalize_task_id(value: Any) -> str: if isinstance(value, float) and value.is_integer(): return str(int(value)) text = str(value).strip() return text[:-2] if text.endswith(".0") and text[:-2].isdigit() else text def _label(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 read_measurement_template(path: str | Path) -> dict[str, dict[str, str]]: """读取任务号对应的项点名称与原始公差文本。""" _, rows = read_xlsx_rows(path) headers = rows[0] task_column = _header_index(headers, ("任务号", "任务编号", "task_id")) name_column = _header_index(headers, ("项点名称", "项目名称", "项目")) tolerance_column = _header_index(headers, ("公差要求", "公差")) result: dict[str, dict[str, str]] = {} for row in rows[1:]: task_id = normalize_task_id(row[task_column] if task_column < len(row) else "") if not task_id: continue item_name = _label(row[name_column] if name_column < len(row) else "") if not item_name: raise ValueError(f"模板任务 {task_id} 缺少项点名称") result[task_id] = { "item_name": item_name, "tolerance_label": _label(row[tolerance_column] if tolerance_column < len(row) else ""), } if not result: raise ValueError(f"模板中没有测量任务: {path}") return result def read_legacy_batch_rows(path: str | Path) -> list[dict[str, Any]]: """读取旧版技术汇总表中的任务号、平均值和状态。""" _, rows = read_xlsx_rows(path) headers = rows[0] task_column = _header_index(headers, ("任务号", "任务编号", "task_id")) mean_column = _header_index(headers, ("平均值", "mean", "mean_dist", "average_distance")) status_column = _header_index(headers, ("状态", "测量结论", "结论")) result: list[dict[str, Any]] = [] for row in rows[1:]: task_id = normalize_task_id(row[task_column] if task_column < len(row) else "") if not task_id: continue result.append({ "task_id": task_id, "mean": row[mean_column] if mean_column < len(row) else None, "status": row[status_column] if status_column < len(row) else "", }) if not result: raise ValueError(f"旧版 Excel 中没有批量结果: {path}") return result def _finite_float(value: Any) -> float | None: try: number = float(value) except (TypeError, ValueError): return None return number if math.isfinite(number) else None def build_measurements( rows: list[dict[str, Any]], template_path: str | Path, ) -> list[dict[str, Any]]: template = read_measurement_template(template_path) status_fields = { "成功": ("pass", "成功"), "部分成功": ("warning", "部分成功"), "失败": ("fail", "失败"), } measurements: list[dict[str, Any]] = [] for row in rows: task_id = normalize_task_id(row.get("task_id", "")) if not task_id: raise ValueError("批量结果包含空任务号") template_row = template.get(task_id) if template_row is None: raise ValueError(f"模板中找不到任务 {task_id}") raw_status = str(row.get("status", "")).strip() if raw_status not in status_fields: raise ValueError(f"任务 {task_id} 的状态无效: {raw_status or '<空>'}") status, _ = status_fields[raw_status] measurements.append({ "name": f"任务{task_id} · {template_row['item_name']}", "value": _finite_float(row.get("mean")), "unit": "mm", "toleranceLabel": template_row["tolerance_label"], "status": status, "passLabel": "成功", "warningLabel": "部分成功", "failLabel": "失败", }) return measurements def measurements_from_batch_result(result: dict[str, Any], template_path: str | Path) -> list[dict[str, Any]]: rows = result.get("rows") if not isinstance(rows, list): raise ValueError("run_contour_batch 返回值缺少 rows") return build_measurements(rows, template_path) def main() -> int: parser = argparse.ArgumentParser(description="从旧版 flatness-suite Excel 构造通用 measurements") parser.add_argument("--excel", required=True) parser.add_argument("--template", required=True) args = parser.parse_args() measurements = build_measurements(read_legacy_batch_rows(args.excel), args.template) print(json.dumps(measurements, ensure_ascii=False, allow_nan=False)) return 0 if __name__ == "__main__": raise SystemExit(main())