aps-agent/server/aps_domain/report_contracts.py

"""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()
    ]