custom_components/bahmcloud_store/providers.py hinzugefügt

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