"""Stable, project-local master objects and explicit flexible-engine projections. Imports own supplied fields until a planner edits that field. Canonical IDs survive repeated imports; missing fields never clear existing values. Solver results and released snapshots are deliberately outside this module's write set. """ from __future__ import annotations import copy from datetime import date, timedelta from functools import wraps from typing import Any World = dict[str, Any] MATERIAL_FIELDS = ("code", "name", "spec", "type", "unit", "stock", "inTransit", "expectedArrivalDate", "safetyStock", "procurementLeadTime", "productFamily", "status", "sourcingType", "sourcingTypeSource") EQUIPMENT_FIELDS = ("code", "name", "model", "status", "capacityPerHour", "efficiencyFactor", "availabilityRate", "operations", "zoneCode") PROVENANCE_FIELDS = ("sourceRef", "sourceProfile", "sourceText", "stdTimeSource", "sourceNote", "dataOrigin", "requiresConfirmation", "sourceValues", "provenance", "inventorySourceRef") ID_TABLES = {"material": "materials", "bom": "boms", "bomItem": "bomItems", "routing": "routings", "routingStep": "routingSteps", "operation": "operations", "equipment": "equipment", "workstation": "workstations", "maintenance": "maintenance", "shiftCalendar": "shiftCalendar", "calendarTemplate": "calendarTemplates", "calendarHoliday": "calendarHolidays"} def _next(world: World, table: str) -> int: return max((r.get("id", 0) for r in world.get(table, []) if isinstance(r.get("id"), int)), default=0) + 1 def _record(world: World, table: str, key: str, value: Any, defaults: dict[str, Any]) -> tuple[dict[str, Any], bool]: rows = world.setdefault(table, []) existing = next((r for r in rows if r.get(key) == value), None) if existing is not None: return existing, False row = {"id": _next(world, table), **copy.deepcopy(defaults), key: value} rows.append(row) return row, True def _merge_import(target: dict, incoming: dict, mapping: dict[str, str], *, refresh: bool, source: str) -> None: overrides = set(target.get("masterOverrides", [])) for dest, src in mapping.items(): if dest in overrides and dest in target: incoming[src] = copy.deepcopy(target[dest]) elif src in incoming and (refresh or dest not in target): target[dest] = copy.deepcopy(incoming[src]) for field in PROVENANCE_FIELDS: if field in incoming and (refresh or field not in target): target[field] = copy.deepcopy(incoming[field]) target.setdefault("masterSource", source) def _ensure_material(world: World, code: str, *, name: str = "", kind: str = "RAW_MATERIAL") -> dict: row, _ = _record(world, "materials", "code", code, { "name": name or code, "type": kind, "unit": "件", "spec": "", "stock": 0, "inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0, "status": "ACTIVE", "masterPlaceholder": True, }) return row def _head(world: World, table: str, product: dict, source: str) -> dict: # Adopt the existing default during migration. Never delete another version. rows = world.setdefault(table, []) head = next((r for r in rows if r.get("productId") == product["id"] and r.get("isDefault", False) and r.get("status", "ACTIVE") != "ARCHIVED"), None) if head is None: head = {"id": _next(world, table), "productId": product["id"], "version": "IMPORT-V1", "versionName": "导入资料", "isDefault": True, "status": "ACTIVE"} rows.append(head) head.setdefault("masterSource", source) head.setdefault("masterSourceKey", product["code"]) return head def reconcile_imported_masterdata(world: World, *, source: str = "flex-import", refresh: bool = True) -> dict[str, int]: """Materialize stable editable objects; call once after an approved import. refresh=False hydrates old flexible-only projects without overwriting existing canonical values. Read endpoints must use a copy when hydrating this way. """ summary: dict[str, int] = {} for incoming in world.get("flexMaterials", []): code = str(incoming.get("code") or "").strip() if not code: continue existed = any(m.get("code") == code for m in world.get("materials", [])) row = _ensure_material(world, code, name=incoming.get("name", code), kind=incoming.get("type", "FINISHED_PRODUCT")) _merge_import(row, incoming, {k: k for k in MATERIAL_FIELDS}, refresh=refresh or not existed, source=source) row.pop("masterPlaceholder", None) row["masterSourceKey"] = code incoming["masterId"] = row["id"] summary["materials"] = summary.get("materials", 0) + 1 operation_rows = list(world.get("flexOperations", [])) operation_rows += [{"code": r.get("operationCode"), "name": r.get("operationName"), "standardTime": r.get("stdTimePerUnit", 0), "isExternal": r.get("isExternal", False), "sourcingType": r.get("sourcingType", "")} for r in world.get("flexRoutings", [])] for incoming in operation_rows: code = str(incoming.get("code") or "") if not code: continue from server.aps_domain.sourcing import infer_op_sourcing sourcing = incoming.get("sourcingType") or infer_op_sourcing(code, incoming.get("name")) row, created = _record(world, "operations", "code", code, { "name": incoming.get("name") or code, "standardTime": incoming.get("standardTime", 0), "type": "EXTERNAL" if incoming.get("isExternal") or sourcing == "OUTSOURCE" else "INTERNAL", "sourcingType": sourcing, "masterSource": source, }) summary["operations"] = summary.get("operations", 0) + int(created) operations = {o["code"]: o for o in world.get("operations", [])} for incoming in world.get("flexRoutings", []): pc, oc = str(incoming.get("productCode") or ""), str(incoming.get("operationCode") or "") if not pc or oc not in operations: continue product = _ensure_material(world, pc, name=incoming.get("productName", pc), kind="FINISHED_PRODUCT") head = _head(world, "routings", product, source) seq = int(incoming.get("seq") or 1) key = f"{pc}:{seq}:{oc}" rows = world.setdefault("routingSteps", []) step = next((r for r in rows if r.get("routingId") == head["id"] and (r.get("masterSourceKey") == key or (r.get("sequenceNo") == seq and r.get("operationId") == operations[oc]["id"]))), None) created = step is None if step is None: step = {"id": _next(world, "routingSteps"), "routingId": head["id"], "operationId": operations[oc]["id"], "sequenceNo": seq, "prevStepId": None, "setupTime": 0, "runTimePerUnit": 0, "waitTime": 0, "transferTime": 0, "isExternal": operations[oc].get("type") == "EXTERNAL"} rows.append(step) _merge_import(step, incoming, {"setupTime": "setupTime", "runTimePerUnit": "stdTimePerUnit", "isExternal": "isExternal", "sourcingType": "sourcingType"}, refresh=refresh or created, source=source) step["masterSourceKey"] = key incoming["masterStepId"] = step["id"] incoming["masterRoutingId"] = head["id"] summary["routingSteps"] = summary.get("routingSteps", 0) + 1 summary["externalSteps"] = summary.get("externalSteps", 0) + int(step.get("isExternal", False)) for head in world.get("routings", []): if not head.get("masterSource"): continue prior = None for step in sorted((s for s in world.get("routingSteps", []) if s["routingId"] == head["id"]), key=lambda s: s["sequenceNo"]): step["prevStepId"] = prior prior = step["id"] summary["routings"] = len({r.get("masterRoutingId") for r in world.get("flexRoutings", []) if r.get("masterRoutingId") is not None}) for incoming in world.get("flexBom", []): pc, mc = str(incoming.get("productCode") or ""), str(incoming.get("materialCode") or "") if not pc or not mc: continue product = _ensure_material(world, pc, kind="FINISHED_PRODUCT") material = _ensure_material(world, mc, name=incoming.get("materialName", mc)) head = _head(world, "boms", product, source) rows = world.setdefault("bomItems", []) item = next((r for r in rows if r.get("bomId") == head["id"] and r.get("materialId") == material["id"]), None) created = item is None if item is None: item = {"id": _next(world, "bomItems"), "bomId": head["id"], "materialId": material["id"], "quantity": 0, "isKeyMaterial": False, "operationId": None} rows.append(item) _merge_import(item, incoming, {"quantity": "quantity", "isKeyMaterial": "isKey", "lossRate": "lossRate", "unit": "unit"}, refresh=refresh or created, source=source) item["masterSourceKey"] = f"{pc}:{mc}" incoming["masterItemId"], incoming["masterBomId"] = item["id"], head["id"] summary["bomItems"] = summary.get("bomItems", 0) + 1 summary["boms"] = len({r.get("masterBomId") for r in world.get("flexBom", []) if r.get("masterBomId") is not None}) for incoming in world.get("flexCalendar", []): code = str(incoming.get("shiftCode") or incoming.get("code") or "") if not code: continue shift, created = _record(world, "shifts", "code", code, { "name": incoming.get("name") or code, "startTime": incoming.get("startTime", ""), "endTime": incoming.get("endTime", ""), "enabled": incoming.get("enabled", True)}) _merge_import(shift, incoming, {"name": "name", "startTime": "startTime", "endTime": "endTime", "breakPeriods": "breaks", "enabled": "enabled", "workdays": "workdays"}, refresh=refresh or created, source=source) incoming["masterId"] = shift["id"] _import_resources(world, source=source, refresh=refresh) return summary def _import_resources(world: World, *, source: str, refresh: bool) -> None: from server.importers.workbook_profiles import normalize_resource_kind if not world.get("flexEquipment"): return source_resources = world.get("flexFactoryResources", []) context = world.get("planningContext") or {} profile_id = context.get("sourceProfile") timezone = context.get("timeZone") source_factories = [r for r in source_resources if (r.get("resourceKind") or normalize_resource_kind(r.get("resourceType"), profile_id)) == "FACTORY"] for incoming in source_factories: factory, created = _record(world, "factories", "code", incoming["code"], { "name": incoming.get("name") or incoming["code"], "timezone": incoming.get("timeZone") or timezone, "status": "ACTIVE"}) _merge_import(factory, incoming, {"name": "name", "status": "status"}, refresh=refresh or created, source=source) factories = {r["code"]: r for r in world.get("factories", [])} for incoming in source_resources: resource_kind = incoming.get("resourceKind") or normalize_resource_kind(incoming.get("resourceType"), profile_id) if resource_kind != "WORKSHOP" or incoming.get("parentCode") not in factories: continue workshop, created = _record(world, "workshops", "code", incoming["code"], { "factoryId": factories[incoming["parentCode"]]["id"], "name": incoming.get("name") or incoming["code"], "status": "ACTIVE"}) _merge_import(workshop, incoming, {"name": "name", "status": "status"}, refresh=refresh or created, source=source) source_workshops = {r["code"]: r for r in world.get("workshops", [])} fallback_factory = next(iter(factories.values()), None) if fallback_factory is None: fallback_factory, _ = _record(world, "factories", "code", "IMPORT-SITE", { "name": "导入资料所属工厂（待确认）", "timezone": timezone, "status": "ACTIVE", "masterSource": source, "mappingPending": True}) for incoming in world["flexEquipment"]: code = str(incoming.get("code") or "") if not code: continue existing = next((r for r in world.get("equipment", []) if r.get("code") == code), None) old_station = next((w for w in world.get("workstations", []) if existing is not None and w.get("id") == existing.get("workstationId")), None) owned_station = bool(old_station and old_station.get("masterSource") and old_station.get("code") == f"IMPORT:{code}") if existing is not None and old_station is not None and not owned_station: _merge_import(existing, incoming, {k: k for k in EQUIPMENT_FIELDS}, refresh=refresh, source=source) existing["masterSourceKey"] = code incoming["masterId"] = existing["id"] continue zone = str(incoming.get("zoneCode") or incoming.get("zone") or "UNASSIGNED") zone_info = next((z for z in source_resources if z.get("code") == zone), next((z for z in world.get("flexZones", []) if z.get("code") == zone), {})) workshop = source_workshops.get(zone_info.get("parentCode")) if workshop is None: parent_factory = factories.get(zone_info.get("parentCode"), fallback_factory) workshop, _ = _record(world, "workshops", "code", f"IMPORT-AREAS:{parent_factory['code']}", { "factoryId": parent_factory["id"], "name": "生产区域（车间关联待确认）", "status": "ACTIVE", "masterSource": source, "mappingPending": True}) line, _ = _record(world, "lines", "code", f"IMPORT:{zone}", { "workshopId": workshop["id"], "name": zone_info.get("name") or zone, "capacityPerDay": 0, "efficiencyFactor": 1.0, "status": "ACTIVE", "masterSource": source}) station, _ = _record(world, "workstations", "code", f"IMPORT:{code}", { "lineId": line["id"], "name": incoming.get("name") or code, "sequenceNo": 1, "status": "ACTIVE", "masterSource": source}) if owned_station and "lineId" not in station.get("masterOverrides", []): station["lineId"] = line["id"] row, created = _record(world, "equipment", "code", code, { "workstationId": station["id"], "name": incoming.get("name") or code, "status": "ACTIVE", "model": "", "capacityPerHour": 0, "efficiencyFactor": 1.0, "availabilityRate": 1.0}) _merge_import(row, incoming, {k: k for k in EQUIPMENT_FIELDS}, refresh=refresh or created, source=source) row["masterSourceKey"] = code incoming["masterId"] = row["id"] eq_by_code = {e["code"]: e for e in world.get("equipment", [])} for incoming in world.get("flexMaintenance", []): eq = eq_by_code.get(incoming.get("equipmentCode")) if eq is None: continue key = incoming.get("code") or f"{eq['code']}:{incoming.get('start')}:{incoming.get('end')}" row, created = _record(world, "maintenance", "masterSourceKey", key, { "equipmentId": eq["id"], "type": "MAINTENANCE", "status": "PLANNED"}) _merge_import(row, incoming, {"plannedStart": "start", "plannedEnd": "end", "description": "reason", "status": "status"}, refresh=refresh or created, source=source) incoming["masterId"] = row["id"] def masterdata_view_world(world: World) -> World: """Read-only migration projection; IDs are identical to the approved write path.""" projected = copy.deepcopy(world) reconcile_imported_masterdata(projected, refresh=False) return projected def _changed(before: World, after: World, table: str) -> list[dict]: old = {r["id"]: r for r in before.get(table, [])} return [r for r in after.get(table, []) if r != old.get(r["id"])] def _mark_override(row: dict, old: dict, fields: tuple[str, ...]) -> None: changed = {k for k in fields if k in row and row.get(k) != old.get(k)} if changed: row["masterOverrides"] = sorted(set(row.get("masterOverrides", [])) | changed) def _project_product(world: World, product_id: int, *, routing: bool) -> None: product = next((m for m in world.get("materials", []) if m["id"] == product_id), None) if product is None: return heads_key, detail_key, target_key, parent_key = ("routings", "routingSteps", "flexRoutings", "routingId") if routing else ("boms", "bomItems", "flexBom", "bomId") head = next((h for h in world.get(heads_key, []) if h.get("productId") == product_id and h.get("isDefault") and h.get("status", "ACTIVE") != "ARCHIVED"), None) if head is None: return old_rows = [r for r in world.get(target_key, []) if r.get("productCode") == product["code"]] if not old_rows and not head.get("masterSource") and not any(m.get("code") == product["code"] for m in world.get("flexMaterials", [])): return output = [] for detail in world.get(detail_key, []): if detail[parent_key] != head["id"]: continue if routing: op = next(o for o in world["operations"] if o["id"] == detail["operationId"]) old = next((r for r in old_rows if r.get("masterStepId") == detail["id"] or (r.get("operationCode") == op["code"] and r.get("seq") == detail["sequenceNo"])), {}) row = copy.deepcopy(old) row.update({"productCode": product["code"], "operationCode": op["code"], "operationName": op["name"], "seq": detail["sequenceNo"], "setupTime": detail["setupTime"], "stdTimePerUnit": detail["runTimePerUnit"], "isExternal": detail.get("isExternal", False), "masterStepId": detail["id"], "masterRoutingId": head["id"]}) else: material = next(m for m in world["materials"] if m["id"] == detail["materialId"]) old = next((r for r in old_rows if r.get("materialCode") == material["code"]), {}) row = copy.deepcopy(old) row.update({"productCode": product["code"], "materialCode": material["code"], "quantity": detail["quantity"], "isKey": detail.get("isKeyMaterial", False), "masterItemId": detail["id"], "masterBomId": head["id"]}) if "lossRate" in detail: row["lossRate"] = detail["lossRate"] for field in PROVENANCE_FIELDS: if field in detail: row[field] = copy.deepcopy(detail[field]) row["masterOverrides"] = list(detail.get("masterOverrides", [])) output.append(row) world[target_key] = [r for r in world.get(target_key, []) if r.get("productCode") != product["code"]] + output def _project_calendar(world: World, payload: dict, result: dict) -> None: selected_lines = set(result["lineIds"]) stations = {w["id"] for w in world.get("workstations", []) if w.get("lineId") in selected_lines} devices = [e for e in world.get("equipment", []) if e.get("workstationId") in stations] shifts = {s["id"]: s for s in world.get("shifts", [])} overrides = world.setdefault("flexCalendarOverrides", []) cursor, end = date.fromisoformat(result["startDate"]), date.fromisoformat(result["endDate"]) while cursor <= end: day = cursor.isoformat() for eq in devices: station = next(w for w in world["workstations"] if w["id"] == eq["workstationId"]) hours = [] for cal in world.get("shiftCalendar", []): if cal.get("date") != day or cal.get("lineId") != station["lineId"] or not cal.get("isWorking"): continue shift = shifts.get(cal.get("shiftId")) if shift: hours.append({"start": shift["startTime"], "end": shift["endTime"], "enabled": True, "shiftCode": shift.get("code") or shift.get("shiftCode"), "breaks": copy.deepcopy(shift.get("breakPeriods", [])), "shiftId": shift["id"], "teamId": cal.get("teamId"), "maxWorkers": cal.get("maxWorkers")}) row = {"equipmentCode": eq["code"], "date": day, "shifts": hours, "source": f"calendarTemplate:{payload['templateId']}"} existing = next((r for r in overrides if r.get("equipmentCode") == eq["code"] and r.get("date") == day), None) if existing is None: overrides.append(row) else: existing.update(row) cursor += timedelta(days=1) def _publish_master_candidate(target: World, candidate: World) -> None: """Publish a successful mutation, preserving live canonical row references.""" for key, value in candidate.items(): if target.get(key) == value: continue if key in ID_TABLES.values() and isinstance(value, list) and isinstance(target.get(key), list): old = {r.get("id"): r for r in target[key]} published = [] for row in value: prior = old.get(row.get("id")) if prior is not None: prior.clear() prior.update(row) published.append(prior) else: published.append(row) target[key][:] = published else: target[key] = value def synchronized_master_action(fn): """Keep legacy action API while synchronizing only the approved action's changes.""" @wraps(fn) def execute(world: World, next_id, action: str, payload: dict) -> dict: # A failed validation must not leave migration rows or half a projection. target_world = world world = copy.deepcopy(world) reconcile_imported_masterdata(world, refresh=False) before = {k: copy.deepcopy(world.get(k, [])) for k in ("materials", "equipment", "boms", "bomItems", "routings", "routingSteps", "maintenance")} def safe_next(kind: str) -> int: table = ID_TABLES.get(kind, kind + "s") maximum = _next(world, table) - 1 allocated = next_id(kind) while allocated <= maximum: allocated = next_id(kind) return allocated result = fn(world, safe_next, action, payload) if action == "master.clear": _publish_master_candidate(target_world, world) return result for table, flex_key, fields in (("materials", "flexMaterials", MATERIAL_FIELDS), ("equipment", "flexEquipment", EQUIPMENT_FIELDS)): prior = {r["id"]: r for r in before[table]} for row in _changed(before, world, table): old = prior.get(row["id"], {}) linked = next((r for r in world.get(flex_key, []) if r.get("masterId") == row["id"] or r.get("code") == old.get("code", row.get("code"))), None) if linked is None: continue _mark_override(row, old, fields) for field in fields: if field in row and (field in payload or field in row.get("masterOverrides", [])): linked[field] = copy.deepcopy(row[field]) linked["masterId"] = row["id"] if action == "master.equipment.upsert": present = {r["id"] for r in world.get("equipment", [])} for old in before["equipment"]: if old["id"] not in present: for eq in world.get("flexEquipment", []): if eq.get("code") == old["code"]: eq["status"] = "INACTIVE" for head_key, detail_key, parent_key, fields, routing in ( ("routings", "routingSteps", "routingId", ("setupTime", "runTimePerUnit", "isExternal"), True), ("boms", "bomItems", "bomId", ("quantity", "isKeyMaterial", "lossRate"), False), ): old_details = {r["id"]: r for r in before[detail_key]} changed_heads = {r["id"] for r in _changed(before, world, head_key)} for row in _changed(before, world, detail_key): _mark_override(row, old_details.get(row["id"], {}), fields) changed_heads.add(row[parent_key]) current_ids = {r["id"] for r in world.get(detail_key, [])} changed_heads |= {r[parent_key] for r in before[detail_key] if r["id"] not in current_ids} for pid in {r["productId"] for r in world.get(head_key, []) if r["id"] in changed_heads}: _project_product(world, pid, routing=routing) if action == "master.calendar.week.copy": _project_calendar(world, payload, result) for row in _changed(before, world, "maintenance"): eq = next((e for e in world.get("equipment", []) if e["id"] == row["equipmentId"]), None) if eq and any(e.get("code") == eq["code"] for e in world.get("flexEquipment", [])): linked = next((r for r in world.get("flexMaintenance", []) if r.get("masterId") == row["id"]), None) data = {"masterId": row["id"], "equipmentCode": eq["code"], "start": row["plannedStart"], "end": row["plannedEnd"], "status": row["status"], "reason": row.get("description", "")} if linked is None: world.setdefault("flexMaintenance", []).append(data) else: linked.update(data) _publish_master_candidate(target_world, world) return result return execute