aps-agent/server/importers/template_workbook.py

"""Generate the data-intake Excel template from the active workbook profile.

Sheet names, column names, enum vocabularies and the mapping digest all come
from the validated JSON profile. Nothing here is customer specific: switching
``APS_WORKBOOK_PROFILE_DIR`` or ``profile_id`` produces that profile's template
and the template round-trips through ``preview_file`` unchanged.
"""
from __future__ import annotations

import hashlib
import json
from collections.abc import Mapping
from datetime import date, datetime
from typing import Any

from server.importers.workbook_profiles import (
    ROLE_FIELDS,
    active_compatibility_profile,
    get_profile,
)

TEMPLATE_EXAMPLE_MAX_ROWS = 3
TEMPLATE_SHEET_KEY = "intake"

_ROLE_KINDS = {role: ("materials" if role == "products" else role) for role in ROLE_FIELDS}
_ROLE_LABELS = {
    "sourceNotes": "资料说明", "factoryResources": "工厂资源", "equipment": "设备",
    "personnel": "人员技能", "calendar": "班次日历", "products": "产品", "materials": "物料",
    "bom": "BOM", "routing": "工艺路线", "inventory": "库存与在途", "partners": "客户供应商",
    "orders": "销售订单", "wip": "在制任务", "planningParameters": "排产参数",
    "sandboxScenario": "插单场景", "sourceValidation": "数据校验",
}
_PLACEHOLDERS = {
    "sourceValidation": {
        "key": ("示例：数量平衡", "Example: quantity balance"),
        "reportedActual": ("实际值由现场核对后填写", "Fill the verified actual value"),
        "reportedTarget": ("目标值由现场核对后填写", "Fill the verified target value"),
        "reportedResult": ("结果由现场核对后填写", "Fill the verified result"),
    }
}

# 采集说明里声明的允许取值来自解析器使用的同一份枚举配置，模板与读取逻辑不允许各写一套。
_ENUM_FIELDS: dict[tuple[str, str], tuple[str, str]] = {
    ("factoryResources", "resourceKind"): ("resourceKind", "资源类型："),
    ("factoryResources", "status"): ("resourceStatus", "资源状态："),
    ("equipment", "status"): ("equipmentStatus", "设备状态："),
    ("calendar", "statusOrReason"): ("enabled", "班次状态："),
    ("products", "type"): ("materialType", "物料类型："),
    ("products", "sourcingType"): ("sourcingType", "供应方式："),
    ("materials", "type"): ("materialType", "物料类型："),
    ("materials", "sourcingType"): ("sourcingType", "供应方式："),
    ("bom", "isKey"): ("yesNo", "是否关键："),
    ("routing", "isExternal"): ("yesNo", "是否外协："),
    ("partners", "partnerType"): ("partnerType", "伙伴类型："),
    ("orders", "orderType"): ("orderType", "订单分类："),
    ("orders", "status"): ("orderStatus", "订单状态："),
    ("wip", "status"): ("wipStatus", "任务状态："),
    ("planningParameters", "delivery"): ("enabled", "交期参数："),
    ("planningParameters", "bottleneck"): ("enabled", "瓶颈参数："),
}

DEFINITION_HEADERS = (
    "角色", "角色名称", "工作表", "字段顺序", "业务字段", "是否必填", "类型", "填写要求", "来源说明",
)


def _enum_rule(profile: dict, role: str, field: str) -> str | None:
    entry = _ENUM_FIELDS.get((role, field))
    if not entry:
        return None
    enum_name, prefix = entry
    options = profile["enums"].get(enum_name) or {}
    if not options:
        return None
    rendered: list[str] = []
    for source, canonical in options.items():
        text = str(source)
        if str(canonical) not in ("", text):
            text = f"{text}→{canonical}"
        if text not in rendered:
            rendered.append(text)
    return prefix + "；".join(rendered)


def _fill_rule(profile: dict, role: str, field: str) -> str:
    enum_rule = _enum_rule(profile, role, field)
    if enum_rule:
        return enum_rule
    kind = _field_kind(role, field)
    if kind == "日期/时间":
        return "日期 YYYY-MM-DD；时间 HH:MM"
    if kind == "数值":
        return "数字，不写单位"
    return "—"


def intake_definition_rows(profile: dict) -> list[list[Any]]:
    """采集说明的逐字段定义：模板与解析器共用同一份 profile。"""
    rows: list[list[Any]] = []
    for role, spec in profile["sheets"].items():
        required = set(spec.get("requiredColumns") or ())
        for column_index, field in enumerate(spec["columns"], 1):
            rows.append([
                role, _ROLE_LABELS.get(role, role), spec["name"], column_index, field,
                "必填" if field in required else "可选", _field_kind(role, field),
                _fill_rule(profile, role, field), f"{spec['columns'][field]}（来源列）",
            ])
    return rows


def _style():
    from openpyxl.styles import Alignment, Border, Font, PatternFill, Side

    thin = Border(
        left=Side(style="thin", color="D0D5DD"), right=Side(style="thin", color="D0D5DD"),
        top=Side(style="thin", color="D0D5DD"), bottom=Side(style="thin", color="D0D5DD"),
    )
    head_fill = PatternFill("solid", fgColor="1F4E79")
    head_font = Font(color="FFFFFF", bold=True, size=11)
    title_font = Font(bold=True, size=14, color="1F4E79")
    body = Alignment(vertical="center", wrap_text=False)
    return thin, head_fill, head_font, title_font, body


