From f292e223012a0a6b8779a94755c433c180aa95e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ren=C3=A9=20Bachmann?= Date: Sun, 18 Jan 2026 09:08:25 +0000 Subject: [PATCH] Fix restore version --- custom_components/bahmcloud_store/core.py | 142 +++++++++++++++++++++- 1 file changed, 138 insertions(+), 4 deletions(-) diff --git a/custom_components/bahmcloud_store/core.py b/custom_components/bahmcloud_store/core.py index 0afdb61..b4ab03c 100644 --- a/custom_components/bahmcloud_store/core.py +++ b/custom_components/bahmcloud_store/core.py @@ -30,6 +30,8 @@ DOMAIN = "bahmcloud_store" SIGNAL_UPDATED = f"{DOMAIN}_updated" RESTART_REQUIRED_ISSUE_ID = "restart_required" +BACKUP_META_FILENAME = ".bcs_backup_meta.json" + class BCSError(Exception): """BCS core error.""" @@ -493,7 +495,19 @@ class BCSCore: await self.hass.async_add_executor_job(_mkdir) - async def _backup_domain(self, domain: str) -> Path | None: + 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 {} + return { + "repo_id": repo_id, + "domain": domain, + "installed_version": inst.get("installed_version") or inst.get("ref"), + "installed_manifest_version": inst.get("installed_manifest_version"), + "ref": inst.get("ref") or inst.get("installed_version"), + "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"), + } + + async def _backup_domain(self, domain: str, *, meta: dict[str, object] | None = None) -> Path | None: """Backup an existing domain folder. Returns the created backup path, or None if the domain folder does not exist. @@ -515,6 +529,13 @@ class BCSCore: if backup_path.exists(): shutil.rmtree(backup_path, ignore_errors=True) shutil.copytree(target, backup_path, dirs_exist_ok=True) + # Store backup metadata (kept inside backup folder; removed from target after restore). + 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 # Retention: keep only the newest N backups per domain. try: @@ -541,6 +562,28 @@ class BCSCore: if target.exists(): shutil.rmtree(target, ignore_errors=True) shutil.copytree(backup_path, target, dirs_exist_ok=True) + try: + meta_file = target / BACKUP_META_FILENAME + if meta_file.exists(): + meta_file.unlink(missing_ok=True) + except Exception: + pass + + # Do not leave backup metadata inside the restored integration folder. + try: + meta_p = target / BACKUP_META_FILENAME + if meta_p.exists(): + meta_p.unlink() + except Exception: + pass + + # Do not leave backup metadata inside the live integration folder. + try: + m = target / BACKUP_META_FILENAME + if m.exists(): + m.unlink() + except Exception: + pass await self.hass.async_add_executor_job(_restore) _LOGGER.info("BCS rollback applied: domain=%s from=%s", domain, backup_path) @@ -579,7 +622,13 @@ class BCSCore: for bid, present in id_map.items(): complete = present == all_domains label = self._format_backup_id(bid) - items.append({"id": bid, "label": label, "complete": complete, "domains": sorted(present)}) + meta = await self._read_backup_meta(dom_list[0], bid) + ver = None + if isinstance(meta, dict): + ver = meta.get("installed_version") or meta.get("ref") + if ver: + label = f"{label} ({ver})" + items.append({"id": bid, "label": label, "complete": complete, "domains": sorted(present), "installed_version": str(ver) if ver else None}) # Sort newest first by id (lexicographic works for timestamp format). items.sort(key=lambda x: str(x.get("id") or ""), reverse=True) @@ -631,12 +680,87 @@ class BCSCore: for d in dom_list: await self._restore_domain_from_backup(d, self._backup_root / d / backup_id) + # Update stored installed version to the restored one (so UI shows the restored state). + # + # Backups created before 0.6.1 may not have metadata. For those legacy backups we fall back to: + # 1) version from the backup's manifest.json (best-effort), else + # 2) a synthetic marker (restored:) so the UI reflects a restored state and updates + # remain available. + restored_meta = await self._read_backup_meta(dom_list[0], backup_id) + restored_version: str | None = None + restored_manifest_version: str | None = None + + if isinstance(restored_meta, dict): + rv = restored_meta.get("installed_version") or restored_meta.get("ref") + if rv is not None and str(rv).strip(): + restored_version = str(rv).strip() + mv = restored_meta.get("installed_manifest_version") + if mv is not None and str(mv).strip(): + restored_manifest_version = str(mv).strip() + + # Legacy backups (no meta): try to read manifest.json version from the backup folder. + if not restored_manifest_version: + restored_manifest_version = await self._read_backup_manifest_version(dom_list[0], backup_id) + + # Use manifest version as a fallback display value if we don't have the exact installed ref. + if not restored_version and restored_manifest_version: + restored_version = restored_manifest_version + + # Last resort: ensure the installed version changes so the UI does not keep showing the newest version. + if not restored_version: + restored_version = f"restored:{backup_id}" + + repo = self.get_repo(repo_id) + repo_url = getattr(repo, "url", None) or "" + + await self.storage.set_installed_repo( + repo_id=repo_id, + url=repo_url, + domains=dom_list, + installed_version=restored_version, + installed_manifest_version=restored_manifest_version, + ref=restored_version, + ) + await self._refresh_installed_cache() self._mark_restart_required() self.signal_updated() _LOGGER.info("BCS restore complete: repo_id=%s backup_id=%s", repo_id, backup_id) - return {"ok": True, "repo_id": repo_id, "backup_id": backup_id, "domains": dom_list, "restart_required": True} + return {"ok": True, "repo_id": repo_id, "backup_id": backup_id, "domains": dom_list, "restored_version": restored_version, "restart_required": True} + + async def _read_backup_meta(self, domain: str, backup_id: str) -> dict[str, Any] | None: + """Read backup metadata for a domain backup. + + Metadata is stored inside the backup folder and will be removed from the + live folder after restore. + """ + try: + p = self._backup_root / domain / backup_id / BACKUP_META_FILENAME + if not p.exists(): + return None + txt = await self.hass.async_add_executor_job(p.read_text, 'utf-8') + data = json.loads(txt) + return data if isinstance(data, dict) else None + except Exception: + return None + + + async def _read_backup_manifest_version(self, domain: str, backup_id: str) -> str | None: + """Best-effort: read manifest.json version from a legacy backup (no metadata).""" + + def _read() -> str | None: + try: + p = self._backup_root / domain / backup_id / 'manifest.json' + if not p.exists(): + return None + data = json.loads(p.read_text(encoding='utf-8')) + v = data.get('version') + return str(v) if v else None + except Exception: + return None + + return await self.hass.async_add_executor_job(_read) async def _list_domain_backup_ids(self, domain: str) -> list[str]: """List backup ids for a domain (newest->oldest).""" @@ -802,6 +926,14 @@ class BCSCore: installed_domains: list[str] = [] backups: dict[str, Path] = {} + + inst_before = self.get_installed(repo_id) or {} + backup_meta = { + "repo_id": repo_id, + "installed_version": inst_before.get("installed_version") or inst_before.get("ref"), + "installed_manifest_version": inst_before.get("installed_manifest_version"), + "ref": inst_before.get("ref") or inst_before.get("installed_version"), + } created_new: set[str] = set() try: @@ -832,7 +964,9 @@ class BCSCore: # Backup only if we are going to overwrite an existing domain. if target.exists() and target.is_dir(): - bkp = await self._backup_domain(domain) + m = dict(backup_meta) + m["domain"] = domain + bkp = await self._backup_domain(domain, meta=m) if bkp: backups[domain] = bkp else: