bahmcloud/bahmcloud_store
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 8 Wiki Activity

Test2

Browse Source
This commit is contained in:
René Bachmann
2026-01-15 12:12:23 +00:00
parent 1c8a83effc
commit 1305656d10
1 changed files with 108 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
custom_components/bahmcloud_store/providers.py
View File

@@ -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 <host>/<owner>/<repo> 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 <entry> then a <link href="..."> 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")