| """Restore and upload a compact Home Assistant configuration snapshot.""" |
| from __future__ import annotations |
| import argparse |
| import os |
| import sys |
| import tarfile |
| import tempfile |
| from pathlib import Path |
|
|
| ARCHIVE_NAME = "homeassistant-latest.tar.gz" |
|
|
|
|
| def hub_api(): |
| from huggingface_hub import HfApi |
| token = os.environ.get("HF_TOKEN") |
| repo_id = os.environ.get("BACKUP_REPO_ID") |
| if not token or not repo_id: |
| raise RuntimeError("HF_TOKEN and BACKUP_REPO_ID are required for Hub backups") |
| api = HfApi(token=token) |
| repo_type = os.environ.get("BACKUP_REPO_TYPE", "dataset") |
| try: |
| api.repo_info(repo_id=repo_id, repo_type=repo_type) |
| except Exception: |
| api.create_repo(repo_id=repo_id, repo_type=repo_type, private=True, exist_ok=True) |
| return api, repo_id, repo_type, token |
|
|
|
|
| def archive_config(config: Path, archive: Path) -> None: |
| with tarfile.open(archive, "w:gz") as tar: |
| for path in config.rglob("*"): |
| if not path.is_file() or path.is_symlink(): continue |
| relative = path.relative_to(config).as_posix() |
| if relative in {"home-assistant_v2.db", "home-assistant.log"}: continue |
| if relative.endswith((".db-wal", ".db-shm")): continue |
| if relative.startswith((".cloud/", "tts/")): continue |
| tar.add(path, arcname=relative, recursive=False) |
|
|
|
|
| def backup(config: Path) -> None: |
| api, repo_id, repo_type, token = hub_api() |
| with tempfile.TemporaryDirectory() as temp_dir: |
| archive = Path(temp_dir) / ARCHIVE_NAME |
| archive_config(config, archive) |
| api.upload_file(path_or_fileobj=str(archive), path_in_repo=ARCHIVE_NAME, |
| repo_id=repo_id, repo_type=repo_type, token=token, |
| commit_message="Update Home Assistant configuration backup") |
| print(f"Uploaded {ARCHIVE_NAME} to {repo_type}/{repo_id}", flush=True) |
|
|
|
|
| def restore(config: Path) -> None: |
| from huggingface_hub.errors import EntryNotFoundError |
| try: |
| _, repo_id, repo_type, token = hub_api() |
| from huggingface_hub import hf_hub_download |
| archive = hf_hub_download(repo_id=repo_id, filename=ARCHIVE_NAME, |
| repo_type=repo_type, token=token, force_download=True) |
| except EntryNotFoundError: |
| print("No existing Hub snapshot; starting a new configuration.", flush=True) |
| return |
| except Exception as exc: |
| raise RuntimeError("Hub restore failed; refusing to overwrite the remote backup with a fresh configuration.") from exc |
| config.mkdir(parents=True, exist_ok=True) |
| with tarfile.open(archive, "r:gz") as tar: |
| root = config.resolve() |
| for member in tar.getmembers(): |
| target = (config / member.name).resolve() |
| if root not in target.parents: raise RuntimeError(f"Unsafe backup path: {member.name}") |
| tar.extractall(config, filter="data") |
| print(f"Restored {ARCHIVE_NAME} into {config}", flush=True) |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("action", choices=("backup", "restore")) |
| parser.add_argument("--config", required=True, type=Path) |
| args = parser.parse_args() |
| if not os.environ.get("HF_TOKEN") or not os.environ.get("BACKUP_REPO_ID"): return 0 |
| (backup if args.action == "backup" else restore)(args.config) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": raise SystemExit(main()) |
