"""Validated, data-only workbook adapter registry. APS_WORKBOOK_PROFILE_DIR replaces the built-in directory. Profiles describe physical layout and vocabularies; supported business roles remain code contracts. No profile can load Python, relax confirmation, or declare production readiness. """ from __future__ import annotations import hashlib import json import os from pathlib import Path from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError BUILTIN_PROFILE_DIR = Path(__file__).parent / "profiles" / "workbooks" ROLE_FIELDS = { "sourceNotes": ("key", "value"), "factoryResources": ("resourceKind", "code", "name", "parentCode", "description", "status"), "equipment": ("code", "name", "internalCode", "assetCode", "spec", "capabilities", "zone", "status", "availabilityRate", "sourceText"), "personnel": ("code", "name", "department", "role", "teamName", "skills", "skillLevel", "shiftCode", "sourceText"), "calendar": ("eventCode", "name", "start", "end", "breaks", "workdaysOrEquipment", "statusOrReason"), "products": ("code", "name", "spec", "unit", "type", "sourcingType", "stock", "sourceText"), "materials": ("code", "name", "spec", "type", "unit", "sourcingType", "safetyStock", "procurementLeadTime", "sourceText"), "bom": ("productCode", "productName", "materialCode", "materialName", "quantity", "isKey", "lossRate"), "routing": ("productCode", "productName", "seq", "operationCode", "operationName", "stdTimePerUnit", "sourceText", "requiredCapability", "isExternal"), "inventory": ("code", "name", "stock", "inTransit", "safetyStock", "procurementLeadTime", "expectedArrivalDate", "inventoryScenario"), "partners": ("partnerType", "code", "name", "level", "contactOrCategory", "phone", "leadTimeText"), "orders": ("orderType", "orderNo", "customerName", "customerLevel", "productCode", "productName", "quantity", "orderDate", "dueDate", "priority", "status", "scenario"), "wip": ("taskNo", "orderNo", "productCode", "operationName", "equipmentCode", "status", "completedQuantity", "completionTime"), "planningParameters": ("key", "delivery", "bottleneck", "description"), "sandboxScenario": ("key", "value"), "sourceValidation": ("key", "reportedActual", "reportedTarget", "reportedResult"), } ENUM_VALUES = { "yesNo": {True, False}, "enabled": {True, False}, "resourceKind": {"FACTORY", "WORKSHOP", "ZONE", "OPERATION"}, "resourceStatus": {"ACTIVE", "INACTIVE"}, "equipmentStatus": {"RUNNING", "DOWN", "MAINTENANCE", "DISABLED", "INACTIVE"}, "materialType": {"FINISHED_PRODUCT", "SEMI_FINISHED", "RAW_MATERIAL"}, "sourcingType": {"MAKE", "PURCHASE", "OUTSOURCE"}, "orderType": {"FORMAL", "SANDBOX"}, "orderStatus": {"RELEASED", "PENDING_EVALUATION"}, "wipStatus": {"DONE", "RUNNING", "WAITING", "MATERIAL_BLOCKED", "NOT_STARTED"}, "partnerType": {"CUSTOMER", "SUPPLIER"}, "calendarEvent": {"MAINTENANCE"}, } METADATA_KEYS = { "sourceNotes": {"dataDate", "boundary"}, "planningParameters": {"sortMode", "horizonDays", "freezeHours", "nightShiftEnabled", "comparisonMetrics"}, "sandboxScenario": {"orderNo", "productCode", "productName", "quantity", "dueDate", "priority", "customerName", "customerLevel", "scenario", "activation"}, } CAPABILITIES = {"complete-planning-workbook", "adoption-review"} def _unique_object(pairs): obj = {} for key, value in pairs: if key in obj: raise ValueError(f"工作簿配置含重复字段：{key}") obj[key] = value return obj def _strings(values: Any, label: str, *, allow_empty: bool = False) -> None: if not isinstance(values, list) or (not allow_empty and not values) or any( not isinstance(v, str) or not v.strip() or v != v.strip() for v in values ) or len(values) != len(set(values)): raise ValueError(f"工作簿配置{label}必须是无重复的非空字符串列表") def _validate_profile(profile: Any, path: Path) -> dict: allowed = {"schemaVersion", "id", "label", "compatibilityDefault", "capabilities", "recognition", "planning", "separators", "provenance", "enums", "metadataKeys", "sheets"} if not isinstance(profile, dict) or set(profile) != allowed or profile.get("schemaVersion") != 1: raise ValueError(f"工作簿配置结构或版本无效：{path.name}") if not isinstance(profile["id"], str) or not profile["id"].strip() or not isinstance(profile["label"], str): raise ValueError("工作簿配置缺少id或label") if not isinstance(profile["compatibilityDefault"], bool): raise ValueError("compatibilityDefault必须是布尔值") # noqa: TRY004 - configuration errors share a public ValueError contract _strings(profile["capabilities"], "capabilities") if set(profile["capabilities"]) != CAPABILITIES: raise ValueError("工作簿配置请求了尚未实现的能力") sheets = profile["sheets"] if not isinstance(sheets, dict) or set(sheets) != set(ROLE_FIELDS): raise ValueError("完整工作簿配置必须覆盖全部已支持的业务角色") names = [] for role, fields in ROLE_FIELDS.items(): spec = sheets[role] if not isinstance(spec, dict) or not {"name", "columns"} <= set(spec) <= {"name", "columns", "requiredColumns"}: raise ValueError(f"工作簿配置角色{role}结构无效") if not isinstance(spec["name"], str) or not spec["name"].strip(): raise ValueError(f"工作簿配置角色{role}缺工作表名称") names.append(spec["name"]) if not isinstance(spec["columns"], dict) or set(spec["columns"]) != set(fields): raise ValueError(f"工作簿配置角色{role}的标准字段不完整或不受支持") _strings(list(spec["columns"].values()), role + ".columns") # requiredColumns 是 schemaVersion 1 后续补充的合同字段：缺失按“全部可选”加载， # 显式声明时逐项校验，避免已部署的旧配置因新增字段整体无法加载。 spec.setdefault("requiredColumns", []) _strings(spec["requiredColumns"], role + ".requiredColumns", allow_empty=True) if not set(spec["requiredColumns"]) <= set(spec["columns"]): raise ValueError(f"工作簿配置角色{role}的必填列不是已声明字段") if len(names) != len(set(names)): raise ValueError("多个业务角色不能使用同一张物理工作表") recognition = profile["recognition"] if not isinstance(recognition, dict) or set(recognition) != {"anchorRoles", "minimumAnchors", "minimumRoles"}: raise ValueError("工作簿配置recognition无效") _strings(recognition["anchorRoles"], "anchorRoles") if not set(recognition["anchorRoles"]) <= set(ROLE_FIELDS): raise ValueError("识别锚点包含未知业务角色") for key, maximum in (("minimumAnchors", len(recognition["anchorRoles"])), ("minimumRoles", len(ROLE_FIELDS))): if type(recognition[key]) is not int or not 1 <= recognition[key] <= maximum: raise ValueError(f"识别阈值{key}无效") planning = profile["planning"] if not isinstance(planning, dict) or set(planning) != {"timeZone", "skillLevelOrder", "trialOnly", "sortModeAliases"}: raise ValueError("工作簿配置planning无效") try: ZoneInfo(planning["timeZone"]) except (ZoneInfoNotFoundError, TypeError, ValueError) as exc: raise ValueError("工作簿配置时区无效") from exc _strings(planning["skillLevelOrder"], "skillLevelOrder") if planning["trialOnly"] is not True: raise ValueError("工作簿配置不能绕过试排和现场确认边界") modes = planning["sortModeAliases"] if not isinstance(modes, dict) or not modes or any(v not in {"ASC", "BOTTLENECK"} for v in modes.values()): raise ValueError("工作簿配置排序模式映射无效") enums = profile["enums"] if not isinstance(enums, dict) or set(enums) != set(ENUM_VALUES): raise ValueError("工作簿配置枚举类型不完整") for name, targets in ENUM_VALUES.items(): aliases = enums[name] if not isinstance(aliases, dict) or not aliases or any( not isinstance(k, str) or not k.strip() or type(v) not in (bool, str) or v not in targets for k, v in aliases.items() ): raise ValueError(f"工作簿配置枚举{name}含未知标准值") metadata = profile["metadataKeys"] if not isinstance(metadata, dict) or set(metadata) != set(METADATA_KEYS): raise ValueError("工作簿配置元数据角色不完整") for role, keys in METADATA_KEYS.items(): aliases = metadata[role] if not isinstance(aliases, dict) or set(aliases.values()) != keys or len(aliases) != len(keys): raise ValueError(f"工作簿配置元数据{role}字段缺失或有歧义") _strings(list(aliases), "metadataKeys." + role) if not isinstance(profile["separators"], dict) or set(profile["separators"]) != {"list", "interval"}: raise ValueError("工作簿配置分隔符无效") _strings(list(profile["separators"].values()), "separators") provenance = profile["provenance"] if not isinstance(provenance, dict) or set(provenance) != {"demoMarkers", "inventorySource", "defaultTimeSource", "bottleneckMarkers"}: raise ValueError("工作簿配置来源规则无效") _strings(provenance["demoMarkers"], "demoMarkers", allow_empty=True) _strings(provenance["bottleneckMarkers"], "bottleneckMarkers", allow_empty=True) if provenance["inventorySource"] not in {"demo", "imported_unconfirmed"} or provenance["defaultTimeSource"] != "imported_unconfirmed": raise ValueError("工作簿配置不能将导入资料标成现场已确认") digest = hashlib.sha256(json.dumps(profile, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest() return {**profile, "profileDigest": digest} def load_profiles(directory: str | Path | None = None) -> dict[str, dict]: configured = directory if directory is not None else os.environ.get("APS_WORKBOOK_PROFILE_DIR") root = Path(configured) if configured is not None else BUILTIN_PROFILE_DIR if not root.is_dir(): raise ValueError(f"工作簿配置目录不存在：{root}") profiles = {} for file in sorted(root.glob("*.json")): if not file.resolve().is_relative_to(root.resolve()) or file.stat().st_size > 1024 * 1024: raise ValueError("工作簿配置文件越界或过大") try: spec = json.loads(file.read_text(encoding="utf-8-sig"), object_pairs_hook=_unique_object) profile = _validate_profile(spec, file) except (json.JSONDecodeError, UnicodeError) as exc: raise ValueError(f"无法解析工作簿配置：{file.name}") from exc if profile["id"] in profiles: raise ValueError(f"工作簿配置id重复：{profile['id']}") profiles[profile["id"]] = profile if not profiles: raise ValueError("工作簿配置目录没有有效的JSON配置") return profiles def get_profile(profile_id: str) -> dict: profile = load_profiles().get(profile_id) if profile is None: raise ValueError(f"未注册工作簿格式：{profile_id}") return profile def is_profile_supported(profile_id: str | None) -> bool: return bool(profile_id and profile_id in load_profiles()) def has_adoption_flow(profile_id: str | None) -> bool: return bool(profile_id and "adoption-review" in load_profiles().get(profile_id, {}).get("capabilities", [])) def normalize_resource_kind(value: Any, profile_id: str | None) -> str | None: if value in ENUM_VALUES["resourceKind"]: return value profile = load_profiles().get(profile_id) return profile["enums"]["resourceKind"].get(str(value)) if profile else None def builtin_compatibility_profile() -> dict: matches = [p for p in load_profiles(BUILTIN_PROFILE_DIR).values() if p["compatibilityDefault"]] if len(matches) != 1: raise ValueError("内置兼容工作簿配置必须唯一") return matches[0] def active_compatibility_profile() -> dict: """无 profileId 的调用（模板下载、合同诊断）也必须跟随部署配置目录。 唯一标记 ``compatibilityDefault`` 的配置优先；目录里只注册了一份配置时， 该配置即为默认。多份配置但默认标记缺失或重复时不猜测，直接报错。 """ profiles = load_profiles() defaults = [profile for profile in profiles.values() if profile["compatibilityDefault"]] if len(defaults) == 1: return defaults[0] if not defaults and len(profiles) == 1: return next(iter(profiles.values())) raise ValueError("工作簿配置目录必须恰好声明一份兼容默认配置（compatibilityDefault）") def select_profile(sheet_headers: dict[str, list[str]]) -> dict | None: candidates = [] for profile in load_profiles().values(): sheets = profile["sheets"] matched_roles = {role for role, spec in sheets.items() if spec["name"] in sheet_headers} recognition = profile["recognition"] if len(matched_roles & set(recognition["anchorRoles"])) >= recognition["minimumAnchors"] \ or len(matched_roles) >= recognition["minimumRoles"]: candidates.append(profile) if len(candidates) > 1: # Identical physical sheets can still be disambiguated by complete # column contracts; equally plausible or partial matches fail closed. complete = [p for p in candidates if all(spec["name"] in sheet_headers and set(spec["columns"].values()) <= set(sheet_headers[spec["name"]]) for spec in p["sheets"].values())] if len(complete) == 1: return complete[0] raise ValueError("工作簿同时匹配多个配置，请明确配置范围后重新核对") return candidates[0] if candidates else None