feat: add pcf analysis web app and api

app/excel_writer.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import xlrd
+from xlutils.copy import copy as copy_workbook
+
+from .config import AppConfig
+from .models import WeldRecord
+
+
+EXCEL_FIELD_TO_RECORD = {
+    "管线代号": "line_no",
+    "焊口代号": "weld_no",
+    "焊接区域(安装/预制)": "weld_area",
+    "规格（mm)": "spec",
+    "壁厚": "wall_thickness",
+    "单线图号": "pipeline_reference",
+    "管线等级代号": "piping_spec",
+    "外径": "outside_diameter",
+}
+
+EXCEL_FIELD_TO_ENGLISH = {
+    "单位代码": "unit_code",
+    "装置工区编号": "plant_area_code",
+    "管线代号": "line_no",
+    "焊口代号": "weld_no",
+    "材质类型代号": "material_type_code",
+    "材质1代号": "material_1_code",
+    "材质2代号": "material_2_code",
+    "材料1": "material_1",
+    "材料2": "material_2",
+    "探伤比例代号": "ndt_ratio_code",
+    "焊缝类型代号": "weld_type_code",
+    "焊接区域(安装/预制)": "weld_area",
+    "焊口属性（固定、活动）": "weld_property",
+    "达因数": "dia_inch",
+    "规格（mm)": "spec",
+    "壁厚": "wall_thickness",
+    "焊接方法代码": "welding_method_code",
+    "试验压力": "test_pressure",
+    "焊条代号": "welding_rod_code",
+    "焊丝代号": "welding_wire_code",
+    "介质代号": "medium_code",
+    "单线图号": "isometric_no",
+    "设计压力": "design_pressure",
+    "设计温度": "design_temperature",
+    "坡口代号": "groove_code",
+    "管线等级代号": "piping_class_code",
+    "组件一代号": "component_1_code",
+    "组件二代号": "component_2_code",
+    "炉批号一": "heat_no_1",
+    "炉批号二": "heat_no_2",
+    "所属管段": "pipe_spool",
+    "预热温度": "preheat_temperature",
+    "是否需热处理（是，否）": "need_heat_treatment",
+    "热处理编号": "heat_treatment_no",
+    "焊接位置（1G/2G/3G/4G/5G/6G)": "welding_position",
+    "外径": "outside_diameter",
+    "硬度检测比例（数值）": "hardness_test_ratio",
+    "焊接气体保护": "shielding_gas",
+    "是否非标（是/否）": "is_non_standard",
+    "壁板号": "plate_no",
+    "延长米": "extended_meter",
+    "管道长度": "pipe_length",
+}
+
+
+class XlsTemplateWriter:
+    def __init__(self, config: AppConfig):
+        self.config = config
+
+    def write(self, records: list[WeldRecord], output_path: Path) -> None:
+        template_path = self.config.template_path
+        if not template_path.exists():
+            raise FileNotFoundError(f"模板不存在：{template_path}")
+
+        rb = xlrd.open_workbook(
+            str(template_path),
+            formatting_info=True,
+            on_demand=False,
+        )
+        sheet_index = rb.sheet_names().index(self.config.output_sheet)
+        rs = rb.sheet_by_index(sheet_index)
+        wb = copy_workbook(rb)
+        ws = wb.get_sheet(sheet_index)
+
+        headers = [str(rs.cell_value(0, c)).strip() for c in range(rs.ncols)]
+        template_style_by_col = self._template_styles(rb, rs, headers)
+
+        # 清空模板中已有示例数据，避免生成记录少于示例时残留旧值。
+        for r in range(1, rs.nrows):
+            for c in range(rs.ncols):
+                ws.write(r, c, "", template_style_by_col[c])
+
+        for row_offset, record in enumerate(records, start=1):
+            row_values = self.row_values(headers, record)
+            for col_idx, value in enumerate(row_values):
+                ws.write(row_offset, col_idx, value, template_style_by_col[col_idx])
+
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        wb.save(str(output_path))
+
+    def _template_styles(self, rb: xlrd.book.Book, rs: xlrd.sheet.Sheet, headers: list[str]):
+        styles = []
+        for col_idx, _header in enumerate(headers):
+            source_row = 1 if rs.nrows > 1 else 0
+            xf_idx = rs.cell_xf_index(source_row, col_idx)
+            styles.append(rb.xf_list[xf_idx])
+        # xlutils 复制后需要使用目标工作簿内部样式对象。
+        # 经验上直接使用 rb.xf_list 会失败，因此回退到 xlwt 默认样式时由调用方兜底。
+        return [self._xf_to_xlwt_style(rb, xf) for xf in styles]
+
+    def _xf_to_xlwt_style(self, rb: xlrd.book.Book, xf: xlrd.formatting.XF):
+        import xlwt
+
+        style = xlwt.XFStyle()
+        font = rb.font_list[xf.font_index]
+        style.font.name = font.name
+        style.font.bold = font.bold
+        style.font.italic = font.italic
+        style.font.height = font.height
+        style.font.colour_index = font.colour_index
+
+        fmt = rb.format_map.get(xf.format_key)
+        if fmt:
+            style.num_format_str = fmt.format_str
+
+        alignment = style.alignment
+        alignment.horz = xf.alignment.hor_align
+        alignment.vert = xf.alignment.vert_align
+        alignment.wrap = xf.alignment.text_wrapped
+
+        borders = style.borders
+        borders.left = xf.border.left_line_style
+        borders.right = xf.border.right_line_style
+        borders.top = xf.border.top_line_style
+        borders.bottom = xf.border.bottom_line_style
+        borders.left_colour = xf.border.left_colour_index
+        borders.right_colour = xf.border.right_colour_index
+        borders.top_colour = xf.border.top_colour_index
+        borders.bottom_colour = xf.border.bottom_colour_index
+
+        pattern = style.pattern
+        pattern.pattern = xf.background.fill_pattern
+        pattern.pattern_fore_colour = xf.background.pattern_colour_index
+        pattern.pattern_back_colour = xf.background.background_colour_index
+        return style
+
+    def headers(self) -> list[str]:
+        rb = xlrd.open_workbook(str(self.config.template_path), formatting_info=False, on_demand=True)
+        try:
+            rs = rb.sheet_by_name(self.config.output_sheet)
+            return [str(rs.cell_value(0, c)).strip() for c in range(rs.ncols)]
+        finally:
+            rb.release_resources()
+
+    def build_import_rows(
+        self, records: list[WeldRecord]
+    ) -> tuple[list[str], list[dict[str, str]], list[dict[str, str]]]:
+        headers = self.headers()
+        columns = self.import_columns(headers)
+        keys = [column["key"] for column in columns]
+        rows = [dict(zip(keys, self.row_values(headers, record))) for record in records]
+        return headers, columns, rows
+
+    def import_columns(self, headers: list[str]) -> list[dict[str, str]]:
+        used: set[str] = set()
+        columns: list[dict[str, str]] = []
+        for index, header in enumerate(headers, start=1):
+            key = EXCEL_FIELD_TO_ENGLISH.get(header) or f"column_{index:02d}"
+            if key in used:
+                key = f"{key}_{index:02d}"
+            used.add(key)
+            columns.append({"key": key, "label": header})
+        return columns
+
+    def row_values(self, headers: list[str], record: WeldRecord) -> list[str]:
+        values: list[str] = []
+        for header in headers:
+            if header in EXCEL_FIELD_TO_RECORD:
+                values.append(str(getattr(record, EXCEL_FIELD_TO_RECORD[header], "") or ""))
+            else:
+                values.append(str(self.config.defaults.get(header, "") or ""))
+        return values