diff --git a/custom_components/bahmcloud_store/providers.py b/custom_components/bahmcloud_store/providers.py index dad6f1a..a491ab8 100644 --- a/custom_components/bahmcloud_store/providers.py +++ b/custom_components/bahmcloud_store/providers.py @@ -1,6 +1,8 @@ from __future__ import annotations import logging +import re +import xml.etree.ElementTree as ET from dataclasses import dataclass from urllib.parse import quote_plus, urlparse @@ -21,7 +23,7 @@ class RepoInfo: default_branch: str | None = None latest_version: str | None = None - latest_version_source: str | None = None # "release" | "tag" | None + latest_version_source: str | None = None # "release" | "tag" | "atom" | None def _normalize_repo_name(name: str | None) -> str | None: @@ -45,13 +47,11 @@ def _split_owner_repo(repo_url: str) -> tuple[str | None, str | None]: def detect_provider(repo_url: str) -> str: host = urlparse(repo_url).netloc.lower() - # Keep explicit providers stable if "github.com" in host: return "github" - if "gitlab" in host: # covers gitlab.com and many self-hosted domains containing "gitlab" + if "gitlab" in host: return "gitlab" - # Heuristic: if it looks like a standard // URL, treat as gitea-like owner, repo = _split_owner_repo(repo_url) if owner and repo: return "gitea" @@ -70,38 +70,94 @@ async def _safe_json(session, url: str, *, headers: dict | None = None, timeout: return None, None -async def _github_latest_version(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]: +async def _safe_text(session, url: str, *, headers: dict | None = None, timeout: int = 20): + try: + async with session.get(url, timeout=timeout, headers=headers) as resp: + status = resp.status + if status != 200: + return None, status + return await resp.text(), status + except Exception: + return None, None + + +def _extract_tag_from_github_url(url: str) -> str | None: + # Examples: + # https://github.com/owner/repo/releases/tag/v1.2.3 + # https://github.com/owner/repo/tag/v1.2.3 (rare) + m = re.search(r"/releases/tag/([^/?#]+)", url) + if m: + return m.group(1) + m = re.search(r"/tag/([^/?#]+)", url) + if m: + return m.group(1) + return None + + +async def _github_latest_version_atom(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]: """ - Robust strategy: - 1) /releases/latest (can be 404 if no releases) - 2) /releases?per_page=1 (some repos have releases but latest endpoint can still behave differently) - 3) /tags?per_page=1 + Uses GitHub public Atom feed (no api.github.com). + This avoids API rate limits and works for most public repos. """ session = async_get_clientsession(hass) - headers = { - "Accept": "application/vnd.github+json", - "User-Agent": UA, - } + headers = {"User-Agent": UA, "Accept": "application/atom+xml,text/xml;q=0.9,*/*;q=0.8"} - # 1) Releases latest - data, status = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/releases/latest", headers=headers) + xml_text, _ = await _safe_text(session, f"https://github.com/{owner}/{repo}/releases.atom", headers=headers) + if not xml_text: + return None, None + + try: + root = ET.fromstring(xml_text) + except Exception: + return None, None + + # Atom namespace can vary; search entries robustly + # Find first then a that points to a release tag. + for entry in root.findall(".//{*}entry"): + for link in entry.findall(".//{*}link"): + href = link.attrib.get("href") + if not href: + continue + tag = _extract_tag_from_github_url(href) + if tag: + return tag, "atom" + + return None, None + + +async def _github_latest_version_redirect(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]: + """ + Fallback: HEAD /releases/latest and parse Location header. + """ + session = async_get_clientsession(hass) + headers = {"User-Agent": UA} + url = f"https://github.com/{owner}/{repo}/releases/latest" + try: + async with session.head(url, allow_redirects=False, timeout=15, headers=headers) as resp: + if resp.status in (301, 302, 303, 307, 308): + loc = resp.headers.get("Location") + if loc: + tag = _extract_tag_from_github_url(loc) + if tag: + return tag, "release" + except Exception: + pass + return None, None + + +async def _github_latest_version_api(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]: + """ + Optional API path (may be rate-limited). Keep as last resort. + """ + session = async_get_clientsession(hass) + headers = {"Accept": "application/vnd.github+json", "User-Agent": UA} + + data, _ = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/releases/latest", headers=headers) if isinstance(data, dict): tag = data.get("tag_name") or data.get("name") if isinstance(tag, str) and tag.strip(): return tag.strip(), "release" - # If rate-limited, log and continue with tags (tags may also be blocked but try) - if status == 403: - _LOGGER.debug("GitHub rate limit or forbidden on releases/latest for %s/%s", owner, repo) - - # 2) Releases list fallback - data, _ = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/releases?per_page=1", headers=headers) - 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" - - # 3) Tag fallback data, _ = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/tags?per_page=1", headers=headers) if isinstance(data, list) and data: tag = data[0].get("name") @@ -111,6 +167,24 @@ async def _github_latest_version(hass: HomeAssistant, owner: str, repo: str) -> return None, None +async def _github_latest_version(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]: + """ + Durable strategy: + 1) Atom feed (no API) + 2) Redirect parse (no API) + 3) API fallback + """ + tag, src = await _github_latest_version_atom(hass, owner, repo) + if tag: + return tag, src + + tag, src = await _github_latest_version_redirect(hass, owner, repo) + if tag: + return tag, src + + return await _github_latest_version_api(hass, owner, repo) + + async def _gitea_latest_version(hass: HomeAssistant, base: str, owner: str, repo: str) -> tuple[str | None, str | None]: session = async_get_clientsession(hass) @@ -130,17 +204,10 @@ async def _gitea_latest_version(hass: HomeAssistant, base: str, owner: str, repo async def _gitlab_latest_version(hass: HomeAssistant, base: str, owner: str, repo: str) -> tuple[str | None, str | None]: - """ - GitLab API: - - releases: /api/v4/projects/:id/releases - - tags: /api/v4/projects/:id/repository/tags - Project id can be URL-encoded full path (owner/repo). - """ session = async_get_clientsession(hass) headers = {"User-Agent": UA} project = quote_plus(f"{owner}/{repo}") - # 1) Releases (newest first) data, _ = await _safe_json( session, f"{base}/api/v4/projects/{project}/releases?per_page=1&order_by=released_at&sort=desc", @@ -151,7 +218,6 @@ async def _gitlab_latest_version(hass: HomeAssistant, base: str, owner: str, rep if isinstance(tag, str) and tag.strip(): return tag.strip(), "release" - # 2) Tags (newest first) data, _ = await _safe_json( session, f"{base}/api/v4/projects/{project}/repository/tags?per_page=1&order_by=updated&sort=desc", @@ -186,18 +252,19 @@ async def fetch_repo_info(hass: HomeAssistant, repo_url: str) -> RepoInfo: try: if provider == "github": - headers = { - "Accept": "application/vnd.github+json", - "User-Agent": UA, - } - - data, _ = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}", headers=headers) + # Repo details: try API first, but don't fail if blocked + headers = {"Accept": "application/vnd.github+json", "User-Agent": UA} + data, status = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}", headers=headers) if isinstance(data, dict): info.description = data.get("description") info.repo_name = _normalize_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"] + else: + # If API blocked, at least keep defaults, provider remains github + if status == 403: + _LOGGER.debug("GitHub API blocked/rate-limited for repo info %s/%s", owner, repo) ver, src = await _github_latest_version(hass, owner, repo) info.latest_version = ver @@ -210,7 +277,6 @@ async def fetch_repo_info(hass: HomeAssistant, repo_url: str) -> RepoInfo: headers = {"User-Agent": UA} project = quote_plus(f"{owner}/{repo}") - # Project info data, _ = await _safe_json(session, f"{base}/api/v4/projects/{project}", headers=headers) if isinstance(data, dict): info.description = data.get("description")