bahmcloud/bahmcloud_store
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 34 Wiki Activity

Fix restore version

Browse Source
This commit is contained in:
René Bachmann
2026-01-18 09:08:25 +00:00
parent 2eb194c001
commit f292e22301
1 changed files with 138 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
custom_components/bahmcloud_store/core.py
View File

@@ -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:<backup_id>) 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: