"""Export maintenance data as JSON or YAML.""" from __future__ import annotations import json import logging from pathlib import Path from typing import Any from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from .const import ( CONF_OBJECT, CONF_TASKS, DEFAULT_WARNING_DAYS, DOMAIN, GLOBAL_UNIQUE_ID, ) from .helpers.schedule import Schedule, read_legacy_fields _LOGGER = logging.getLogger(__name__) def _export_documents(doc_store: Any, object_id: str) -> list[dict[str, Any]]: """Export an object's document metadata + web-links (blobs ride the backup). File binaries are NOT in the JSON export (they live under /config and travel via the HA backup). An import without a matching backup therefore recreates file metadata pointing at a missing blob — the storage-hygiene repair issue catches those as dangling. Web-links round-trip fully. ``task_ids`` are carried too (as the OLD task ids); the importer remaps them onto the fresh task ids so a doc's task links survive a backup/restore. """ out: list[dict[str, Any]] = [] for d in doc_store.for_object(object_id): if d.get("kind") == "weblink": out.append( { "kind": "weblink", "url": d.get("url"), "title": d.get("title"), "tags": d.get("tags") or [], "task_ids": d.get("task_ids") or [], "part_ids": d.get("part_ids") or [], } ) else: out.append( { "kind": "file", "hash": d.get("hash"), "title": d.get("title"), "filename": d.get("filename"), "mime": d.get("mime"), "size": d.get("size"), "tags": d.get("tags") or [], "task_ids": d.get("task_ids") or [], "part_ids": d.get("part_ids") or [], } ) return out def _build_export_object( hass: HomeAssistant, entry: ConfigEntry, coordinator_data: dict[str, Any] | None, include_history: bool, ) -> dict[str, Any]: """Build a single object's export dict.""" obj_data = entry.data.get(CONF_OBJECT, {}) # Merge static + Store dynamic data for each task rd = getattr(entry, "runtime_data", None) store = getattr(rd, "store", None) if rd else None static_tasks = entry.data.get(CONF_TASKS, {}) tasks_data = store.merge_all_tasks(static_tasks) if store is not None else static_tasks ct_tasks = (coordinator_data or {}).get(CONF_TASKS, {}) tasks = [] for tid, tdata in tasks_data.items(): ct = ct_tasks.get(tid, {}) sched = read_legacy_fields(tdata) task: dict[str, Any] = { "id": tid, "name": tdata.get("name", ""), "type": tdata.get("type", "custom"), "enabled": tdata.get("enabled", True), # Provenance + lifecycle a backup must preserve: created_at is the # next_due fallback anchor when last_performed is None (dropping it # silently shifts due dates on restore), and the archived_* pair # keeps a retired task retired — without them an archived task comes # back ACTIVE after a restore. "created_at": tdata.get("created_at"), "archived_at": tdata.get("archived_at"), "archived_reason": tdata.get("archived_reason"), "schedule_type": sched["schedule_type"], "interval_days": sched["interval_days"], "interval_unit": sched["interval_unit"], "due_date": sched["due_date"], "interval_anchor": sched["interval_anchor"], # Nested recurrence — carries the calendar kinds (weekdays / # nth_weekday / day_of_month) that the flat fields above can't. "schedule": Schedule.parse(tdata).to_dict(), "last_planned_due": tdata.get("last_planned_due"), # A pending per-occurrence postpone is user intent — round-trip it # like last_planned_due so a backup/restore keeps the deferral. "due_override": tdata.get("due_override"), "warning_days": tdata.get("warning_days", DEFAULT_WARNING_DAYS), "last_performed": tdata.get("last_performed"), "notes": tdata.get("notes"), "documentation_url": tdata.get("documentation_url"), "custom_icon": tdata.get("custom_icon"), "nfc_tag_id": tdata.get("nfc_tag_id"), "responsible_user_id": tdata.get("responsible_user_id"), "entity_slug": tdata.get("entity_slug"), "adaptive_config": tdata.get("adaptive_config"), "checklist": tdata.get("checklist") or [], "schedule_time": tdata.get("schedule_time"), # v2.17+ / #83 task fields — persisted and user-facing, so a JSON # backup must restore them (same field-completeness contract as #67 # for documentation_url/notes; import mirrors these keys). "priority": tdata.get("priority", "normal"), "labels": tdata.get("labels") or [], "earliest_completion_days": tdata.get("earliest_completion_days"), "on_complete_action": tdata.get("on_complete_action"), "quick_complete_defaults": tdata.get("quick_complete_defaults"), "assignee_pool": tdata.get("assignee_pool") or [], "rotation_strategy": tdata.get("rotation_strategy"), "reading_unit": tdata.get("reading_unit"), # Spare parts: consumption links + the auto-buy-task marker. "consumes_parts": tdata.get("consumes_parts"), "part_ref": tdata.get("part_ref"), "status": ct.get("_status", "ok"), "days_until_due": ct.get("_days_until_due"), "next_due": ct.get("_next_due"), "times_performed": ct.get("_times_performed", 0), "total_cost": ct.get("_total_cost", 0.0), "average_duration": ct.get("_average_duration"), } trigger_config = tdata.get("trigger_config") if trigger_config: task["trigger_config"] = trigger_config if include_history: task["history"] = tdata.get("history") or [] tasks.append(task) # (roadmap P6) attach document metadata + web-links. from . import DOCUMENT_STORE_KEY doc_store = hass.data.get(DOMAIN, {}).get(DOCUMENT_STORE_KEY) object_id = obj_data.get("id", "") documents = _export_documents(doc_store, object_id) if doc_store is not None and object_id else [] return { "entry_id": entry.entry_id, "object": { "name": obj_data.get("name", ""), "area_id": obj_data.get("area_id"), "manufacturer": obj_data.get("manufacturer"), "model": obj_data.get("model"), "serial_number": obj_data.get("serial_number"), "installation_date": obj_data.get("installation_date"), "warranty_expiry": obj_data.get("warranty_expiry"), # Round-tripped so a JSON backup restores the full asset record # (these were added in v1.4.0/v1.4.10 but missed here until #67). "documentation_url": obj_data.get("documentation_url"), "notes": obj_data.get("notes"), # 2.19: device link / parent hierarchy. Instance-specific ids — # meaningful when restoring on the SAME instance; dangling values # on a foreign instance are harmless (device_info falls back). "ha_device_id": obj_data.get("ha_device_id"), "parent_entry_id": obj_data.get("parent_entry_id"), # 2.20: seasonal pause (a paused pool restored in winter stays # paused) + replace-flow lineage (instance-specific entry ids, # same caveat as parent_entry_id above). "paused_at": obj_data.get("paused_at"), "paused_until": obj_data.get("paused_until"), "predecessor_entry_id": obj_data.get("predecessor_entry_id"), "replaced_by_entry_id": obj_data.get("replaced_by_entry_id"), }, "tasks": tasks, "documents": documents, # Spare parts: full static definition + the tracked stock (dynamic, # read from the Store) so a backup/restore keeps the shelf state. "parts": [ {**part, "stock": store.get_part_stock(part["id"]) if store is not None else None} for part in (entry.data.get("parts") or {}).values() ], } def object_entries(hass: HomeAssistant, entry_ids: set[str] | None = None) -> list[ConfigEntry]: """The maintenance OBJECT entries (never the global hub), optionally narrowed to a selection. Shared by every exporter so JSON/YAML/CSV apply the same selective-export filter. ``entry_ids=None`` means all objects.""" return [ entry for entry in hass.config_entries.async_entries(DOMAIN) if entry.unique_id != GLOBAL_UNIQUE_ID and (entry_ids is None or entry.entry_id in entry_ids) ] def build_export_data( hass: HomeAssistant, include_history: bool = True, entry_ids: set[str] | None = None, ) -> dict[str, Any]: """Gather all maintenance data into a plain dict. This must be called from the event loop (accesses HA APIs). The returned dict contains no HA objects and is safe to serialize in an executor thread. ``entry_ids`` narrows the export to a selection of objects (None = all). """ entries = object_entries(hass, entry_ids) objects = [] for entry in entries: rd = getattr(entry, "runtime_data", None) coord_data = rd.coordinator.data if rd and rd.coordinator else None objects.append(_build_export_object(hass, entry, coord_data, include_history)) return { "version": 1, "objects": objects, } def serialize_export(data: dict[str, Any], fmt: str = "json") -> str: """Serialize an export data dict to a JSON or YAML string. Pure function with no HA dependencies — safe to run in an executor. """ if fmt == "yaml": try: import yaml # type: ignore[import-untyped] # Normalize through JSON first: yaml.safe_dump rejects types the # JSON path coerces (e.g. tuples → lists), so YAML export would # crash on data JSON handles fine. Round-tripping keeps both # formats consistent and YAML-safe. normalized = json.loads(json.dumps(data, ensure_ascii=False)) return str(yaml.safe_dump(normalized, default_flow_style=False, allow_unicode=True)) except ImportError: _LOGGER.warning("PyYAML not available, falling back to JSON") return json.dumps(data, indent=2, ensure_ascii=False) return json.dumps(data, indent=2, ensure_ascii=False) def serialize_export_to_file(data: dict[str, Any], fmt: str, file_path: str) -> str: """Serialize export data and write to a file. Pure sync function — safe to run in an executor via ``hass.async_add_executor_job``. Returns: The file path written to. """ content = serialize_export(data, fmt) Path(file_path).write_text(content, encoding="utf-8") return file_path def export_maintenance_data( hass: HomeAssistant, fmt: str = "json", include_history: bool = True, ) -> str: """Export all maintenance data as a JSON or YAML string. Legacy convenience wrapper used by the WebSocket export handler. For the service handler, prefer ``build_export_data`` + ``serialize_export_to_file`` (via executor) to avoid blocking the event loop. """ data = build_export_data(hass, include_history=include_history) return serialize_export(data, fmt)