From 72ce95525cd1d7a189be31e0338151e37887398e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ren=C3=A9=20Bachmann?= Date: Thu, 15 Jan 2026 15:28:35 +0000 Subject: [PATCH] custom_components/bahmcloud_store/providers.py aktualisiert --- .../bahmcloud_store/providers.py | 244 ++++-------------- 1 file changed, 49 insertions(+), 195 deletions(-) diff --git a/custom_components/bahmcloud_store/providers.py b/custom_components/bahmcloud_store/providers.py index 494f61d..0898d2f 100644 --- a/custom_components/bahmcloud_store/providers.py +++ b/custom_components/bahmcloud_store/providers.py @@ -51,12 +51,7 @@ def detect_provider(repo_url: str) -> str: return "github" if "gitlab" in host: return "gitlab" - - owner, repo = _split_owner_repo(repo_url) - if owner and repo: - return "gitea" - - return "generic" + return "gitea" async def _safe_json(session, url: str, *, headers: dict | None = None, timeout: int = 20): @@ -81,179 +76,41 @@ async def _safe_text(session, url: str, *, headers: dict | None = None, timeout: return None, None -def _extract_tag_from_github_url(url: str) -> str | None: - 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 - - -def _strip_html(s: str) -> str: - # minimal HTML entity cleanup for meta descriptions - out = ( - s.replace("&", "&") - .replace(""", '"') - .replace("'", "'") - .replace("<", "<") - .replace(">", ">") - ) - return re.sub(r"\s+", " ", out).strip() - - -def _extract_meta(html: str, *, prop: str | None = None, name: str | None = None) -> str | None: - # Extract - # or - if prop: - # property="..." content="..." - m = re.search( - r']+property=["\']' + re.escape(prop) + r'["\'][^>]+content=["\']([^"\']+)["\']', - html, - flags=re.IGNORECASE, - ) - if m: - return _strip_html(m.group(1)) - m = re.search( - r']+content=["\']([^"\']+)["\'][^>]+property=["\']' + re.escape(prop) + r'["\']', - html, - flags=re.IGNORECASE, - ) - if m: - return _strip_html(m.group(1)) - - if name: - m = re.search( - r']+name=["\']' + re.escape(name) + r'["\'][^>]+content=["\']([^"\']+)["\']', - html, - flags=re.IGNORECASE, - ) - if m: - return _strip_html(m.group(1)) - m = re.search( - r']+content=["\']([^"\']+)["\'][^>]+name=["\']' + re.escape(name) + r'["\']', - html, - flags=re.IGNORECASE, - ) - if m: - return _strip_html(m.group(1)) - - return None - - async def _github_description_html(hass: HomeAssistant, owner: str, repo: str) -> str | None: - """ - GitHub API may be rate-limited; fetch public HTML and read meta description. - """ session = async_get_clientsession(hass) - headers = { - "User-Agent": UA, - "Accept": "text/html,application/xhtml+xml", - } - - html, status = await _safe_text(session, f"https://github.com/{owner}/{repo}", headers=headers) - if not html or status != 200: + url = f"https://github.com/{owner}/{repo}" + html, status = await _safe_text(session, url, headers={"User-Agent": UA}) + if status != 200 or not html: return None - desc = _extract_meta(html, prop="og:description") - if desc: - return desc + # Try to locate the repository description meta tags + m = re.search(r' tuple[str | None, str | None]: - session = async_get_clientsession(hass) - headers = {"User-Agent": UA, "Accept": "application/atom+xml,text/xml;q=0.9,*/*;q=0.8"} - - 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 - - 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]: - 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]: - 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" - - 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") - if isinstance(tag, str) and tag.strip(): - return tag.strip(), "tag" - - return None, None - - async def _github_latest_version(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]: - 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) + headers = {"Accept": "application/vnd.github+json", "User-Agent": UA} - data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/releases?limit=1") - 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" + # Prefer releases + data, status = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/releases/latest", headers=headers) + if isinstance(data, dict) and data.get("tag_name"): + return str(data["tag_name"]), "release" - data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/tags?limit=1") - if isinstance(data, list) and data: - tag = data[0].get("name") - if isinstance(tag, str) and tag.strip(): - return tag.strip(), "tag" + if status == 404: + # No releases -> try tags + 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: + t = data[0] + if isinstance(t, dict) and t.get("name"): + return str(t["name"]), "tag" return None, None @@ -261,27 +118,36 @@ 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]: session = async_get_clientsession(hass) headers = {"User-Agent": UA} + project = quote_plus(f"{owner}/{repo}") - data, _ = await _safe_json( - session, - f"{base}/api/v4/projects/{project}/releases?per_page=1&order_by=released_at&sort=desc", - headers=headers, - ) + # Releases + data, _ = await _safe_json(session, f"{base}/api/v4/projects/{project}/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" + r = data[0] + if isinstance(r, dict) and r.get("tag_name"): + return str(r["tag_name"]), "release" - data, _ = await _safe_json( - session, - f"{base}/api/v4/projects/{project}/repository/tags?per_page=1&order_by=updated&sort=desc", - headers=headers, - ) + # Tags + data, _ = await _safe_json(session, f"{base}/api/v4/projects/{project}/repository/tags?per_page=1", headers=headers) if isinstance(data, list) and data: - tag = data[0].get("name") - if isinstance(tag, str) and tag.strip(): - return tag.strip(), "tag" + t = data[0] + if isinstance(t, dict) and t.get("name"): + return str(t["name"]), "tag" + + # Atom feed fallback (public instances) + atom, status = await _safe_text(session, f"{base}/{owner}/{repo}/-/tags?format=atom", headers=headers) + if status == 200 and atom: + try: + root = ET.fromstring(atom) + ns = {"a": "http://www.w3.org/2005/Atom"} + entry = root.find("a:entry", ns) + if entry is not None: + title = entry.find("a:title", ns) + if title is not None and title.text: + return title.text.strip(), "atom" + except Exception: + pass return None, None @@ -363,16 +229,4 @@ async def fetch_repo_info(hass: HomeAssistant, repo_url: str) -> RepoInfo: 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"] - - ver, src = await _gitea_latest_version(hass, base, owner, repo) - info.latest_version = ver - info.latest_version_source = src - return info - - return info - - except Exception as e: - _LOGGER.debug("Provider fetch failed for %s: %s", repo_url, e) - return info \ No newline at end of file + if isinstance(data.get("owner"), dict) and data["owner"].get("lo