diff --git a/custom_components/bahmcloud_store/providers.py b/custom_components/bahmcloud_store/providers.py index be6a1b5..4580e11 100644 --- a/custom_components/bahmcloud_store/providers.py +++ b/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