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