diff --git a/.idea/changes.md b/.idea/changes.md index 21a4208..3bdd5c9 100644 --- a/.idea/changes.md +++ b/.idea/changes.md @@ -19,6 +19,8 @@ - Started the blueprint/category-aware installer preparation: installation metadata now distinguishes install type and installed paths, repo payloads expose install targets, and the active panel shows install-target context for future non-integration categories. - Added initial blueprint install-path handling groundwork so the codebase is no longer fully hard-wired to `custom_components/`. - Bumped the integration version from `0.7.4` to `0.7.5` and added the `0.7.5` release entry to `CHANGELOG.md` for blueprint support and the documentation refresh. +- Fixed blueprint update backups: blueprint file updates now create content backups before overwrite and can roll back copied blueprint files if installation fails. +- Kept the no-restart behavior for blueprints, because blueprint deployment does not normally require a Home Assistant restart. ### Documented - Captured the verified project identity from the repository and README files: Bahmcloud Store is a Home Assistant custom integration intended to behave like a provider-neutral store for custom integrations, similar to HACS but broader than GitHub-only workflows. diff --git a/custom_components/bahmcloud_store/core.py b/custom_components/bahmcloud_store/core.py index 3e372e2..335c58d 100644 --- a/custom_components/bahmcloud_store/core.py +++ b/custom_components/bahmcloud_store/core.py @@ -1335,6 +1335,11 @@ class BCSCore: await self.hass.async_add_executor_job(_mkdir) + @staticmethod + def _backup_repo_key(repo_id: str) -> str: + safe = "".join(ch if ch.isalnum() or ch in ("-", "_") else "_" for ch in str(repo_id or "").strip()) + return safe or "repo" + def _build_backup_meta(self, repo_id: str, domain: str) -> dict[str, object]: """Build metadata for backup folders so restores can recover the stored version.""" inst = self.get_installed(repo_id) or {} @@ -1391,6 +1396,68 @@ class BCSCore: _LOGGER.info("BCS backup created: domain=%s path=%s", domain, backup_path) return backup_path + async def _backup_paths( + self, + repo_id: str, + relative_paths: list[str], + *, + meta: dict[str, object] | None = None, + ) -> Path | None: + """Backup arbitrary files under the Home Assistant config root.""" + + clean_paths = [str(p).strip().replace("\\", "/") for p in (relative_paths or []) if str(p).strip()] + if not clean_paths: + return None + + cfg_root = Path(self.hass.config.path("")) + existing: list[str] = [] + for rel in clean_paths: + target = cfg_root / rel + if target.exists(): + existing.append(rel) + if not existing: + return None + + await self._ensure_backup_root() + + ts = time.strftime("%Y%m%d_%H%M%S") + repo_root = self._backup_root / "_content" / self._backup_repo_key(repo_id) + backup_path = repo_root / ts + + def _do_backup() -> None: + repo_root.mkdir(parents=True, exist_ok=True) + if backup_path.exists(): + shutil.rmtree(backup_path, ignore_errors=True) + backup_path.mkdir(parents=True, exist_ok=True) + + for rel in existing: + src = cfg_root / rel + dest = backup_path / rel + dest.parent.mkdir(parents=True, exist_ok=True) + if src.is_file(): + shutil.copy2(src, dest) + elif src.is_dir(): + shutil.copytree(src, dest, dirs_exist_ok=True) + + if meta: + try: + meta_path = backup_path / BACKUP_META_FILENAME + meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") + except Exception: + pass + + try: + backups = [p for p in repo_root.iterdir() if p.is_dir()] + backups.sort(key=lambda p: p.name, reverse=True) + for old in backups[self._backup_keep_per_domain :]: + shutil.rmtree(old, ignore_errors=True) + except Exception: + pass + + await self.hass.async_add_executor_job(_do_backup) + _LOGGER.info("BCS content backup created: repo_id=%s path=%s files=%s", repo_id, backup_path, len(existing)) + return backup_path + async def _restore_domain_from_backup(self, domain: str, backup_path: Path) -> None: """Restore a domain folder from a backup.""" dest_root = Path(self.hass.config.path("custom_components")) @@ -1428,6 +1495,37 @@ class BCSCore: await self.hass.async_add_executor_job(_restore) _LOGGER.info("BCS rollback applied: domain=%s from=%s", domain, backup_path) + async def _restore_paths_from_backup(self, backup_path: Path, *, remove_targets: list[str] | None = None) -> None: + """Restore arbitrary backed up files under the Home Assistant config root.""" + + cfg_root = Path(self.hass.config.path("")) + remove_list = [str(p).strip().replace("\\", "/") for p in (remove_targets or []) if str(p).strip()] + + def _restore() -> None: + if not backup_path.exists() or not backup_path.is_dir(): + return + + for rel in remove_list: + target = cfg_root / rel + if target.exists(): + if target.is_dir(): + shutil.rmtree(target, ignore_errors=True) + elif target.is_file(): + target.unlink(missing_ok=True) + + for src in backup_path.rglob("*"): + if not src.is_file(): + continue + if src.name == BACKUP_META_FILENAME: + continue + rel = src.relative_to(backup_path) + dest = cfg_root / rel + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(src, dest) + + await self.hass.async_add_executor_job(_restore) + _LOGGER.info("BCS content rollback applied: from=%s", backup_path) + async def list_repo_backups(self, repo_id: str) -> list[dict[str, Any]]: """List available backup sets for an installed repository. @@ -1806,6 +1904,7 @@ class BCSCore: installed_domains: list[str] = [] installed_paths: list[str] = [] backups: dict[str, Path] = {} + content_backup: Path | None = None install_type = self._repo_install_type(repo) inst_before = self.get_installed(repo_id) or {} @@ -1832,8 +1931,22 @@ class BCSCore: if not blueprints_root: raise BCSInstallError("blueprints folder not found in repository ZIP") - config_root = Path(self.hass.config.path("")) target_root = Path(self.hass.config.path("blueprints")) + planned_paths: list[str] = [] + + for src in blueprints_root.rglob("*"): + if not src.is_file(): + continue + rel = src.relative_to(blueprints_root) + planned_paths.append(str(Path("blueprints") / rel).replace("\\", "/")) + + if not planned_paths: + raise BCSInstallError("No blueprint files found under blueprints/") + + m = dict(backup_meta) + m["install_type"] = "blueprint" + m["installed_paths"] = planned_paths + content_backup = await self._backup_paths(repo_id, planned_paths, meta=m) def _copy_blueprints() -> list[str]: copied: list[str] = [] @@ -1849,8 +1962,6 @@ class BCSCore: return copied installed_paths = await self.hass.async_add_executor_job(_copy_blueprints) - if not installed_paths: - raise BCSInstallError("No blueprint files found under blueprints/") else: cc_root = self._find_custom_components_root(extract_dir) if not cc_root: @@ -1934,6 +2045,12 @@ class BCSCore: except Exception: _LOGGER.debug("BCS rollback failed for domain=%s", domain, exc_info=True) + if content_backup is not None: + try: + await self._restore_paths_from_backup(content_backup, remove_targets=installed_paths) + except Exception: + _LOGGER.debug("BCS content rollback failed for repo_id=%s", repo_id, exc_info=True) + # Remove newly created domains if the install did not complete. for domain in created_new: try: