"""Portable House Plan backup/export and two-phase import service. The browser never interprets an import document. It uploads bytes for a bounded server-side preview, receives a short-lived opaque token and can apply only the exact candidate that produced that preview. """ from __future__ import annotations import hashlib import json import math import re import secrets import time import unicodedata from datetime import UTC, datetime from pathlib import Path from typing import Any, Callable from urllib.parse import urlsplit import voluptuous as vol from .const import ( ASSETS_DIR, CONTENT_URL, EXPORT_VERSION, FILES_DIR, FILES_URL, IMPORT_PREVIEW_TTL_S, MAX_EXPORT_BYTES, MAX_IMPORT_PREVIEWS_PER_USER, MAX_IMPORT_PREVIEWS_TOTAL, PLAN_MODEL_VERSION, PLANS_DIR, PLANS_URL, VERSION, ) from .radar_validation import validate_marker_radars from .store import HouseplanData from .validation import ( CONFIG_SCHEMA, LAYOUT_SCHEMA, MAX_CONFIG_BYTES, MAX_CONTROLS, MAX_LAYOUT, MAX_MARKERS, MAX_SPACES, DuplicateMarkerIdError, MarkerControlError, OpeningPassageError, PartitionOpeningHostError, PartitionOpeningJambMarginError, sanitize_filename, sanitize_marker_id, validate_active_marker_ids, validate_marker_controls, validate_marker_light_entities, validate_marker_vacuum_routes, validate_marker_value_badges, validate_opening_passages, validate_partition_opening_hosts, ) from .wall_segment_model import ( WallSegmentMigrationError, commit_wall_segment_model, ) FORMAT = "houseplan-export" _PROTO_KEYS = {"__proto__", "prototype", "constructor"} _SAFE_FILE = re.compile(r"[^A-Za-z0-9._-]+") _LIVE_TEXT_TOKEN = re.compile(r"\{([^{}\r\n]+)\}") _LIVE_TEXT_ENTITY = re.compile(r"^[a-z0-9_]+\.[a-z0-9_]+$") _LIVE_TEXT_ATTRIBUTE = re.compile(r"^[a-zA-Z0-9_.-]+$") _PLAN_ONLY_DASH = "—" _IMPORT_ID_NAMESPACES = { "space", "room", "marker", "partition", "wall", "opening", "decor", "column", "stair", } _MAX_IMPORT_LINEAGE_DEPTH = 16 _REPORT_EXAMPLE_LIMIT = 24 _DECOR_ASSET_MIME_BY_SUFFIX = { ".png": "image/png", ".jpg": "image/jpeg", ".webp": "image/webp", ".svg": "image/svg+xml", } _DECOR_ASSET_MIME_TYPES = frozenset(_DECOR_ASSET_MIME_BY_SUFFIX.values()) _SPACE_PLAN_FIELDS = ( "id", "title", "cell_cm", "plan_url", "plan_aspect", "plan_x", "plan_y", "plan_scale", "plan_scale_x", "plan_scale_y", "plan_angle", "view_box", "zero_wall_style", ) _SPACE_DISPLAY_FIELDS = ( "show_borders", "show_names", "room_color", "bg_color", "room_opacity", "fill_mode", "custom_fill", "glow_enabled", "temp_min", "temp_max", "show_lqi", "hide_decor", "hide_openings", "label_temp", "label_hum", "label_lqi", "label_light", "card_font_scale", "north_deg", "bg_mode", "sun_rays", ) _ROOM_PLAN_FIELDS = ("id", "name", "open_to", "x", "y", "w", "h", "poly", "wall_ids") _ROOM_DISPLAY_FIELDS = ( "fill_mode", "custom_fill", "glow", "temp_min", "temp_max", "name_scale", "label_scale", ) _DECOR_COMMON_FIELDS = ("id", "kind", "color", "opacity", "width_cm", "width") _DECOR_KIND_FIELDS = { "line": ("x1", "y1", "x2", "y2", "line_style"), "rect": ("x", "y", "w", "h", "angle", "fill", "fill_color", "fill_opacity"), "ellipse": ("x", "y", "w", "h", "angle", "fill", "fill_color", "fill_opacity"), "text": ("x", "y", "text", "size", "size_cm", "scale", "angle"), "furniture": ("symbol", "x", "y", "w", "h", "angle", "flip_h", "flip_v"), "image": ("asset_id", "x", "y", "w", "h", "angle", "flip_h", "flip_v"), } class ImportFailure(Exception): """Stable public import error.""" def __init__(self, code: str, message: str = "") -> None: super().__init__(message or code) self.code = code self.message = message or code def source_fingerprint(instance_id: str) -> str: raw = "houseplan-export-v1|" + instance_id return "sha256:" + hashlib.sha256(raw.encode()).hexdigest() def _json_copy(value: Any) -> Any: return json.loads(json.dumps(value, ensure_ascii=False, allow_nan=False)) def _background_mode(settings: Any) -> str | None: if not isinstance(settings, dict): return None mode = settings.get("bg_mode") return mode if mode in ("static", "daynight") else None def _materialize_global_background(config: dict[str, Any], fallback: str = "static") -> str: """Make the global mode portable instead of relying on a code default.""" settings = config.get("settings") if not isinstance(settings, dict): settings = {} config["settings"] = settings mode = _background_mode(settings) or fallback settings["bg_mode"] = mode return mode def _materialize_space_background(space: dict[str, Any], fallback: str) -> str: """Make one space independent from the target installation's default.""" settings = space.get("settings") if not isinstance(settings, dict): settings = {} space["settings"] = settings mode = _background_mode(settings) or fallback settings["bg_mode"] = mode return mode def _stored_model_version(config: dict[str, Any]) -> int: """Return the model actually stored without silently upgrading it.""" value = config.get("model_version", 0) if value is None: return 0 if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise ImportFailure("invalid_config", "Invalid configuration model version") return value def _document_digest(document: dict[str, Any]) -> str: """Digest the normalized candidate a preview token is allowed to apply.""" raw = json.dumps( document, ensure_ascii=False, separators=(",", ":"), sort_keys=True, allow_nan=False, ).encode("utf-8") return hashlib.sha256(raw).hexdigest() def live_layout(config: dict[str, Any], layout: dict[str, Any]) -> dict[str, Any]: """Return only positions that still have a possible live owner.""" markers = config.get("markers") or [] removed = {str(m.get("id")) for m in markers if m.get("removed") is True} explicit = { str(m.get("id")) for m in markers if m.get("removed") is not True } return { key: dict(pos) for key, pos in layout.items() if key not in removed and (not key.startswith("v_") or key in explicit) } def _pick_fields(source: dict[str, Any], fields: tuple[str, ...]) -> dict[str, Any]: """Copy only fields explicitly classified as portable plan data.""" return {key: _json_copy(source[key]) for key in fields if key in source} def _is_live_text_reference(raw: str) -> bool: """Mirror ``liveTextReference`` without evaluating Home Assistant state.""" ref = raw.strip() if not ref: return False entity = ref attribute = "" colon = ref.find(":") if colon >= 0: entity = ref[:colon].strip() attribute = ref[colon + 1:].strip() else: parts = ref.split(".") if len(parts) > 2: entity = ".".join(parts[:2]) attribute = ".".join(parts[2:]) if _LIVE_TEXT_ENTITY.fullmatch(entity) is None: return False if colon >= 0 and not attribute: return False return not attribute or _LIVE_TEXT_ATTRIBUTE.fullmatch(attribute) is not None def _plan_only_text(value: str) -> str: """Freeze every recognized live reference while preserving authored copy.""" replaced = _LIVE_TEXT_TOKEN.sub( lambda match: _PLAN_ONLY_DASH if _is_live_text_reference(match.group(1)) else match.group(0), value, ) return replaced.replace("{}", _PLAN_ONLY_DASH) def _project_plan_only_room(room: dict[str, Any]) -> dict[str, Any]: projected = _pick_fields(room, _ROOM_PLAN_FIELDS) if "settings" in room: settings = room.get("settings") projected["settings"] = ( _pick_fields(settings, _ROOM_DISPLAY_FIELDS) if isinstance(settings, dict) else None ) return projected def _project_plan_only_decor(shape: dict[str, Any]) -> dict[str, Any]: kind = str(shape.get("kind", "")) projected = _pick_fields( shape, _DECOR_COMMON_FIELDS + _DECOR_KIND_FIELDS.get(kind, ()), ) if kind == "text" and isinstance(projected.get("text"), str): projected["text"] = _plan_only_text(projected["text"]) return projected def _project_plan_only_space(space: dict[str, Any]) -> dict[str, Any]: """Build the fail-closed geometry/presentation projection for #167.""" projected = _pick_fields(space, _SPACE_PLAN_FIELDS) if "settings" in space: projected["settings"] = _pick_fields( space.get("settings") or {}, _SPACE_DISPLAY_FIELDS, ) projected["rooms"] = [ _project_plan_only_room(room) for room in space.get("rooms") or [] ] collections: tuple[tuple[str, tuple[str, ...]], ...] = ( ("walls", ("key", "cm", "a", "b")), ("wall_segments", ("id", "a", "b", "cm")), ("partitions", ("id", "a", "b", "cm")), ("wall_columns", ("id", "shape", "center", "cm", "angle")), ("open_spans", ("a", "b")), ) for name, fields in collections: if name not in space: continue values = [] for item in space.get(name) or []: selected = _pick_fields(item, fields) values.append(selected) projected[name] = values if "openings" in space: projected["openings"] = [ _pick_fields( opening, ("id", "type", "x", "y", "angle", "length") + (() if opening.get("type") == "passage" else ("flip_h", "flip_v")) + (("host",) if opening.get("host") else ()), ) for opening in space.get("openings") or [] ] if "decor" in space: projected["decor"] = [ _project_plan_only_decor(shape) for shape in space.get("decor") or [] ] if "stairs" in space: projected["stairs"] = [ _pick_fields(stair, ( "id", "kind", "x", "y", "angle", "direction", "target_space_id", "length", "width", "radius", )) for stair in space.get("stairs") or [] ] return projected def _plan_only_room_label_layout( layout: dict[str, Any], space: dict[str, Any], ) -> dict[str, Any]: space_id = str(space.get("id", "")) room_ids = {str(room.get("id", "")) for room in space.get("rooms") or []} projected: dict[str, Any] = {} for key, pos in layout.items(): if not ( isinstance(key, str) and key.startswith("rl_") and key[3:] in room_ids and isinstance(pos, dict) and str(pos.get("s", "")) == space_id ): continue value = _pick_fields(pos, ("x", "y", "s")) scale = pos.get("k") if isinstance(scale, (int, float)) and not isinstance(scale, bool) \ and math.isfinite(scale) and 0.5 <= scale <= 3: value["k"] = scale projected[key] = value return projected def _marker_owned(marker: dict[str, Any], space: dict[str, Any], layout: dict[str, Any]) -> bool: marker_id = str(marker.get("id", "")) room_ids = {str(room.get("id")) for room in space.get("rooms") or []} pos = layout.get(marker_id) return bool( (isinstance(pos, dict) and str(pos.get("s")) == str(space.get("id"))) or str(marker.get("space") or "") == str(space.get("id")) or str(marker.get("room_id") or "") in room_ids ) def placement_manifest(config: dict[str, Any], layout: dict[str, Any]) -> list[dict[str, Any]]: markers = {str(m.get("id")): m for m in config.get("markers") or []} room_ids = { str(r.get("id")) for sp in config.get("spaces") or [] for r in sp.get("rooms") or [] } out: list[dict[str, Any]] = [] for key, pos in layout.items(): marker = markers.get(key) if marker: kind = "marker" binding = marker.get("binding") label = marker.get("name") icon = marker.get("icon") owner_id = key elif key.startswith("rl_") and key[3:] in room_ids: kind, binding, label, icon, owner_id = "room_label", None, None, None, key[3:] elif key.startswith("lg_"): kind, binding, label, icon, owner_id = "light_group", "entity:" + key[3:], None, None, key else: kind, binding, label, icon, owner_id = "auto_device", "device:" + key, None, None, key out.append({ "layout_id": key, "space_id": pos.get("s") if isinstance(pos, dict) else None, "owner": kind, "owner_id": owner_id, "binding": binding, "label": label, "icon": icon, }) return out def _internal_path(root: Path, url: str) -> tuple[str, Path] | None: # A url is parsed as a url, not as a string: everything after "?" or "#" # addresses the transfer, never the file. Legacy attachments carry a # cache-buster (".../files/m1/doc.pdf?v=1783170649"), and while the string # form fed "doc.pdf?v=1783170649" to sanitize_filename the name never # matched itself — the reference read as internal-but-non-canonical and # every backup holding one refused to import (issue #225). Path segments # keep doing the guarding: dropping the query cannot widen what a segment # is allowed to be. # # Only a same-document reference may be trusted this way: with a scheme or # an authority the path belongs to another host, and taking it would let # "https://evil.example/houseplan_files/files/m1/doc.pdf" resolve onto a # local file (review CODE-REVIEW-225-r1, M1). Such a url stays external, # which is also what _looks_internal says about it. parsed = urlsplit(url) if parsed.scheme or parsed.netloc: return None url = parsed.path content_plan = CONTENT_URL + "/plans/_/" if url.startswith(content_plan) or url.startswith(PLANS_URL + "/"): prefix = content_plan if url.startswith(content_plan) else PLANS_URL + "/" raw_name = url[len(prefix):] if not raw_name or "/" in raw_name: return None name = sanitize_filename(raw_name) if name != raw_name: return None return "plan", root / PLANS_DIR / name asset_prefix = CONTENT_URL + "/assets/_/" if url.startswith(asset_prefix): raw_name = url[len(asset_prefix):] if not re.fullmatch(r"[0-9a-f]{64}\.(?:png|jpg|webp|svg)", raw_name): return None return "decor_asset", root / ASSETS_DIR / raw_name for prefix in (CONTENT_URL + "/files/", FILES_URL + "/"): if url.startswith(prefix): tail = url[len(prefix):].split("/") if len(tail) != 2: return None marker, name = sanitize_marker_id(tail[0]), sanitize_filename(tail[1]) if marker != tail[-2] or name != tail[-1]: return None return "attachment", root / FILES_DIR / marker / name return None def _looks_internal(url: str) -> bool: """Whether a URL claims one of House Plan's local namespaces.""" return any(url.startswith(prefix) for prefix in ( CONTENT_URL + "/", PLANS_URL + "/", FILES_URL + "/", )) def content_manifest(config: dict[str, Any], config_root: Path) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] def add(kind: str, owner: str, owner_id: str, field: str, value: Any) -> None: if not isinstance(value, str) or not value: return internal = _internal_path(config_root, value) out.append({ "kind": kind, "owner": owner, "owner_id": owner_id, "field": field, "url": value, "storage": "internal" if internal else "external", "exists_at_export": internal[1].is_file() if internal else None, }) for space in config.get("spaces") or []: add("plan", "space", str(space.get("id")), "plan_url", space.get("plan_url")) for marker in config.get("markers") or []: for index, pdf in enumerate(marker.get("pdfs") or []): add( "attachment", "marker", str(marker.get("id")), f"pdfs[{index}].url", pdf.get("url") if isinstance(pdf, dict) else None, ) # Asset bytes are deliberately not embedded. The manifest records identity # and availability so a cross-instance import can preserve the image box as # a repairable editor placeholder instead of silently deleting geometry. assets_root = config_root / ASSETS_DIR for space in config.get("spaces") or []: for shape in space.get("decor") or []: if not isinstance(shape, dict) or shape.get("kind") != "image": continue aid = shape.get("asset_id") if not isinstance(aid, str) or not re.fullmatch(r"[0-9a-f]{64}", aid): continue matches = sorted(assets_root.glob(f"{aid}.*")) if assets_root.is_dir() else [] blob = next((path for path in matches if path.suffix in {".png", ".jpg", ".webp", ".svg"}), None) metadata: dict[str, Any] = {} try: metadata = json.loads((assets_root / f"{aid}.json").read_text(encoding="utf-8")) except (OSError, ValueError, TypeError): pass exists = False if blob and blob.is_file(): try: exists = hashlib.sha256(blob.read_bytes()).hexdigest() == aid except OSError: pass out.append({ "kind": "decor_asset", "owner": "decor", "owner_id": f"{space.get('id', '')}:{shape.get('id', '')}", "field": "asset_id", # Identity is extension-neutral: a missing target must compute # exactly the same expected manifest as the source. "url": aid, "asset_id": aid, "storage": "internal", "mime": metadata.get("mime") or _DECOR_ASSET_MIME_BY_SUFFIX.get( blob.suffix if blob else "" ), "hash": aid, "exists_at_export": exists, }) return out def _safe_title(value: str) -> str: safe = _SAFE_FILE.sub("-", unicodedata.normalize("NFC", value).strip()).strip("-._") return (safe or "space")[:60] def create_export( runtime: HouseplanData, config_data: dict[str, Any], layout_data: dict[str, Any], *, kind: str, space_id: str | None, plan_only: bool = False, card_version: str, config_root: Path, ) -> tuple[dict[str, Any], str]: if not isinstance(plan_only, bool) or plan_only and kind != "space": raise ImportFailure("invalid_format", "Plan-only export requires one space") raw_config = _json_copy( config_data.get("config") or {"spaces": [], "markers": [], "settings": {}} ) # Read before voluptuous coercions: bool is an int subclass and must not # become a plausible model 1 in an otherwise valid export. model_version = _stored_model_version(raw_config) if model_version > PLAN_MODEL_VERSION: raise ImportFailure("future_model", "The stored plan uses a newer data model") try: raw_config["model_version"] = model_version config, _ = commit_wall_segment_model(raw_config) config = CONFIG_SCHEMA(config) except (WallSegmentMigrationError, vol.Invalid) as err: code = err.code if isinstance(err, WallSegmentMigrationError) else "invalid_config" raise ImportFailure(code, str(err)) from err model_version = PLAN_MODEL_VERSION # Compatibility belongs to the envelope. Keeping this field inside the # payload would either duplicate it or tempt an exporter to silently stamp # the current version over an older/future stored model. config.pop("model_version", None) global_background = _materialize_global_background(config) layout = live_layout(config, LAYOUT_SCHEMA(_json_copy(layout_data.get("layout") or {}))) stamp = datetime.now(UTC).strftime("%Y-%m-%d_%H-%M-%S") title = "" dropped_marker_links = 0 if kind == "space": space = next((sp for sp in config.get("spaces") or [] if str(sp.get("id")) == space_id), None) if not space: raise ImportFailure("space_not_found", "Space was not found") _materialize_space_background(space, global_background) title = str(space.get("title") or space.get("id") or "space") selected_layout = { key: pos for key, pos in layout.items() if isinstance(pos, dict) and str(pos.get("s")) == str(space_id) } if plan_only: projected_space = _project_plan_only_space(space) config = {"spaces": [projected_space], "markers": []} layout = _plan_only_room_label_layout(selected_layout, projected_space) else: selected_markers = [ m for m in config.get("markers") or [] if m.get("removed") is not True and _marker_owned(m, space, selected_layout) ] selected_ids = {str(marker.get("id")) for marker in selected_markers} for marker in selected_markers: controls = marker.get("controls") if isinstance(controls, list): kept = [] for ref in controls: if isinstance(ref, str) and ref.startswith("marker:") \ and ref[len("marker:"):] not in selected_ids: dropped_marker_links += 1 continue kept.append(ref) marker["controls"] = kept or None # #385(г): the two neutralisation formats below differ on # purpose and are a PAIR — each disarms an external ref the # way its own model represents "no source": the badge keeps # its object with enabled=False/source=None (enabled is part # of the badge model), while value_source's absence IS auto, # so the key is dropped. Both count in dropped_marker_links; # changing either format would break round-trips of existing # exports. badge = marker.get("value_badge") source = badge.get("source") if isinstance(badge, dict) else None ref = source.get("ref") if isinstance(source, dict) \ and source.get("kind") == "derived_marker_state" else None if isinstance(ref, str) and ref.startswith("marker:") \ and ref[len("marker:"):] not in selected_ids: badge["enabled"] = False badge["source"] = None dropped_marker_links += 1 value_source = marker.get("value_source") ref = value_source.get("ref") if isinstance(value_source, dict) \ and value_source.get("kind") == "derived_marker_state" else None if isinstance(ref, str) and ref.startswith("marker:") \ and ref[len("marker:"):] not in selected_ids: marker.pop("value_source", None) dropped_marker_links += 1 # #162: a robot map routed into ANOTHER space cannot travel # with a single-space export — its target would not exist on # import, and a route with a dead space resolves to nothing. # Same neutralisation contract as the marker refs above: drop # it and count it, so the preview states the loss up front. vacuum = marker.get("vacuum") routes = vacuum.get("map_routes") if isinstance(vacuum, dict) else None if isinstance(routes, list): kept_routes = [ route for route in routes if isinstance(route, dict) and str(route.get("space")) == str(space_id) ] dropped_marker_links += len(routes) - len(kept_routes) # An explicit list is route authority even when filtering # leaves it empty. Turning [] into None would revive any # legacy calibration dictionary after import (#443). vacuum["map_routes"] = kept_routes config = { "spaces": [_json_copy(space)], "markers": _json_copy(selected_markers), } layout = selected_layout elif kind != "full": raise ImportFailure("invalid_format", "Unknown export kind") document = { "format": FORMAT, "export_version": EXPORT_VERSION, "kind": kind, "created_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"), "source_fingerprint": source_fingerprint(runtime.instance_id), "card_version": card_version, "integration_version": VERSION, "model_version": model_version, "payload": {"config": config, "layout": layout}, "placement_manifest": placement_manifest(config, layout), "content_manifest": content_manifest(config, config_root), "transfer": { "dropped_marker_links": dropped_marker_links, **({"plan_only": True} if plan_only else {}), }, } if len(json.dumps( document, ensure_ascii=False, separators=(",", ":"), allow_nan=False ).encode("utf-8")) > MAX_EXPORT_BYTES: raise ImportFailure("too_large", "Export is larger than 8 MiB") filename = ( f"houseplan-full-{stamp}.json" if kind == "full" else f"houseplan-space-{_safe_title(title)}-{stamp}.json" ) return document, filename def _strict_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: out: dict[str, Any] = {} for key, value in pairs: if key in out: raise ImportFailure("invalid_json", f"Duplicate JSON key: {key}") if key in _PROTO_KEYS: raise ImportFailure("invalid_json", f"Forbidden JSON key: {key}") out[key] = value return out def parse_document(raw: bytes) -> dict[str, Any]: if len(raw) > MAX_EXPORT_BYTES: raise ImportFailure("too_large", "Import is larger than 8 MiB") try: # utf-8-sig accepts a BOM only at the very start. A later U+FEFF is a # normal character and cannot disguise trailing JSON. text = raw.decode("utf-8-sig") if "\ufeff" in text: raise ImportFailure("invalid_json", "A UTF-8 BOM is allowed only at the start") document = json.loads( text, object_pairs_hook=_strict_object, parse_constant=lambda value: (_ for _ in ()).throw( ImportFailure("invalid_json", f"Non-finite number: {value}") ), ) except ImportFailure: raise except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as err: raise ImportFailure("invalid_json", str(err)) from err if not isinstance(document, dict) or document.get("format") != FORMAT: raise ImportFailure("invalid_format", "Not a House Plan export") if document.get("export_version") not in (1, EXPORT_VERSION): raise ImportFailure("unsupported_export_version", "Unsupported export version") if document.get("kind") not in ("full", "space"): raise ImportFailure("invalid_format", "Export kind must be full or space") model = document.get("model_version", 0) if isinstance(model, bool) or not isinstance(model, int) or model < 0: raise ImportFailure("invalid_format", "Invalid data model version") if model > PLAN_MODEL_VERSION: raise ImportFailure("future_model", "The export uses a newer data model") payload = document.get("payload") if not isinstance(payload, dict): raise ImportFailure("invalid_format", "Missing payload") try: config_candidate = _json_copy(payload.get("config")) # The envelope is authoritative. Export deliberately omits this field # inside payload.config, but v8 semantic validation still has to run # before an import token can be issued. if model > 0: config_candidate["model_version"] = model config = CONFIG_SCHEMA(config_candidate) config.pop("model_version", None) except (vol.Invalid, TypeError, ValueError) as err: raise ImportFailure("invalid_config", str(err)) from err if document["kind"] == "full": _materialize_global_background(config) else: for space in config.get("spaces") or []: if isinstance(space, dict): _materialize_space_background(space, "static") try: layout = LAYOUT_SCHEMA(_json_copy(payload.get("layout") or {})) except (vol.Invalid, TypeError, ValueError) as err: raise ImportFailure("invalid_layout", str(err)) from err document["payload"] = {"config": config, "layout": layout} placement = document.get("placement_manifest") if not isinstance(placement, list): raise ImportFailure("invalid_format", "Missing placement manifest") if len(placement) > MAX_LAYOUT: raise ImportFailure("too_large", "Placement manifest exceeds the layout limit") placement_ids: set[str] = set() for item in placement: if not isinstance(item, dict): raise ImportFailure("invalid_format", "Invalid placement manifest") layout_id = item.get("layout_id", item.get("key")) if not isinstance(layout_id, str) or layout_id in placement_ids: raise ImportFailure("invalid_format", "Invalid placement manifest id") placement_ids.add(layout_id) if document["kind"] == "space" and placement_ids != set(layout): raise ImportFailure("invalid_format", "Placement manifest does not match layout") dropped_marker_links = _transfer_dropped_marker_links(document) plan_only = _transfer_plan_only(document) document["transfer"] = { **(document.get("transfer") or {}), "dropped_marker_links": dropped_marker_links, } if plan_only: _validate_plan_only_document(document, config, layout, placement) if len(json.dumps( config, ensure_ascii=False, separators=(",", ":"), allow_nan=False ).encode("utf-8")) > MAX_CONFIG_BYTES: raise ImportFailure("too_large", "Imported configuration exceeds the store limit") return document def _counts(config: dict[str, Any], layout: dict[str, Any]) -> dict[str, int]: spaces = config.get("spaces") or [] return { "spaces": len(spaces), "rooms": sum(len(sp.get("rooms") or []) for sp in spaces), "walls": sum( len(sp.get("walls") or []) for sp in spaces ), "markers": len(config.get("markers") or []), "openings": sum(len(sp.get("openings") or []) for sp in spaces), "decor": sum(len(sp.get("decor") or []) for sp in spaces), "stairs": sum(len(sp.get("stairs") or []) for sp in spaces), "layout": len(layout), } def _binding_key(marker: dict[str, Any]) -> str | None: binding = marker.get("binding") return binding if isinstance(binding, str) and binding != "virtual" else None def _layout_binding( key: str, marker_ids: set[str], manifest: dict[str, dict[str, Any]] | None = None, ) -> str | None: """Recover the literal HA binding owned by a layout-only placement.""" if key in marker_ids or key.startswith("rl_") or key.startswith("v_"): return None declared = (manifest or {}).get(key, {}).get("binding") if isinstance(declared, str) and declared != "virtual": return declared if key.startswith("lg_") and len(key) > 3: return "entity:" + key[3:] return "device:" + key def _binding_inventory( config: dict[str, Any], layout: dict[str, Any], manifest: dict[str, dict[str, Any]] | None = None, ) -> set[str]: markers = config.get("markers") or [] marker_ids = {str(marker.get("id")) for marker in markers} out = {binding for marker in markers if (binding := _binding_key(marker))} out.update( binding for key in layout if (binding := _layout_binding(str(key), marker_ids, manifest)) ) return out def _transfer_dropped_marker_links(document: dict[str, Any]) -> int: """Return the bounded, schema-independent transfer loss counter.""" transfer = document.get("transfer") if transfer is None: return 0 if not isinstance(transfer, dict): raise ImportFailure("invalid_format", "Transfer metadata must be an object") value = transfer.get("dropped_marker_links", 0) maximum = MAX_MARKERS * (MAX_CONTROLS + 2) if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= maximum: raise ImportFailure("invalid_format", "Invalid dropped marker link count") return value def _transfer_plan_only(document: dict[str, Any]) -> bool: """Read strict additive plan-only metadata without widening old exports.""" transfer = document.get("transfer") if transfer is None: return False if not isinstance(transfer, dict): raise ImportFailure("invalid_format", "Transfer metadata must be an object") value = transfer.get("plan_only", False) if not isinstance(value, bool): raise ImportFailure("invalid_format", "Plan-only metadata must be a boolean") if value and document.get("kind") != "space": raise ImportFailure("invalid_format", "Plan-only metadata requires one space") return value def _validate_plan_only_document( document: dict[str, Any], config: dict[str, Any], layout: dict[str, Any], placement: list[Any], ) -> None: """Reject forged plan-only flags unless every privacy invariant is true.""" spaces = config.get("spaces") or [] if len(spaces) != 1 or config.get("markers") != []: raise ImportFailure("invalid_format", "Plan-only export has invalid owners") expected_config = CONFIG_SCHEMA({ "spaces": [_project_plan_only_space(spaces[0])], "markers": [], }) if config != expected_config: raise ImportFailure("invalid_format", "Plan-only export contains private fields") expected_layout = _plan_only_room_label_layout(layout, spaces[0]) if layout != expected_layout: raise ImportFailure("invalid_format", "Plan-only export contains device layout") expected_placement = placement_manifest(config, layout) if placement != expected_placement: raise ImportFailure("invalid_format", "Plan-only placement manifest is not canonical") content = document.get("content_manifest") if not isinstance(content, list) or any( not isinstance(item, dict) or not ( item.get("owner") == "space" or item.get("owner") == "decor" and item.get("kind") == "decor_asset" ) for item in content ): raise ImportFailure("invalid_format", "Plan-only export contains private content") def _drop_invalid_import_marker_links( config: dict[str, Any], *, clean_ids: set[str] | None = None, ) -> int: """Drop import-only dormant/broken marker links without touching legacy data. Self-links and cycles remain hard errors in ``validate_marker_controls``. Missing, removed, non-light and duplicate targets are a normal loss of import context and are removed with an explicit preview counter. """ markers = config.get("markers") or [] by_id = {str(marker.get("id")): marker for marker in markers} dropped = 0 for marker in markers: marker_id = str(marker.get("id")) if clean_ids is not None and marker_id not in clean_ids: continue controls = marker.get("controls") if isinstance(controls, list): kept: list[Any] = [] seen_targets: set[str] = set() for ref in controls: if not isinstance(ref, str) or not ref.startswith("marker:"): kept.append(ref) continue target_id = ref[len("marker:"):] # Preserve one self-link so the semantic validator returns the # stable marker_control_self error instead of silently repairing it. if target_id == marker_id: if target_id in seen_targets: dropped += 1 continue seen_targets.add(target_id) kept.append(ref) continue target = by_id.get(target_id) if target_id in seen_targets or target is None or target.get("removed") is True \ or target.get("is_light") is not True: dropped += 1 continue seen_targets.add(target_id) kept.append(ref) marker["controls"] = kept or None badge = marker.get("value_badge") source = badge.get("source") if isinstance(badge, dict) else None if isinstance(source, dict) and source.get("kind") == "derived_marker_state": ref = source.get("ref") target_id = ref[len("marker:"):] if isinstance(ref, str) and ref.startswith("marker:") else "" target = by_id.get(target_id) if not target_id or target is None or target.get("removed") is True \ or target.get("is_light") is not True: badge["enabled"] = False badge["source"] = None dropped += 1 source = marker.get("value_source") if isinstance(source, dict) and source.get("kind") == "derived_marker_state": ref = source.get("ref") target_id = ref[len("marker:"):] \ if isinstance(ref, str) and ref.startswith("marker:") else "" target = by_id.get(target_id) if not target_id or target is None or target.get("removed") is True \ or target.get("is_light") is not True: marker.pop("value_source", None) dropped += 1 return dropped def _unique_title(base: str, existing: list[dict[str, Any]]) -> str: base = unicodedata.normalize("NFC", base).strip() or "Space" names = { unicodedata.normalize("NFC", str(sp.get("title") or "").strip()).casefold() for sp in existing } if unicodedata.normalize("NFC", base).casefold() not in names: return base n = 2 while unicodedata.normalize("NFC", f"{base} ({n})").casefold() in names: n += 1 return f"{base} ({n})" def canonical_import_root(prefix: str, value: str) -> tuple[str, int, bool]: """Return the bounded root of ids generated by a previous space import. Only the exact ``__<8 lowercase hex>`` envelope is reversible. Similar user ids, other namespaces and uppercase/short hashes remain literal. The layer count and bounded flag are shared with the TypeScript helper through one conformance fixture. The flag reports that another valid layer remained after the safety bound. """ root = str(value) pattern = re.compile(rf"^{re.escape(prefix)}_(.+)_([0-9a-f]{{8}})$") layers = 0 for _depth in range(_MAX_IMPORT_LINEAGE_DEPTH): match = pattern.fullmatch(root) if match is None: return root, layers, False root = match.group(1) layers += 1 return root, layers, pattern.fullmatch(root) is not None def _fresh(prefix: str, old: str, used: set[str]) -> str: root, _layers, _bounded = canonical_import_root(prefix, old) \ if prefix in _IMPORT_ID_NAMESPACES else (old, 0, False) stem = re.sub(r"[^a-zA-Z0-9_-]+", "_", root).strip("_")[:35] or prefix while True: value = f"{prefix}_{stem}_{secrets.token_hex(4)}" if value not in used: used.add(value) return value def _orphan_marker(key: str, manifest: dict[str, Any] | None, space_id: str, used: set[str]) -> dict[str, Any]: marker_id = _fresh("v_import", key, used) return { "id": marker_id, "binding": "virtual", "space": space_id, "name": (manifest or {}).get("label") or key, "icon": (manifest or {}).get("icon") or "mdi:map-marker-outline", "display": "static_icon", } def _empty_reference_report() -> dict[str, Any]: return { "remapped": {"incoming": {}, "target": {}}, "collisions": {}, "preservedUnresolved": {}, "droppedIncomingLinks": {}, "boundedLineages": 0, "examples": [], } def _report_reference( report: dict[str, Any], bucket: str, category: str, owner: str, reference: str, ) -> None: section = report.setdefault(bucket, {}) section[category] = int(section.get(category, 0)) + 1 examples = report.setdefault("examples", []) if len(examples) < _REPORT_EXAMPLE_LIMIT: examples.append({ "bucket": bucket, "category": category, "owner": str(owner)[:160], "reference": str(reference)[:160], }) def _report_remap( report: dict[str, Any], side: str, category: str, owner: str, reference: str, ) -> None: section = report.setdefault("remapped", {}).setdefault(side, {}) section[category] = int(section.get(category, 0)) + 1 examples = report.setdefault("examples", []) if len(examples) < _REPORT_EXAMPLE_LIMIT: examples.append({ "bucket": f"remapped.{side}", "category": category, "owner": str(owner)[:160], "reference": str(reference)[:160], }) def _record_bounded_lineage( report: dict[str, Any], seen: set[tuple[str, str]], prefix: str, value: str, ) -> None: _root, _layers, bounded = canonical_import_root(prefix, value) if bounded: seen.add((prefix, value)) report["boundedLineages"] = len(seen) def _lineage_resolver( prefix: str, exact_map: dict[str, str], live_ids: set[str], report: dict[str, Any], bounded_seen: set[tuple[str, str]], ) -> Callable[[str], tuple[str | None, str]]: """Resolve only dead exact or uniquely provable cross-generation refs.""" imported: dict[str, set[str]] = {} live: dict[str, set[str]] = {} for old, new in exact_map.items(): root, _layers, bounded = canonical_import_root(prefix, old) imported.setdefault(root, set()).add(new) if bounded: _record_bounded_lineage(report, bounded_seen, prefix, old) for value in live_ids: root, _layers, bounded = canonical_import_root(prefix, value) live.setdefault(root, set()).add(value) if bounded: _record_bounded_lineage(report, bounded_seen, prefix, value) def resolve(reference: str) -> tuple[str | None, str]: if reference in live_ids: return None, "live" if reference in exact_map: return exact_map[reference], "exact" root, _layers, bounded = canonical_import_root(prefix, reference) if bounded: _record_bounded_lineage(report, bounded_seen, prefix, reference) candidates = imported.get(root, set()) if not candidates: return None, "unrelated" if len(candidates) == 1 and not live.get(root): return next(iter(candidates)), "lineage" return None, "ambiguous" return resolve def _repair_target_space_refs( current_config: dict[str, Any], current_layout: dict[str, Any], id_maps: dict[str, dict[str, str]], marker_link_map: dict[str, str], report: dict[str, Any], bounded_seen: set[tuple[str, str]], ) -> tuple[dict[str, Any], dict[str, Any], int]: """Repair target refs by exact map, then by one safe lineage candidate.""" config = _json_copy(current_config) layout = _json_copy(current_layout) spaces = config.get("spaces") or [] markers = config.get("markers") or [] live_spaces = {str(item.get("id")) for item in spaces if item.get("id") is not None} live_rooms = { str(room.get("id")) for space in spaces for room in space.get("rooms") or [] if room.get("id") is not None } live_partitions = { str(item.get("id")) for space in spaces for item in space.get("partitions") or [] if item.get("id") is not None } live_markers = { str(marker.get("id")) for marker in markers if marker.get("id") is not None } resolve_space = _lineage_resolver( "space", id_maps.get("space", {}), live_spaces, report, bounded_seen, ) resolve_room = _lineage_resolver( "room", id_maps.get("room", {}), live_rooms, report, bounded_seen, ) resolve_partition = _lineage_resolver( "partition", id_maps.get("partition", {}), live_partitions, report, bounded_seen, ) resolve_marker = _lineage_resolver( "marker", id_maps.get("marker", {}), live_markers, report, bounded_seen, ) resolve_marker_link = _lineage_resolver( "marker", marker_link_map, live_markers, report, bounded_seen, ) imported_space_ids = set(id_maps.get("space", {}).values()) repaired = 0 unresolved: set[tuple[str, str, str]] = set() def replace( owner: str, category: str, reference: Any, resolver: Callable[[str], tuple[str | None, str]], ) -> str | None: nonlocal repaired if not isinstance(reference, str) or not reference: return None mapped, reason = resolver(reference) if mapped is not None: repaired += 1 _report_remap(report, "target", category, owner, reference) return mapped if reason == "ambiguous": preserve_once(owner, category, reference) return None def preserve_once(owner: str, category: str, reference: str) -> None: key = (owner, category, reference) if key in unresolved: return unresolved.add(key) _report_reference( report, "preservedUnresolved", category, owner, reference, ) def preserve_related( owner: str, category: str, reference: Any, resolver: Callable[[str], tuple[str | None, str]], ) -> None: if not isinstance(reference, str) or not reference: return mapped, reason = resolver(reference) if mapped is not None or reason == "ambiguous": preserve_once(owner, category, reference) for space in spaces: space_id = str(space.get("id", "?")) for stair in space.get("stairs") or []: target = stair.get("target_space_id") if isinstance(stair, dict) else None mapped = replace( f"{space_id}:{stair.get('id', '?')}", "stair.target_space_id", target, resolve_space, ) if mapped is not None: stair["target_space_id"] = mapped for room in space.get("rooms") or []: room_id = str(room.get("id", "?")) values = room.get("open_to") if isinstance(values, list): for value in values: preserve_related( f"{space_id}:{room_id}", "room.open_to", value, resolve_room, ) for opening in space.get("openings") or []: host = opening.get("host") if isinstance(opening, dict) else None if isinstance(host, dict) and host.get("kind") == "partition": # A partition host is space-local. An imported partition lives # in the new independent copy, so a pre-existing opening in a # different target space cannot be rebound to it safely. preserve_related( f"{space_id}:{opening.get('id', '?')}", "opening.host", host.get("id"), resolve_partition, ) for marker in markers: marker_id = str(marker.get("id", "?")) mapped = replace(marker_id, "marker.space", marker.get("space"), resolve_space) if mapped is not None: marker["space"] = mapped may_rebind_room = marker.get("space") in imported_space_ids if may_rebind_room: mapped = replace(marker_id, "marker.room_id", marker.get("room_id"), resolve_room) if mapped is not None: marker["room_id"] = mapped else: preserve_related( marker_id, "marker.room_id", marker.get("room_id"), resolve_room, ) vacuum = marker.get("vacuum") segment_map = vacuum.get("segment_map") if isinstance(vacuum, dict) else None if isinstance(segment_map, dict): for key, room_id in list(segment_map.items()): if may_rebind_room: mapped = replace( marker_id, "marker.vacuum.segment_map", room_id, resolve_room, ) if mapped is not None: segment_map[key] = mapped else: preserve_related( marker_id, "marker.vacuum.segment_map", room_id, resolve_room, ) routes = vacuum.get("map_routes") if isinstance(vacuum, dict) else None if isinstance(routes, list): for route in routes: if not isinstance(route, dict): continue mapped = replace( marker_id, "marker.vacuum.map_routes.space", route.get("space"), resolve_space, ) if mapped is not None: route["space"] = mapped controls = marker.get("controls") if isinstance(controls, list): for index, ref in enumerate(controls): if not isinstance(ref, str) or not ref.startswith("marker:"): continue old_target = ref[len("marker:"):] mapped = replace( marker_id, "marker.controls", old_target, resolve_marker_link, ) if mapped is not None: controls[index] = "marker:" + mapped else: # A duplicate may have been virtualised or otherwise lost # light-source semantics. Such a marker remains a valid # layout owner, but it is not a valid controls target. preserve_related( marker_id, "marker.controls", old_target, resolve_marker, ) badge = marker.get("value_badge") source = badge.get("source") if isinstance(badge, dict) else None if isinstance(source, dict) and source.get("kind") == "derived_marker_state": ref = source.get("ref") old_target = ref[len("marker:"):] \ if isinstance(ref, str) and ref.startswith("marker:") else None mapped = replace( marker_id, "marker.value_badge", old_target, resolve_marker_link, ) if mapped is not None: source["ref"] = "marker:" + mapped else: preserve_related( marker_id, "marker.value_badge", old_target, resolve_marker, ) source = marker.get("value_source") if isinstance(source, dict) and source.get("kind") == "derived_marker_state": ref = source.get("ref") old_target = ref[len("marker:"):] \ if isinstance(ref, str) and ref.startswith("marker:") else None mapped = replace( marker_id, "marker.value_source", old_target, resolve_marker_link, ) if mapped is not None: source["ref"] = "marker:" + mapped else: preserve_related( marker_id, "marker.value_source", old_target, resolve_marker, ) radar = marker.get("radar") if isinstance(radar, dict): if may_rebind_room: mapped = replace( marker_id, "marker.radar.room_id", radar.get("room_id"), resolve_room, ) if mapped is not None: radar["room_id"] = mapped allowed = radar.get("allowed_room_ids") if isinstance(allowed, list): remapped_allowed: list[Any] = [] for room_id in allowed: mapped = replace( marker_id, "marker.radar.allowed_room_ids", room_id, resolve_room, ) remapped_allowed.append(mapped if mapped is not None else room_id) radar["allowed_room_ids"] = remapped_allowed else: preserve_related( marker_id, "marker.radar.room_id", radar.get("room_id"), resolve_room, ) for room_id in radar.get("allowed_room_ids") or []: preserve_related( marker_id, "marker.radar.allowed_room_ids", room_id, resolve_room, ) # Destination wins collisions. Rekey only proven dead plan-owned keys; # opaque HA owners remain literal. for key in list(layout): position = layout[key] if isinstance(position, dict): mapped_space = replace(key, "layout.space", position.get("s"), resolve_space) if mapped_space is not None: position["s"] = mapped_space new_key: str | None = None category = "" if key.startswith("rl_"): if isinstance(position, dict) and position.get("s") in imported_space_ids: mapped_room = replace(key, "layout.room_label", key[3:], resolve_room) if mapped_room is not None: new_key, category = "rl_" + mapped_room, "layout.room_label" else: preserve_related(key, "layout.room_label", key[3:], resolve_room) elif key not in live_markers: mapped_marker = replace(key, "layout.marker", key, resolve_marker) if mapped_marker is not None: new_key, category = mapped_marker, "layout.marker" if new_key is None or new_key == key: continue if new_key in layout: _report_reference(report, "collisions", category, key, new_key) else: layout[new_key] = position del layout[key] return config, layout, repaired def build_space_merge( document: dict[str, Any], current_config: dict[str, Any], current_layout: dict[str, Any], duplicate_policy: str, *, same_source: bool = False, ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: """Create the exact independent space copy used by preview and apply.""" if duplicate_policy not in ("skip", "virtual"): raise ImportFailure("invalid_format", "Unknown duplicate policy") incoming = _json_copy(document["payload"]["config"]) spaces = incoming.get("spaces") or [] if len(spaces) != 1: raise ImportFailure("invalid_config", "A space export must contain exactly one space") space = spaces[0] _materialize_space_background(space, "static") reference_report = _empty_reference_report() bounded_seen: set[tuple[str, str]] = set() target_config = _json_copy(current_config) target_layout = _json_copy(current_layout) used = { str(value) for sp in current_config.get("spaces") or [] for collection in ( (sp,), sp.get("rooms") or [], sp.get("partitions") or [], sp.get("wall_segments") or [], sp.get("wall_columns") or [], sp.get("openings") or [], sp.get("decor") or [], sp.get("stairs") or [], ) for item in collection if isinstance(item, dict) for value in [item.get("id")] if value is not None } | { str(m.get("id")) for m in current_config.get("markers") or [] if m.get("id") is not None } | { str(value) for incoming_space in spaces for collection in ( (incoming_space,), incoming_space.get("rooms") or [], incoming_space.get("partitions") or [], incoming_space.get("wall_segments") or [], incoming_space.get("wall_columns") or [], incoming_space.get("openings") or [], incoming_space.get("decor") or [], incoming_space.get("stairs") or [], ) for item in collection if isinstance(item, dict) for value in [item.get("id")] if value is not None } | { str(m.get("id")) for m in incoming.get("markers") or [] if m.get("id") is not None } old_space_id = str(space.get("id")) _record_bounded_lineage( reference_report, bounded_seen, "space", old_space_id, ) new_space_id = _fresh("space", old_space_id, used) id_map: dict[str, str] = {old_space_id: new_space_id} id_maps: dict[str, dict[str, str]] = { prefix: {} for prefix in _IMPORT_ID_NAMESPACES } id_maps["space"][old_space_id] = new_space_id for collection, prefix in ( ("rooms", "room"), ("partitions", "partition"), ("wall_segments", "wall"), ("wall_columns", "column"), ("openings", "opening"), ("decor", "decor"), ("stairs", "stair"), ): for item in space.get(collection) or []: if isinstance(item, dict) and item.get("id") is not None: old = str(item["id"]) _record_bounded_lineage( reference_report, bounded_seen, prefix, old, ) id_map[old] = _fresh(prefix, old, used) id_maps[prefix][old] = id_map[old] item["id"] = id_map[old] old_room_ids = id_maps["room"] for room in space.get("rooms") or []: if room.get("open_to"): remapped_open_to = [] for value in room["open_to"]: old_value = str(value) mapped_value = old_room_ids.get(old_value, old_value) remapped_open_to.append(mapped_value) if mapped_value != old_value: _report_remap( reference_report, "incoming", "room.open_to", str(room.get("id", "?")), old_value, ) room["open_to"] = remapped_open_to if room.get("wall_ids"): room["wall_ids"] = [ id_maps["wall"].get(str(value), str(value)) for value in room["wall_ids"] ] # Opening ownership is part of the same space-local id graph. Remap the # nested reference together with the partition itself; otherwise the # invariant validator correctly rejects a copied space whose host.id still # names the source partition. for opening in space.get("openings") or []: host = opening.get("host") if isinstance(opening, dict) else None if isinstance(host, dict) and host.get("kind") in ("partition", "wall"): old_host_id = str(host.get("id")) if old_host_id in id_map: host["id"] = id_map[old_host_id] _report_remap( reference_report, "incoming", "opening.host", str(opening.get("id", "?")), old_host_id, ) # A one-space export cannot carry a trustworthy external floor target. # Self-links are invalid too because the imported source gets a new id. for stair in space.get("stairs") or []: target = stair.get("target_space_id") if isinstance(stair, dict) else None if target is not None: stair["target_space_id"] = None _report_remap( reference_report, "incoming", "stair.target_space_id", str(stair.get("id", "?")), str(target), ) space["id"] = new_space_id space["title"] = _unique_title( str(space.get("title") or old_space_id), current_config.get("spaces") or [] ) incoming_layout = document["payload"].get("layout") or {} manifest = { str(item.get("layout_id", item.get("key"))): item for item in document.get("placement_manifest") or [] if isinstance(item, dict) } incoming_bindings = _binding_inventory(incoming, incoming_layout, manifest) current_bindings = _binding_inventory(target_config, target_layout) duplicate = incoming_bindings & current_bindings marker_map: dict[str, str] = {} output_markers: list[dict[str, Any]] = [] skipped: set[str] = set() virtualized_targets: set[str] = set() virtualized = 0 dropped_marker_links = _transfer_dropped_marker_links(document) for marker in incoming.get("markers") or []: old_id = str(marker.get("id")) _record_bounded_lineage( reference_report, bounded_seen, "marker", old_id, ) binding = _binding_key(marker) if binding in duplicate and duplicate_policy == "skip": skipped.add(old_id) continue new_id = _fresh("marker", old_id, used) marker_map[old_id] = new_id id_maps["marker"][old_id] = new_id marker["id"] = new_id marker["space"] = new_space_id _report_remap( reference_report, "incoming", "marker.space", new_id, old_space_id, ) if marker.get("room_id") is not None: old_room_id = str(marker["room_id"]) marker["room_id"] = old_room_ids.get(old_room_id, marker["room_id"]) if marker["room_id"] != old_room_id: _report_remap( reference_report, "incoming", "marker.room_id", new_id, old_room_id, ) radar = marker.get("radar") if isinstance(radar, dict): radar_room = radar.get("room_id") if radar_room is not None: old_radar_room = str(radar_room) radar["room_id"] = old_room_ids.get(old_radar_room, radar_room) if radar["room_id"] != old_radar_room: _report_remap( reference_report, "incoming", "marker.radar.room_id", new_id, old_radar_room, ) allowed_rooms = radar.get("allowed_room_ids") if isinstance(allowed_rooms, list): radar["allowed_room_ids"] = [ old_room_ids.get(str(room_id), room_id) for room_id in allowed_rooms ] vacuum = marker.get("vacuum") if isinstance(vacuum, dict) and isinstance(vacuum.get("segment_map"), dict): remapped_segments = {} for key, value in vacuum["segment_map"].items(): old_room_id = str(value) remapped_segments[key] = old_room_ids.get(old_room_id, value) if remapped_segments[key] != old_room_id: _report_remap( reference_report, "incoming", "marker.vacuum.segment_map", new_id, old_room_id, ) vacuum["segment_map"] = remapped_segments # A single-space export can retain only routes owned by that space. # The copy itself gets a fresh id, so the retained route must follow # it just like marker.space does; otherwise foreign imports fail the # referential validator and same-instance imports silently target the # source floor (#611). if isinstance(vacuum, dict) and isinstance(vacuum.get("map_routes"), list): for route in vacuum["map_routes"]: if not isinstance(route, dict) \ or str(route.get("space")) != old_space_id: continue route["space"] = new_space_id _report_remap( reference_report, "incoming", "marker.vacuum.map_routes.space", new_id, old_space_id, ) if binding in duplicate and duplicate_policy == "virtual": virtualized += 1 virtualized_targets.add(old_id) dropped_marker_links += sum( 1 for ref in marker.get("controls") or [] if isinstance(ref, str) and ref.startswith("marker:") ) badge = marker.get("value_badge") source = badge.get("source") if isinstance(badge, dict) else None if isinstance(source, dict) and source.get("kind") == "derived_marker_state": dropped_marker_links += 1 value_source = marker.get("value_source") if isinstance(value_source, dict) \ and value_source.get("kind") == "derived_marker_state": dropped_marker_links += 1 if isinstance(marker.get("radar"), dict): _report_reference( reference_report, "droppedIncomingLinks", "marker.radar", new_id, old_id, ) marker["binding"] = "virtual" marker["display"] = "static_icon" for field in ( "area", "controls", "tap_action", "tap_target", "tap_confirm", "vacuum", "is_light", "use_climate_temp", "glow_color", "glow_radius_cm", "light_entity", "toggle_entity", "hidden", "removed", "value_badge", "value_source", "radar", ): marker.pop(field, None) output_markers.append(marker) # Marker links share the marker id namespace and therefore follow the same # remap as layout ownership. Targets skipped/virtualised by duplicate # policy are dropped explicitly; raw marker:* values must never leak into # the imported independent copy. for marker in output_markers: controls = marker.get("controls") if isinstance(controls, list): remapped = [] for ref in controls: if not isinstance(ref, str) or not ref.startswith("marker:"): remapped.append(ref) continue old_target = ref[len("marker:"):] target = None if old_target in virtualized_targets else marker_map.get(old_target) if not target: dropped_marker_links += 1 continue remapped.append("marker:" + target) _report_remap( reference_report, "incoming", "marker.controls", str(marker.get("id", "?")), old_target, ) marker["controls"] = remapped or None badge = marker.get("value_badge") source = badge.get("source") if isinstance(badge, dict) else None if isinstance(source, dict) and source.get("kind") == "derived_marker_state": ref = source.get("ref") old_target = ref[len("marker:"):] if isinstance(ref, str) and ref.startswith("marker:") else "" target = None if old_target in virtualized_targets else marker_map.get(old_target) if target: source["ref"] = "marker:" + target _report_remap( reference_report, "incoming", "marker.value_badge", str(marker.get("id", "?")), old_target, ) else: badge["enabled"] = False badge["source"] = None dropped_marker_links += 1 source = marker.get("value_source") if isinstance(source, dict) and source.get("kind") == "derived_marker_state": ref = source.get("ref") old_target = ref[len("marker:"):] \ if isinstance(ref, str) and ref.startswith("marker:") else "" target = None if old_target in virtualized_targets else marker_map.get(old_target) if target: source["ref"] = "marker:" + target _report_remap( reference_report, "incoming", "marker.value_source", str(marker.get("id", "?")), old_target, ) else: marker.pop("value_source", None) dropped_marker_links += 1 output_markers_by_id = { str(marker.get("id")): marker for marker in output_markers if marker.get("id") is not None } marker_link_map = { old_id: new_id for old_id, new_id in marker_map.items() if (target := output_markers_by_id.get(new_id)) is not None and target.get("removed") is not True and target.get("is_light") is True } target_config, target_layout, repaired_target_refs = _repair_target_space_refs( current_config, current_layout, id_maps, marker_link_map, reference_report, bounded_seen, ) output_layout: dict[str, Any] = {} for key, pos in (document["payload"].get("layout") or {}).items(): if key in skipped: continue new_key = marker_map.get(key) if new_key is None and key.startswith("rl_"): room_id = old_room_ids.get(key[3:]) new_key = "rl_" + room_id if room_id else None layout_binding = _layout_binding( key, {str(marker.get("id")) for marker in incoming.get("markers") or []}, manifest ) if new_key is None and layout_binding in duplicate and duplicate_policy == "skip": skipped.add(key) continue if new_key is None and same_source and layout_binding and layout_binding not in duplicate: # Same-instance layout-only owners remain discoverable. Foreign # instances cannot safely claim that their opaque HA ids identify # the same target object and are materialised as virtual orphans. new_key = key if new_key is None: orphan = _orphan_marker(key, manifest.get(key), new_space_id, used) output_markers.append(orphan) new_key = orphan["id"] if layout_binding in duplicate and duplicate_policy == "virtual": virtualized += 1 if new_key != key: _report_remap( reference_report, "incoming", "layout.owner", new_key, key, ) old_layout_space = str(pos.get("s", "")) if isinstance(pos, dict) else "" if old_layout_space != new_space_id: _report_remap( reference_report, "incoming", "layout.space", new_key, old_layout_space, ) output_layout[new_key] = {**pos, "s": new_space_id} merged_config = target_config merged_config.setdefault("spaces", []).append(space) merged_config.setdefault("markers", []).extend(output_markers) dropped_marker_links += _drop_invalid_import_marker_links( merged_config, clean_ids={str(marker.get("id")) for marker in output_markers}, ) merged_layout = {**target_layout, **output_layout} if len(merged_config.get("spaces") or []) > MAX_SPACES \ or len(merged_config.get("markers") or []) > MAX_MARKERS \ or len(merged_layout) > MAX_LAYOUT: raise ImportFailure("capacity_exceeded", "Merged plan exceeds an item limit") try: merged_config = CONFIG_SCHEMA(merged_config) except vol.Invalid as err: raise ImportFailure("invalid_config", str(err)) from err try: validate_marker_controls(merged_config, current_config) validate_marker_light_entities(merged_config, current_config) validate_marker_radars(merged_config, current_config) validate_marker_value_badges(merged_config, current_config) validate_marker_vacuum_routes(merged_config, current_config) validate_opening_passages(merged_config, current_config) validate_partition_opening_hosts(merged_config, current_config) except ( MarkerControlError, OpeningPassageError, PartitionOpeningHostError, PartitionOpeningJambMarginError, ) as err: raise ImportFailure(err.code, str(err)) from err except vol.Invalid as err: raise ImportFailure("invalid_config", str(err)) from err try: merged_layout = LAYOUT_SCHEMA(merged_layout) except vol.Invalid as err: raise ImportFailure("invalid_layout", str(err)) from err if len(json.dumps( merged_config, ensure_ascii=False, separators=(",", ":"), allow_nan=False ).encode("utf-8")) > MAX_CONFIG_BYTES: raise ImportFailure("capacity_exceeded", "Merged configuration exceeds the store limit") if dropped_marker_links: reference_report["droppedIncomingLinks"]["marker.links"] = dropped_marker_links return merged_config, merged_layout, { "space_id": new_space_id, "space_title": space["title"], "duplicates": len(duplicate), "skipped": len(skipped), "virtualized": virtualized, "orphan_markers": len(output_markers) - len(marker_map), "dropped_marker_links": dropped_marker_links, "repaired_target_refs": repaired_target_refs, "preserved_unresolved_refs": sum( int(value) for value in reference_report["preservedUnresolved"].values() ), "reference_report": reference_report, } def _content_state(document: dict[str, Any], same_source: bool, config_root: Path) -> tuple[list[dict[str, Any]], bool]: """Validate the manifest against payload refs and resolve local policy. The manifest is descriptive, never authoritative: otherwise a crafted file could omit an internal ref or label it external and bypass the mandatory detach decision. Every row must correspond exactly to a normalized config reference and its storage class is derived again on the target instance. """ expected = content_manifest(document["payload"]["config"], config_root) supplied = document.get("content_manifest") if not isinstance(supplied, list): raise ImportFailure("invalid_content", "Missing content manifest") def identity(item: dict[str, Any]) -> tuple[str, str, str, str, str]: return ( str(item.get("kind", "")), str(item.get("owner", "")), str(item.get("owner_id", "")), str(item.get("field", "")), str(item.get("url", "")), ) supplied_by_id: dict[tuple[str, str, str, str, str], dict[str, Any]] = {} for item in supplied: if not isinstance(item, dict) or not isinstance(item.get("url"), str): raise ImportFailure("invalid_content", "Invalid content manifest") key = identity(item) if key in supplied_by_id: raise ImportFailure("invalid_content", "Duplicate content manifest row") supplied_by_id[key] = item expected_ids = {identity(item) for item in expected} if set(supplied_by_id) != expected_ids: raise ImportFailure("invalid_content", "Content manifest does not match payload references") rows: list[dict[str, Any]] = [] confirmation = False for item in expected: row = dict(item) declared = supplied_by_id[identity(item)] declared_exists = declared.get("exists_at_export") row["exists_at_export"] = declared_exists if item.get("kind") == "decor_asset": aid = str(item.get("asset_id") or item.get("url") or "") declared_mime = declared.get("mime") valid_mime = ( isinstance(declared_mime, str) and declared_mime in _DECOR_ASSET_MIME_TYPES ) missing_mime = declared_exists is False and declared_mime is None if (declared.get("asset_id") != aid or declared.get("hash") != aid or type(declared_exists) is not bool or not (valid_mime or missing_mime)): raise ImportFailure("invalid_content", "Invalid decor asset manifest row") # Preserve the source description for preview/digest purposes. It # never selects or serves target bytes; the exact local hash below # remains authoritative for availability. row["mime"] = declared_mime candidates = sorted((config_root / ASSETS_DIR).glob(f"{aid}.*")) blob = next((path for path in candidates if path.suffix in {".png", ".jpg", ".webp", ".svg"}), None) exists = False if blob: try: exists = hashlib.sha256(blob.read_bytes()).hexdigest() == aid except OSError: pass row["exists_on_target"] = exists row["state"] = "available" if exists else "missing_preserved" confirmation = confirmation or not exists rows.append(row) continue internal = _internal_path(config_root, item["url"]) if _looks_internal(item["url"]) and internal is None: raise ImportFailure("invalid_content", "Non-canonical internal content URL") if internal is not None: exists = bool(internal and internal[1].is_file()) row["exists_on_target"] = exists state = "available" if same_source and exists else "detach_required" confirmation = confirmation or state == "detach_required" else: state = "external" row["state"] = state rows.append(row) return rows, confirmation def _detach_missing(config: dict[str, Any], content: list[dict[str, Any]]) -> None: spaces = {str(sp.get("id")): sp for sp in config.get("spaces") or []} markers = {str(m.get("id")): m for m in config.get("markers") or []} for item in content: if item.get("state") != "detach_required": continue if item.get("owner") == "space": owner = spaces.get(str(item.get("owner_id"))) if owner: owner["plan_url"] = None elif item.get("owner") == "marker": owner = markers.get(str(item.get("owner_id"))) if owner: url = item.get("url") owner["pdfs"] = [pdf for pdf in owner.get("pdfs") or [] if pdf.get("url") != url] def _binding_preview( config: dict[str, Any], layout: dict[str, Any], document: dict[str, Any], registry: dict[str, set[str]] | None, ) -> dict[str, int]: manifest = { str(item.get("layout_id", item.get("key"))): item for item in document.get("placement_manifest") or [] if isinstance(item, dict) } bindings = _binding_inventory(config, layout, manifest) result = { "device": sum(value.startswith("device:") for value in bindings), "entity": sum(value.startswith("entity:") for value in bindings), "virtual": sum( marker.get("binding") == "virtual" for marker in config.get("markers") or [] ), "active": 0, "disabled": 0, "missing": 0, } if registry is None: return result for binding in bindings: prefix, _, target = binding.partition(":") if target in registry.get(f"disabled_{prefix}", set()): result["disabled"] += 1 elif target in registry.get(f"active_{prefix}", set()): result["active"] += 1 else: result["missing"] += 1 return result def _missing_areas( config: dict[str, Any], registry: dict[str, set[str]] | None, ) -> list[str]: if registry is None: return [] available = registry.get("areas", set()) incoming = { str(room.get("area")) for space in config.get("spaces") or [] for room in space.get("rooms") or [] if room.get("area") } return sorted(incoming - available) def _materialize_import_candidate( document: dict[str, Any], current_config: dict[str, Any], current_layout: dict[str, Any], *, duplicate_policy: str, same_source: bool, content: list[dict[str, Any]], confirmation_required: bool, ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: """Build once the normalized config/layout that the preview may apply.""" prepared = _json_copy(document) imported_config = prepared["payload"]["config"] if confirmation_required: _detach_missing(imported_config, content) incoming_model = int(prepared.get("model_version", 0) or 0) migration: dict[str, int] = {} try: # Import preview and apply operate only on the current materialized # wall model. Legacy drafts are converted before copy/remap, so no # current merge path can recreate their carrier. imported_config["model_version"] = incoming_model imported_config, _ = commit_wall_segment_model( imported_config, migration_report=migration, ) prepared["payload"]["config"] = imported_config if prepared["kind"] == "space": current_config, _ = commit_wall_segment_model(current_config) except WallSegmentMigrationError as err: raise ImportFailure(err.code, str(err)) from err if prepared["kind"] == "space": config, layout, details = build_space_merge( prepared, current_config, current_layout, duplicate_policy, same_source=same_source, ) else: config = imported_config _materialize_global_background(config) layout = _json_copy(prepared["payload"]["layout"]) if not same_source: settings = config.get("settings") or {} settings.pop("known_devices", None) settings.pop("new_device_ids", None) settings.pop("marker_area_snapshot", None) details = { "dropped_marker_links": _transfer_dropped_marker_links(prepared), } try: config = CONFIG_SCHEMA(config) except vol.Invalid as err: raise ImportFailure("invalid_config", str(err)) from err try: validate_active_marker_ids( config, current_config if prepared["kind"] == "space" else None, validate_all=prepared["kind"] == "full", ) except DuplicateMarkerIdError as err: raise ImportFailure(err.code, str(err)) from err try: layout = LAYOUT_SCHEMA(layout) except vol.Invalid as err: raise ImportFailure("invalid_layout", str(err)) from err details = _json_copy(details) details["migration"] = migration details["counts"] = _counts(config, layout) return config, layout, details def _candidate_digest(candidate: dict[str, Any]) -> str: """Bind a token to both its source and the exact normalized target pair.""" return _document_digest({ "document": candidate.get("document"), "duplicate_policy": candidate.get("duplicate_policy"), "config_rev": candidate.get("config_rev"), "layout_rev": candidate.get("layout_rev"), "content": candidate.get("content"), "confirmation_required": candidate.get("confirmation_required"), "same_source": candidate.get("same_source"), "target_config": candidate.get("target_config"), "target_layout": candidate.get("target_layout"), "details": candidate.get("details"), }) def create_preview( runtime: HouseplanData, raw: bytes, *, owner_id: str, duplicate_policy: str, current_config_data: dict[str, Any], current_layout_data: dict[str, Any], config_root: Path, registry_snapshot: dict[str, set[str]] | None = None, ) -> dict[str, Any]: document = parse_document(raw) same_source = document.get("source_fingerprint") == source_fingerprint(runtime.instance_id) incoming_config = document["payload"]["config"] incoming_layout = document["payload"]["layout"] current_config = current_config_data.get("config") or {"spaces": [], "markers": [], "settings": {}} current_layout = current_layout_data.get("layout") or {} details: dict[str, Any] = {} try: # Every imported opening is new to this installation. A forged full or # one-space file must not use the broken-read exception reserved for an # already stored legacy passage. validate_opening_passages(incoming_config, validate_all=True) except OpeningPassageError as err: raise ImportFailure(err.code, str(err)) from err if document["kind"] == "full": dropped_marker_links = _transfer_dropped_marker_links(document) dropped_marker_links += _drop_invalid_import_marker_links(incoming_config) document["transfer"]["dropped_marker_links"] = dropped_marker_links try: validate_marker_controls(incoming_config, validate_all=True) validate_marker_light_entities(incoming_config, validate_all=True) validate_marker_radars(incoming_config, validate_all=True) validate_marker_value_badges(incoming_config, validate_all=True) validate_marker_vacuum_routes(incoming_config, validate_all=True) except MarkerControlError as err: raise ImportFailure(err.code, str(err)) from err except vol.Invalid as err: raise ImportFailure("invalid_config", str(err)) from err content, confirmation = _content_state(document, same_source, config_root) target_config, target_layout, details = _materialize_import_candidate( document, current_config, current_layout, duplicate_policy=duplicate_policy, same_source=same_source, content=content, confirmation_required=confirmation, ) token = secrets.token_urlsafe(32) now = time.time() candidate = { "owner_id": owner_id, "created": now, "expires": now + IMPORT_PREVIEW_TTL_S, "document": document, "duplicate_policy": duplicate_policy, "config_rev": int(current_config_data.get("rev", 0)), "layout_rev": int(current_layout_data.get("rev", 0)), "content": content, "confirmation_required": confirmation, "same_source": same_source, "target_config": target_config, "target_layout": target_layout, "details": details, } candidate["digest"] = _candidate_digest(candidate) # Opportunistic expiry and per-user oldest-first eviction. runtime.import_previews = { key: value for key, value in runtime.import_previews.items() if value.get("expires", 0) > now } mine = [key for key, value in runtime.import_previews.items() if value.get("owner_id") == owner_id] while len(mine) >= MAX_IMPORT_PREVIEWS_PER_USER: runtime.import_previews.pop(mine.pop(0), None) while len(runtime.import_previews) >= MAX_IMPORT_PREVIEWS_TOTAL: runtime.import_previews.pop(next(iter(runtime.import_previews)), None) runtime.import_previews[token] = candidate preview = { "kind": document["kind"], "plan_only": _transfer_plan_only(document), "created_at": document.get("created_at"), "card_version": document.get("card_version"), "integration_version": document.get("integration_version"), "model_version": document.get("model_version", 0), "source": "same" if same_source else "foreign", "counts": _counts(incoming_config, incoming_layout), "current_counts": _counts(current_config, current_layout), "legacy_positions": sum( 1 for pos in incoming_layout.values() if isinstance(pos, dict) and not pos.get("s") ), "bindings": _binding_preview(incoming_config, incoming_layout, document, registry_snapshot), "missing_areas": _missing_areas(incoming_config, registry_snapshot), "content": content, "confirmation_required": confirmation, "warnings": (["foreign_source"] if not same_source else []) + (["instance_bookkeeping_dropped"] if not same_source and document["kind"] == "full" else []) + (["content_detach_required"] if confirmation else []) + (["full_replaces_current"] if document["kind"] == "full" else []), **details, } return { "token": token, "preview": preview, "expected_config_rev": candidate["config_rev"], "expected_layout_rev": candidate["layout_rev"], "expires_at": datetime.fromtimestamp(candidate["expires"], UTC).isoformat().replace("+00:00", "Z"), } def get_candidate(runtime: HouseplanData, token: str, owner_id: str, *, consume: bool = False) -> dict[str, Any]: candidate = runtime.import_previews.get(token) if not candidate or candidate.get("expires", 0) <= time.time(): runtime.import_previews.pop(token, None) raise ImportFailure("preview_expired", "Import preview expired") if candidate.get("owner_id") != owner_id: raise ImportFailure("preview_owner_mismatch", "Preview belongs to another user") if candidate.get("digest") != _candidate_digest(candidate): runtime.import_previews.pop(token, None) raise ImportFailure("invalid_format", "Import preview candidate changed") if consume: runtime.import_previews.pop(token, None) return candidate def revalidate_candidate( candidate: dict[str, Any], current_config_data: dict[str, Any], current_layout_data: dict[str, Any], *, duplicate_policy: str, registry_snapshot: dict[str, set[str]] | None = None, config_root: Path | None = None, ) -> dict[str, Any]: """Refresh target-dependent preview data without uploading the file again.""" document = candidate["document"] current_config = current_config_data.get("config") or { "spaces": [], "markers": [], "settings": {}, } current_layout = current_layout_data.get("layout") or {} if config_root is not None: content, confirmation = _content_state( document, bool(candidate.get("same_source")), config_root ) candidate["content"] = content candidate["confirmation_required"] = confirmation target_config, target_layout, details = _materialize_import_candidate( document, current_config, current_layout, duplicate_policy=duplicate_policy, same_source=bool(candidate.get("same_source")), content=candidate.get("content") or [], confirmation_required=bool(candidate.get("confirmation_required")), ) candidate["duplicate_policy"] = duplicate_policy candidate["config_rev"] = int(current_config_data.get("rev", 0)) candidate["layout_rev"] = int(current_layout_data.get("rev", 0)) candidate["target_config"] = target_config candidate["target_layout"] = target_layout candidate["details"] = details candidate["digest"] = _candidate_digest(candidate) incoming = document["payload"] return { "preview": { "kind": document["kind"], "plan_only": _transfer_plan_only(document), "counts": _counts(incoming["config"], incoming["layout"]), "current_counts": _counts(current_config, current_layout), "source": "same" if candidate.get("same_source") else "foreign", "content": candidate.get("content") or [], "confirmation_required": candidate.get("confirmation_required", False), "bindings": _binding_preview( incoming["config"], incoming["layout"], candidate["document"], registry_snapshot ), "missing_areas": _missing_areas(incoming["config"], registry_snapshot), **details, }, "expected_config_rev": candidate["config_rev"], "expected_layout_rev": candidate["layout_rev"], } def prepare_apply( candidate: dict[str, Any], current_config: dict[str, Any], current_layout: dict[str, Any], *, duplicate_policy: str | None = None, confirm_missing_content: bool, ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: if candidate.get("confirmation_required") and not confirm_missing_content: raise ImportFailure("content_confirmation_required", "Missing content must be confirmed") if duplicate_policy is not None and duplicate_policy != candidate.get("duplicate_policy"): raise ImportFailure("conflict", "Duplicate policy changed after the preview") if candidate.get("digest") != _candidate_digest(candidate): raise ImportFailure("invalid_format", "Import preview candidate changed") config = candidate.get("target_config") layout = candidate.get("target_layout") details = candidate.get("details") if not isinstance(config, dict) or not isinstance(layout, dict) \ or not isinstance(details, dict): raise ImportFailure("invalid_format", "Import preview has no materialized candidate") # The current stores are intentionally unused: revisions are checked by # the websocket handler under the write lock before this function runs. _ = current_config, current_layout return _json_copy(config), _json_copy(layout), _json_copy(details)