"""Versioned contracts for every scheduling workbook and intake template. The contracts are data, not prose: sheet names, header rows, data offsets, column order, units, required state and the value source of each column are declared once and reused by the Excel builders and by the verifiers/tests. Business values never live here. """ from __future__ import annotations from collections.abc import Mapping, Sequence from dataclasses import dataclass from typing import Any TEMPLATE_HEADER_STYLE = { "fillColor": "1F4E79", "fontColor": "FFFFFF", "fontSize": 11, "borderColor": "D0D5DD", } BLOCKED_HEADER_STYLE = { "fillColor": "C2410C", "fontColor": "FFFFFF", "fontSize": 11, "borderColor": "D0D5DD", } OVERVIEW_TITLE_STYLE = {"fontColor": "1F4E79", "fontSize": 14} @dataclass(frozen=True) class ColumnSpec: """One physical column: header text plus its declared meaning.""" key: str header: str kind: str unit: str required_state: str source: str @dataclass(frozen=True) class SheetSpec: name: str title: str title_row: int header_row: int data_start_row: int freeze_panes: str | None auto_filter: bool columns: tuple[ColumnSpec, ...] @property def headers(self) -> tuple[str, ...]: return tuple(column.header for column in self.columns) @property def last_column_index(self) -> int: return len(self.columns) @dataclass(frozen=True) class ReportContract: id: str schema_version: int report_type: str label: str filename_template: str sheets: Mapping[str, SheetSpec] primary_sheet: str def _columns(*specs: tuple[str, str, str, str, str, str]) -> tuple[ColumnSpec, ...]: return tuple(ColumnSpec(*spec) for spec in specs) _PLAN_COLUMNS = _columns( ("orderNo", "订单号", "string", "-", "always", "flexWorkOrders.flexOrderNo|orderNo"), ("productCode", "产品编码", "string", "-", "always", "flexWorkOrders.productCode"), ("quantity", "数量", "number", "件", "always", "flexOrders.quantity"), ("dueDate", "交期", "date", "-", "always", "flexOrders.dueDate"), ("seq", "序", "number", "-", "always", "flexWorkOrders.seq"), ("operationCode", "工序编码", "string", "-", "always", "flexWorkOrders.operationCode"), ("operationName", "工序名称", "string", "-", "always", "flexWorkOrders.operationName"), ("equipmentName", "工位/设备", "string", "-", "always", "flexWorkOrders.equipmentName|equipmentCode"), ("equipmentCode", "设备编码", "string", "-", "always", "flexWorkOrders.equipmentCode"), ("zone", "区域", "string", "-", "optional", "flexWorkOrders.zone"), ("moldCode", "模具", "string", "-", "optional", "flexWorkOrders.moldCode"), ("changeoverMin", "换型(分)", "number", "分钟", "always", "flexWorkOrders.changeoverMin"), ("moveMin", "移栽(分)", "number", "分钟", "always", "flexWorkOrders.moveMin"), ("runMin", "加工(分)", "number", "分钟", "always", "flexWorkOrders.runMin"), ("start", "计划开始", "datetime", "-", "always", "flexWorkOrders.plannedStartTime"), ("end", "计划结束", "datetime", "-", "always", "flexWorkOrders.plannedEndTime"), ("isBottleneck", "瓶颈", "boolean", "-", "always", "flexWorkOrders.isBottleneck"), ("status", "状态", "string", "-", "always", "flexWorkOrders.status"), ) _ORDER_COLUMNS = _columns( ("orderNo", "订单号", "string", "-", "always", "flexWorkOrders.flexOrderNo|orderNo"), ("workOrderNo", "工单号", "string", "-", "always", "flexWorkOrders.orderNo"), ("productCode", "产品编码", "string", "-", "always", "flexWorkOrders.productCode"), ("operationCode", "工序编码", "string", "-", "always", "flexWorkOrders.operationCode"), ("operationName", "工序名称", "string", "-", "always", "flexWorkOrders.operationName"), ("resourceCode", "资源编码", "string", "-", "always", "flexWorkOrders.equipmentCode"), ("resourceName", "资源名称", "string", "-", "always", "flexWorkOrders.equipmentName|equipmentCode"), ("start", "计划开始", "datetime", "-", "always", "flexWorkOrders.plannedStartTime"), ("end", "计划结束", "datetime", "-", "always", "flexWorkOrders.plannedEndTime"), ("status", "状态", "string", "-", "always", "flexWorkOrders.status"), ("plannedQuantity", "预计数量", "number", "件", "always", "flexWorkOrders.quantity"), ("doneQuantity", "实际数量", "number", "件", "always", "flexWorkOrders.qtyDone"), ) _EQUIPMENT_COLUMNS = _columns( ("equipmentCode", "设备编码", "string", "-", "always", "flexWorkOrders.equipmentCode"), ("equipmentName", "设备名称", "string", "-", "always", "flexWorkOrders.equipmentName|equipmentCode"), ("operationCode", "工序编码", "string", "-", "always", "flexWorkOrders.operationCode"), ("operationName", "工序名称", "string", "-", "always", "flexWorkOrders.operationName"), ("start", "占用开始", "datetime", "-", "always", "flexWorkOrders.plannedStartTime"), ("end", "占用结束", "datetime", "-", "always", "flexWorkOrders.plannedEndTime"), ("loadMinutes", "负荷分钟", "number", "分钟", "always", "derived.runMin+moldMin+moveMin"), ("utilization", "利用率%", "number", "%", "always", "derived.loadMinutes/availableMinutes"), ) _BLOCKED_COLUMNS = _columns( ("conflictType", "阻断类型", "string", "-", "always", "flexConflicts.conflictType|type"), ("subject", "对象", "string", "-", "optional", "flexConflicts.orderNo|resourceName|resourceType"), ("severity", "严重度", "string", "-", "always", "flexConflicts.severity"), ("description", "说明", "string", "-", "always", "flexConflicts.description"), ("suggestion", "修复建议", "string", "-", "always", "flexConflicts.suggestedSolution|suggestion"), ) def _sheet( name: str, title: str, columns: tuple[ColumnSpec, ...], *, title_row: int = 1, header_row: int = 4, auto_filter: bool = True, ) -> SheetSpec: return SheetSpec( name=name, title=title, title_row=title_row, header_row=header_row, data_start_row=header_row + 1, freeze_panes=f"A{header_row + 1}", auto_filter=auto_filter, columns=columns, ) PLAN_REPORT_CONTRACT = ReportContract( id="schedule-plan.v1", schema_version=1, report_type="plan", label="排产工作计划表", filename_template="排产工作计划_{scope}_{versionNo}.xlsx", primary_sheet="工作计划", sheets={ "plan": _sheet("工作计划", "排产工作计划表", _PLAN_COLUMNS), "overview": SheetSpec( name="方案概览", title="方案概览", title_row=1, header_row=12, data_start_row=13, freeze_panes=None, auto_filter=False, columns=_columns( ("orderNo", "订单号", "string", "-", "always", "flexVirtualLines.orderNo"), ("productCode", "产品", "string", "-", "always", "flexVirtualLines.productCode"), ("quantity", "数量", "number", "件", "always", "flexVirtualLines.quantity"), ("start", "计划开始", "datetime", "-", "always", "flexVirtualLines.plannedStart"), ("end", "计划结束", "datetime", "-", "always", "flexVirtualLines.plannedEnd"), ("dueDate", "交期", "date", "-", "always", "flexOrders.dueDate"), ("onTime", "是否按期", "enum", "-", "always", "derived.end<=dueDate"), )), "conflicts": _sheet( "冲突", "冲突与风险", _columns( ("conflictType", "类型", "string", "-", "always", "flexConflicts.conflictType|type"), ("orderNo", "订单号", "string", "-", "optional", "flexConflicts.orderNo"), ("severity", "严重度", "string", "-", "always", "flexConflicts.severity"), ("description", "描述", "string", "-", "always", "flexConflicts.description"), ("suggestion", "建议", "string", "-", "always", "flexConflicts.suggestedSolution|suggestion"), ), header_row=3, ), }, ) SCHEDULE_ORDER_CONTRACT = ReportContract( id="schedule-order.v1", schema_version=1, report_type="schedule-order", label="排产工单方案", filename_template="排产工单方案_{scope}_{versionNo}.xlsx", primary_sheet="排产工单", sheets={ "orders": _sheet("排产工单", "排产工单方案", _ORDER_COLUMNS), "overview": SheetSpec( name="方案概览", title="方案概览", title_row=1, header_row=0, data_start_row=3, freeze_panes=None, auto_filter=False, columns=()), }, ) SCHEDULE_EQUIPMENT_CONTRACT = ReportContract( id="schedule-equipment.v1", schema_version=1, report_type="schedule-equipment", label="排产设备方案", filename_template="排产设备方案_{scope}_{versionNo}.xlsx", primary_sheet="设备负荷", sheets={ "equipment": _sheet("设备负荷", "排产设备方案", _EQUIPMENT_COLUMNS), "overview": SheetSpec( name="方案概览", title="方案概览", title_row=1, header_row=0, data_start_row=3, freeze_panes=None, auto_filter=False, columns=()), }, ) SCHEDULE_BLOCKED_CONTRACT = ReportContract( id="schedule-blocked.v1", schema_version=1, report_type="plan", label="排产阻断分析", filename_template="排产阻断分析_{scope}_{versionNo}.xlsx", primary_sheet="阻断分析", sheets={"blocked": _sheet("阻断分析", "排产阻断分析", _BLOCKED_COLUMNS)}, ) PLAN_EMPTY_STATE = { "id": "schedule-plan-empty", "schema_version": 1, "report_type": "plan", "label": "排产工作计划表空态", "xlsxBytes": None, "filename": None, "reason": "no-flexible-version", "message": "尚无排产版本。请先生成柔性排产方案或交期优先试排方案。", } REPORT_CONTRACTS: Mapping[str, ReportContract] = { PLAN_REPORT_CONTRACT.id: PLAN_REPORT_CONTRACT, SCHEDULE_ORDER_CONTRACT.id: SCHEDULE_ORDER_CONTRACT, SCHEDULE_EQUIPMENT_CONTRACT.id: SCHEDULE_EQUIPMENT_CONTRACT, SCHEDULE_BLOCKED_CONTRACT.id: SCHEDULE_BLOCKED_CONTRACT, } def contract_for_report_type(report_type: str) -> ReportContract | None: """Resolve the primary workbook contract for a report type.""" aliases = { "plan": PLAN_REPORT_CONTRACT.id, "schedule-plan": PLAN_REPORT_CONTRACT.id, "flex-plan": PLAN_REPORT_CONTRACT.id, "schedule-order": SCHEDULE_ORDER_CONTRACT.id, "schedule-equipment": SCHEDULE_EQUIPMENT_CONTRACT.id, } key = aliases.get(report_type) return REPORT_CONTRACTS.get(key) if key else None def build_contract_template(report_type: str) -> dict[str, Any]: """把合同本身导出成一份表头级空白工作簿，供现场核对导出格式。 模板只写标题、表头、冻结窗格与筛选区域，不含任何业务行；生成后立刻用 ``validate_workbook_contract`` 复核，因此下载到的模板与真正导出的文件由 同一份声明驱动，不会出现模板和导出对不上的情况。 """ import hashlib from io import BytesIO from openpyxl import Workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter from server.aps_domain.reports import _deterministic_xlsx_bytes from server.timeutil import today0 contract = contract_for_report_type(report_type) or REPORT_CONTRACTS.get(report_type) if contract is None: raise ValueError(f"未注册的排产模板：{report_type}") thin = Border( left=Side(style="thin", color=TEMPLATE_HEADER_STYLE["borderColor"]), right=Side(style="thin", color=TEMPLATE_HEADER_STYLE["borderColor"]), top=Side(style="thin", color=TEMPLATE_HEADER_STYLE["borderColor"]), bottom=Side(style="thin", color=TEMPLATE_HEADER_STYLE["borderColor"]), ) head_fill = PatternFill("solid", fgColor=TEMPLATE_HEADER_STYLE["fillColor"]) head_font = Font( color=TEMPLATE_HEADER_STYLE["fontColor"], bold=True, size=TEMPLATE_HEADER_STYLE["fontSize"] ) title_font = Font( bold=True, size=OVERVIEW_TITLE_STYLE["fontSize"], color=OVERVIEW_TITLE_STYLE["fontColor"] ) workbook = Workbook() for index, spec in enumerate(contract.sheets.values()): sheet = workbook.active if index == 0 else workbook.create_sheet() sheet.title = spec.name if spec.title_row and spec.title: sheet.cell(spec.title_row, 1, spec.title).font = title_font note_row = (spec.title_row or 0) + 1 if note_row and (not spec.header_row or note_row < spec.header_row): sheet.cell( note_row, 1, f"模板 {contract.id} · 表头行 {spec.header_row or '—'} · " f"数据起始行 {spec.data_start_row} · 表头与列顺序不可改动", ) if spec.header_row: for column_index, column in enumerate(spec.columns, 1): cell = sheet.cell(spec.header_row, column_index, column.header) cell.fill = head_fill cell.font = head_font cell.alignment = Alignment(horizontal="center", vertical="center") cell.border = thin sheet.column_dimensions[get_column_letter(column_index)].width = min( 28, max(10, len(column.header) * 2 + 4) ) if spec.freeze_panes: sheet.freeze_panes = spec.freeze_panes if spec.auto_filter: sheet.auto_filter.ref = ( f"A{spec.header_row}:" f"{get_column_letter(spec.last_column_index)}{spec.header_row}" ) workbook.properties.created = workbook.properties.modified = today0() buffer = BytesIO() workbook.save(buffer) payload = _deterministic_xlsx_bytes(buffer.getvalue()) defects = validate_workbook_contract(payload, contract) if defects: raise RuntimeError(f"排产模板不符合 {contract.id}：" + "；".join(defects)) return { "templateId": contract.id, "contractId": contract.id, "contractDigest": contract_digest(), "schemaVersion": contract.schema_version, "reportType": contract.report_type, "label": f"{contract.label}模板", "primarySheet": contract.primary_sheet, "sheets": [ { "key": key, "name": spec.name, "headerRow": spec.header_row, "dataStartRow": spec.data_start_row, "headers": list(spec.headers), } for key, spec in contract.sheets.items() ], "xlsxBytes": payload, "filename": f"{contract.label}模板_{contract.id}.xlsx", "format": "xlsx", "sha256": hashlib.sha256(payload).hexdigest(), } def validate_workbook_contract(payload: bytes, contract: ReportContract) -> list[str]: """Verify a generated workbook against the declared contract. Returns a list of human-readable defects; an empty list means every sheet, header cell, row offset, freeze pane and auto-filter matches the contract. """ from io import BytesIO import openpyxl from openpyxl.utils import get_column_letter defects: list[str] = [] try: workbook = openpyxl.load_workbook(BytesIO(payload), read_only=False, data_only=False) except Exception as exc: # noqa: BLE001 - callers need the raw defect message return [f"工作簿无法打开：{exc}"] try: expected_names = {spec.name for spec in contract.sheets.values()} actual_names = set(workbook.sheetnames) for missing in sorted(expected_names - actual_names): defects.append(f"缺少工作表：{missing}") for extra in sorted(actual_names - expected_names): defects.append(f"存在合同外工作表：{extra}") for key, spec in contract.sheets.items(): if spec.name not in actual_names: continue sheet = workbook[spec.name] if spec.header_row: for column_index, column in enumerate(spec.columns, 1): actual = sheet.cell(spec.header_row, column_index).value if actual != column.header: defects.append( f"{key}!{get_column_letter(column_index)}{spec.header_row} " f"表头应为“{column.header}”，实际为“{actual}”" ) if spec.freeze_panes is not None and sheet.freeze_panes != spec.freeze_panes: defects.append(f"{key} 冻结窗格应为 {spec.freeze_panes}，实际为 {sheet.freeze_panes}") if spec.auto_filter and sheet.auto_filter.ref is None: defects.append(f"{key} 缺少自动筛选区域") if spec.title_row and spec.title: actual_title = str(sheet.cell(spec.title_row, 1).value or "") if not actual_title.startswith(spec.title): defects.append(f"{key}!A{spec.title_row} 标题应以“{spec.title}”开头，实际为“{actual_title}”") finally: workbook.close() return defects def contract_digest() -> str: """Stable digest of the declared scheduling report contracts.""" import hashlib import json payload = { key: { "schemaVersion": contract.schema_version, "reportType": contract.report_type, "primarySheet": contract.primary_sheet, "filenameTemplate": contract.filename_template, "sheets": { sheet_key: { "name": spec.name, "title": spec.title, "titleRow": spec.title_row, "headerRow": spec.header_row, "dataStartRow": spec.data_start_row, "freezePanes": spec.freeze_panes, "autoFilter": spec.auto_filter, "columns": [ {"key": column.key, "header": column.header, "kind": column.kind, "unit": column.unit, "required": column.required_state, "source": column.source} for column in spec.columns ], } for sheet_key, spec in contract.sheets.items() }, } for key, contract in REPORT_CONTRACTS.items() } return hashlib.sha256( json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode() ).hexdigest() def contract_summary() -> Sequence[dict[str, Any]]: """Machine-readable summary used by docs, diagnostics and tests.""" return [ { "id": contract.id, "schemaVersion": contract.schema_version, "reportType": contract.report_type, "label": contract.label, "primarySheet": contract.primary_sheet, "sheets": [ { "name": spec.name, "headerRow": spec.header_row, "dataStartRow": spec.data_start_row, "columns": [column.header for column in spec.columns], } for spec in contract.sheets.values() ], } for contract in REPORT_CONTRACTS.values() ]