78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.storage import Store
|
|
|
|
_STORAGE_VERSION = 1
|
|
_STORAGE_KEY = "bcs_store"
|
|
|
|
|
|
@dataclass
|
|
class CustomRepo:
|
|
id: str
|
|
url: str
|
|
name: str | None = None
|
|
|
|
|
|
class BCSStorage:
|
|
"""Persistent storage for manually added repositories."""
|
|
|
|
def __init__(self, hass: HomeAssistant) -> None:
|
|
self.hass = hass
|
|
self._store = Store(hass, _STORAGE_VERSION, _STORAGE_KEY)
|
|
|
|
async def _load(self) -> dict[str, Any]:
|
|
data = await self._store.async_load()
|
|
if not data:
|
|
return {"custom_repos": []}
|
|
if "custom_repos" not in data:
|
|
data["custom_repos"] = []
|
|
return data
|
|
|
|
async def _save(self, data: dict[str, Any]) -> None:
|
|
await self._store.async_save(data)
|
|
|
|
async def list_custom_repos(self) -> list[CustomRepo]:
|
|
data = await self._load()
|
|
repos = data.get("custom_repos", [])
|
|
out: list[CustomRepo] = []
|
|
for r in repos:
|
|
if not isinstance(r, dict):
|
|
continue
|
|
rid = str(r.get("id") or "")
|
|
url = str(r.get("url") or "")
|
|
name = r.get("name")
|
|
if rid and url:
|
|
out.append(CustomRepo(id=rid, url=url, name=str(name) if name else None))
|
|
return out
|
|
|
|
async def add_custom_repo(self, url: str, name: str | None) -> CustomRepo:
|
|
data = await self._load()
|
|
repos = data.get("custom_repos", [])
|
|
|
|
# Deduplicate by URL
|
|
for r in repos:
|
|
if isinstance(r, dict) and str(r.get("url", "")).strip() == url.strip():
|
|
# Update name if provided
|
|
if name:
|
|
r["name"] = name
|
|
await self._save(data)
|
|
return CustomRepo(id=str(r["id"]), url=str(r["url"]), name=r.get("name"))
|
|
|
|
rid = f"custom:{uuid.uuid4().hex[:10]}"
|
|
entry = {"id": rid, "url": url.strip(), "name": name.strip() if name else None}
|
|
repos.append(entry)
|
|
data["custom_repos"] = repos
|
|
await self._save(data)
|
|
return CustomRepo(id=rid, url=entry["url"], name=entry["name"])
|
|
|
|
async def remove_custom_repo(self, repo_id: str) -> None:
|
|
data = await self._load()
|
|
repos = data.get("custom_repos", [])
|
|
data["custom_repos"] = [r for r in repos if not (isinstance(r, dict) and r.get("id") == repo_id)]
|
|
await self._save(data)
|