diff --git a/custom_components/bahmcloud_store/providers.py b/custom_components/bahmcloud_store/providers.py new file mode 100644 index 0000000..5449b1f --- /dev/null +++ b/custom_components/bahmcloud_store/providers.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import logging +from dataclasses import dataclass +from urllib.parse import urlparse + +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession + +_LOGGER = logging.getLogger(__name__) + + +@dataclass +class RepoInfo: + owner: str | None = None + description: str | None = None + provider: str | None = None + + +def _split_owner_repo(repo_url: str) -> tuple[str | None, str | None]: + u = urlparse(repo_url.rstrip("/")) + parts = [p for p in u.path.strip("/").split("/") if p] + if len(parts) < 2: + return None, None + return parts[0], parts[1] + + +def detect_provider(repo_url: str) -> str: + """Best-effort provider detection by hostname.""" + host = urlparse(repo_url).netloc.lower() + if "github.com" in host: + return "github" + if "gitlab.com" in host: + return "gitlab" + # Default for self-hosted setups: assume Gitea when URL looks like owner/repo + owner, repo = _split_owner_repo(repo_url) + if owner and repo: + return "gitea" + return "generic" + + +async def fetch_repo_info(hass: HomeAssistant, repo_url: str) -> RepoInfo: + """ + Fetch owner/description from a provider API if possible. + This is best-effort and must never break the store if it fails. + """ + provider = detect_provider(repo_url) + owner, repo = _split_owner_repo(repo_url) + + info = RepoInfo(owner=owner, description=None, provider=provider) + + if not owner or not repo: + return info + + session = async_get_clientsession(hass) + + try: + if provider == "github": + api = f"https://api.github.com/repos/{owner}/{repo}" + async with session.get( + api, + timeout=15, + headers={"Accept": "application/vnd.github+json"}, + ) as resp: + if resp.status != 200: + _LOGGER.debug("GitHub repo info failed (%s): %s", resp.status, api) + return info + data = await resp.json() + info.description = data.get("description") + # Owner returned by API is canonical + if isinstance(data.get("owner"), dict) and data["owner"].get("login"): + info.owner = data["owner"]["login"] + return info + + if provider == "gitea": + u = urlparse(repo_url.rstrip("/")) + base = f"{u.scheme}://{u.netloc}" + api = f"{base}/api/v1/repos/{owner}/{repo}" + async with session.get(api, timeout=15) as resp: + if resp.status != 200: + _LOGGER.debug("Gitea repo info failed (%s): %s", resp.status, api) + return info + data = await resp.json() + info.description = data.get("description") + if isinstance(data.get("owner"), dict) and data["owner"].get("login"): + info.owner = data["owner"]["login"] + return info + + # GitLab and generic providers will be implemented later + return info + + except Exception as e: + _LOGGER.debug("Provider fetch failed for %s: %s", repo_url, e) + return info