186 lines
7.0 KiB
Python
186 lines
7.0 KiB
Python
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
|