def build_intake_template(
    profile_id: str | None = None,
    *,
    include_examples: bool = False,
    world: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Build one profile-driven intake workbook.

    ``include_examples`` writes the returned row examples next to the required
    source headings so planners can see the expected shape. Examples are never
    written unless the caller opts in, and never replace a blank template.
    """
    profile = get_profile(profile_id) if profile_id else active_compatibility_profile()
    rows = _example_rows(profile, world or {}) if include_examples else {}
    return _build_workbook(profile, rows)


def _reverse_enum(profile: dict, name: str, value: Any) -> str | None:
    for source, canonical in profile["enums"].get(name, {}).items():
        if canonical == value:
            return source
    return None


def _row_values(profile: dict, role: str, row: Mapping[str, Any]) -> dict[str, Any]:
    """Reverse-map one imported business row back to physical source columns."""
    values: dict[str, Any] = {}
    spec = profile["sheets"][role]
    for field in spec["columns"]:
        value = row.get(field)
        if role == "factoryResources" and field == "resourceKind":
            value = _reverse_enum(profile, "resourceKind", row.get("resourceKind")) or row.get("resourceType")
        elif role == "factoryResources" and field == "status":
            value = _reverse_enum(profile, "resourceStatus", row.get("status"))
        elif role == "equipment" and field == "status":
            value = _reverse_enum(profile, "equipmentStatus", row.get("status"))
        elif role in ("products", "materials") and field == "type":
            value = _reverse_enum(profile, "materialType", row.get("type"))
        elif role in ("products", "materials") and field == "sourcingType":
            value = _reverse_enum(profile, "sourcingType", row.get("sourcingType"))
        elif role == "orders" and field == "orderType":
            value = _reverse_enum(profile, "orderType", row.get("orderType"))
        elif role == "orders" and field == "status":
            value = _reverse_enum(profile, "orderStatus", row.get("status"))
        elif role == "wip" and field == "status":
            value = _reverse_enum(profile, "wipStatus", row.get("status"))
        elif role == "partners" and field == "partnerType":
            value = _reverse_enum(profile, "partnerType", row.get("partnerType"))
        if isinstance(value, (list, tuple)):
            value = profile["separators"]["list"].join(str(item) for item in value)
        if isinstance(value, date) and not isinstance(value, datetime):
            value = value.isoformat()
        values[field] = value
    return values


def _example_rows(profile: dict, world: Mapping[str, Any]) -> dict[str, list[dict[str, Any]]]:
    tables = {
        "factoryResources": world.get("flexFactoryResources") or [],
        "equipment": world.get("flexEquipment") or [],
        "personnel": world.get("flexPersonnel") or [],
        "calendar": world.get("flexCalendar") or [],
        "products": [row for row in (world.get("flexMaterials") or []) if row.get("type") == "FINISHED_PRODUCT"],
        "materials": [row for row in (world.get("flexMaterials") or []) if row.get("type") != "FINISHED_PRODUCT"],
        "bom": world.get("flexBom") or [],
        "routing": world.get("flexRoutings") or [],
        "inventory": world.get("flexMaterials") or [],
        "partners": world.get("flexPartners") or [],
        "orders": world.get("flexOrders") or [],
        "wip": world.get("flexWip") or [],
    }
    rows: dict[str, list[dict[str, Any]]] = {}
    for role in profile["sheets"]:
        if role in ("sourceNotes", "planningParameters", "sandboxScenario", "sourceValidation"):
            continue
        examples = []
        for row in (tables.get(role) or [])[:TEMPLATE_EXAMPLE_MAX_ROWS]:
            examples.append(_row_values(profile, role, row))
        rows[role] = examples
    return rows


def _build_workbook(profile: dict, examples: Mapping[str, list[dict[str, Any]]]) -> dict[str, Any]:
    from io import BytesIO

    from openpyxl import Workbook
    from openpyxl.utils import get_column_letter

    from server.aps_domain.reports import _deterministic_xlsx_bytes
    from server.timeutil import today0

    thin, head_fill, head_font, title_font, body = _style()
    workbook = Workbook()
    workbook.properties.created = workbook.properties.modified = today0()
    intake = workbook.active
    intake.title = "数据采集说明"
    intake["A1"] = "APS 排产数据采集模板"
    intake["A1"].font = title_font
    intake["A2"] = (
        f"配置 {profile['id']} · schemaVersion {profile['schemaVersion']} · "
        f"映射摘要 {profile['profileDigest']}"
    )
    intake["A3"] = "填写规则：所有工作表表头不要改动；未提供的行留空；缺项必须保持可见，系统不会自动补默认值。"
    intake["A4"] = "能力与限制：" + "、".join(profile["capabilities"])
    for column, header in enumerate(DEFINITION_HEADERS, 1):
        cell = intake.cell(6, column, header)
        cell.fill = head_fill
        cell.font = head_font
        cell.border = thin
    for row_index, values in enumerate(intake_definition_rows(profile), 7):
        for column, value in enumerate(values, 1):
            cell = intake.cell(row_index, column, value)
            cell.border = thin
    for column, width in enumerate([16, 14, 16, 10, 22, 10, 12, 44, 26], 1):
        intake.column_dimensions[get_column_letter(column)].width = width
    intake.freeze_panes = "A7"

    for role, spec in profile["sheets"].items():
        sheet = workbook.create_sheet(spec["name"])
        for column, field in enumerate(spec["columns"], 1):
            cell = sheet.cell(1, column, spec["columns"][field])
            cell.fill = head_fill
            cell.font = head_font
            cell.border = thin
        for offset, example in enumerate(examples.get(role) or [], 2):
            for column, field in enumerate(spec["columns"], 1):
                value = example.get(field)
                if value in (None, ""):
                    value = _placeholder(profile, role, field)
                cell = sheet.cell(offset, column, value if value is not None else "")
                cell.border = thin
                cell.alignment = body
        for column, field in enumerate(spec["columns"], 1):
            sheet.column_dimensions[get_column_letter(column)].width = min(
                32, max(10, len(str(spec["columns"][field])) * 2 + 4)
            )
        sheet.freeze_panes = "A2"

    buffer = BytesIO()
    workbook.save(buffer)
    payload = _deterministic_xlsx_bytes(buffer.getvalue())
    return {
        "templateId": f"intake-{profile['id']}.v1",
        "profileId": profile["id"],
        "profileDigest": profile["profileDigest"],
        "schemaVersion": profile["schemaVersion"],
        "capabilities": list(profile["capabilities"]),
        "sheetNames": [profile["sheets"][role]["name"] for role in profile["sheets"]],
        "roleOrder": list(profile["sheets"]),
        "intakeSheet": "数据采集说明",
        "rows": {role: len(examples.get(role) or []) for role in profile["sheets"]},
        "xlsxBytes": payload,
        "filename": f"{profile['label']}_{profile['id']}_采集模板.xlsx",
        "format": "xlsx",
        "sha256": hashlib.sha256(payload).hexdigest(),
    }


def _field_kind(role: str, field: str) -> str:
    if (role, field) in _ENUM_FIELDS:
        return "枚举"
    if field.lower().endswith(("date", "time")) or field in {
        "start", "end", "completionTime", "orderDate", "dueDate", "expectedArrivalDate", "shiftCode",
    }:
        return "日期/时间"
    if field in {
        "quantity", "stock", "inTransit", "safetyStock", "procurementLeadTime", "availabilityRate",
        "seq", "stdTimePerUnit", "lossRate", "completedQuantity", "priority", "horizonDays",
        "freezeHours", "reportedActual", "reportedTarget",
    }:
        return "数值"
    if field in {"isKey", "isExternal", "status", "type", "sourcingType", "resourceKind", "partnerType"}:
        return "枚举/布尔"
    return "文本"


def _placeholder(profile: dict, role: str, field: str) -> str | None:
    schemas = _PLACEHOLDERS.get(role)
    if schemas and field in schemas:
        return schemas[field][0]
    return None


def validate_intake_template(payload: bytes, profile: dict | None = None) -> list[str]:
    """Round-trip contract check: generated headings must match the profile."""
    from io import BytesIO

    import openpyxl

    selected = profile or active_compatibility_profile()
    defects: list[str] = []
    try:
        workbook = openpyxl.load_workbook(BytesIO(payload), read_only=True, data_only=False)
    except Exception as exc:  # noqa: BLE001 - callers need the raw defect message
        return [f"模板无法打开：{exc}"]
    try:
        expected_sheets = [selected["sheets"][role]["name"] for role in selected["sheets"]]
        actual_sheets = list(workbook.sheetnames)
        if not actual_sheets or actual_sheets[0] != "数据采集说明":
            defects.append("模板第一张工作表必须是“数据采集说明”")
        for name in expected_sheets:
            if name not in actual_sheets:
                defects.append(f"模板缺少工作表：{name}")
        for spec in selected["sheets"].values():
            name = spec["name"]
            if name not in actual_sheets:
                continue
            sheet = workbook[name]
            headers = [cell.value for cell in next(sheet.iter_rows(min_row=1, max_row=1))]
            expected = list(spec["columns"].values())
            if [str(value or "") for value in headers[: len(expected)]] != expected:
                defects.append(f"{name} 表头与配置不一致")
            if any(value not in (None, "") for value in headers[len(expected):]):
                defects.append(f"{name} 存在配置外的列")
    finally:
        workbook.close()
    return defects


def template_contract() -> dict[str, Any]:
    """Machine-readable template contract shared with docs and diagnostics."""
    profile = active_compatibility_profile()
    return {
        "id": "intake-template.v1",
        "schemaVersion": 1,
        "intakeSheet": "数据采集说明",
        "definitionHeaders": list(DEFINITION_HEADERS),
        "profileId": profile["id"],
        "profileSchemaVersion": profile["schemaVersion"],
        "roleOrder": list(profile["sheets"]),
        "sheets": [
            {
                "role": role,
                "name": spec["name"],
                "columns": list(spec["columns"].values()),
                "requiredColumns": list(spec.get("requiredColumns") or ()),
            }
            for role, spec in profile["sheets"].items()
        ],
    }


def template_contract_json() -> str:
    return json.dumps(template_contract(), ensure_ascii=False, indent=2, sort_keys=True)