custom_components/bahmcloud_store/providers.py aktualisiert

custom_components/bahmcloud_store/providers.py

@@ -18,6 +18,10 @@ class RepoInfo:
     provider: str | None = None
     default_branch: str | None = None
 
+    # Latest version best-effort
+    latest_version: str | None = None
+    latest_version_source: str | None = None  # "release" | "tag" | None
+
 
 def _split_owner_repo(repo_url: str) -> tuple[str | None, str | None]:
     u = urlparse(repo_url.rstrip("/"))
@@ -39,10 +43,87 @@ def detect_provider(repo_url: str) -> str:
     return "generic"
 
 
+async def _safe_json(session, url: str, *, headers: dict | None = None, timeout: int = 15):
+    try:
+        async with session.get(url, timeout=timeout, headers=headers) as resp:
+            if resp.status != 200:
+                return None, resp.status
+            return await resp.json(), resp.status
+    except Exception:
+        return None, None
+
+
+async def _github_latest_version(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]:
+    session = async_get_clientsession(hass)
+
+    # 1) Latest release
+    data, status = await _safe_json(
+        session,
+        f"https://api.github.com/repos/{owner}/{repo}/releases/latest",
+        headers={"Accept": "application/vnd.github+json"},
+        timeout=15,
+    )
+    if isinstance(data, dict):
+        tag = data.get("tag_name") or data.get("name")
+        if isinstance(tag, str) and tag.strip():
+            return tag.strip(), "release"
+
+    # 2) Latest tag (first item)
+    data, status = await _safe_json(
+        session,
+        f"https://api.github.com/repos/{owner}/{repo}/tags?per_page=1",
+        headers={"Accept": "application/vnd.github+json"},
+        timeout=15,
+    )
+    if isinstance(data, list) and data:
+        tag = data[0].get("name")
+        if isinstance(tag, str) and tag.strip():
+            return tag.strip(), "tag"
+
+    return None, None
+
+
+async def _gitea_latest_version(hass: HomeAssistant, base: str, owner: str, repo: str) -> tuple[str | None, str | None]:
+    session = async_get_clientsession(hass)
+
+    # 1) Releases (first item)
+    data, status = await _safe_json(
+        session,
+        f"{base}/api/v1/repos/{owner}/{repo}/releases?limit=1",
+        timeout=15,
+    )
+    if isinstance(data, list) and data:
+        tag = data[0].get("tag_name") or data[0].get("name")
+        if isinstance(tag, str) and tag.strip():
+            return tag.strip(), "release"
+
+    # 2) Tags (first item)
+    data, status = await _safe_json(
+        session,
+        f"{base}/api/v1/repos/{owner}/{repo}/tags?limit=1",
+        timeout=15,
+    )
+    if isinstance(data, list) and data:
+        tag = data[0].get("name")
+        if isinstance(tag, str) and tag.strip():
+            return tag.strip(), "tag"
+
+    return None, None
+
+
 async def fetch_repo_info(hass: HomeAssistant, repo_url: str) -> RepoInfo:
     provider = detect_provider(repo_url)
     owner, repo = _split_owner_repo(repo_url)
-    info = RepoInfo(owner=owner, repo_name=repo, description=None, provider=provider, default_branch=None)
+
+    info = RepoInfo(
+        owner=owner,
+        repo_name=repo,
+        description=None,
+        provider=provider,
+        default_branch=None,
+        latest_version=None,
+        latest_version_source=None,
+    )
 
     if not owner or not repo:
         return info
@@ -52,33 +133,41 @@ async def fetch_repo_info(hass: HomeAssistant, repo_url: str) -> RepoInfo:
     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()
+            data, status = await _safe_json(
+                session,
+                api,
+                headers={"Accept": "application/vnd.github+json"},
+                timeout=15,
+            )
+            if isinstance(data, dict):
                 info.description = data.get("description")
                 info.repo_name = data.get("name") or repo
                 info.default_branch = data.get("default_branch") or "main"
                 if isinstance(data.get("owner"), dict) and data["owner"].get("login"):
                     info.owner = data["owner"]["login"]
-                return info
+
+            ver, src = await _github_latest_version(hass, owner, repo)
+            info.latest_version = ver
+            info.latest_version_source = src
+            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()
+            data, status = await _safe_json(session, api, timeout=15)
+            if isinstance(data, dict):
                 info.description = data.get("description")
                 info.repo_name = data.get("name") or repo
                 info.default_branch = data.get("default_branch") or "main"
                 if isinstance(data.get("owner"), dict) and data["owner"].get("login"):
                     info.owner = data["owner"]["login"]
-                return info
+
+            ver, src = await _gitea_latest_version(hass, base, owner, repo)
+            info.latest_version = ver
+            info.latest_version_source = src
+            return info
 
         return info