custom_components/bahmcloud_store/providers.py aktualisiert

This commit is contained in:
2026-01-15 10:12:52 +00:00
parent bf29faab04
commit acb01b9768

View File

@@ -9,6 +9,8 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
UA = "BahmcloudStore/0.4.0 (Home Assistant)"
@dataclass @dataclass
class RepoInfo: class RepoInfo:
@@ -18,17 +20,27 @@ class RepoInfo:
provider: str | None = None provider: str | None = None
default_branch: str | None = None default_branch: str | None = None
# Latest version best-effort
latest_version: 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" | None
def _normalize_repo_name(name: str | None) -> str | None:
if not name:
return None
n = name.strip()
if n.endswith(".git"):
n = n[:-4]
return n or None
def _split_owner_repo(repo_url: str) -> tuple[str | None, str | None]: def _split_owner_repo(repo_url: str) -> tuple[str | None, str | None]:
u = urlparse(repo_url.rstrip("/")) u = urlparse(repo_url.rstrip("/"))
parts = [p for p in u.path.strip("/").split("/") if p] parts = [p for p in u.path.strip("/").split("/") if p]
if len(parts) < 2: if len(parts) < 2:
return None, None return None, None
return parts[0], parts[1] owner = parts[0].strip() or None
repo = _normalize_repo_name(parts[1])
return owner, repo
def detect_provider(repo_url: str) -> str: def detect_provider(repo_url: str) -> str:
@@ -55,26 +67,18 @@ async def _safe_json(session, url: str, *, headers: dict | None = None, timeout:
async def _github_latest_version(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]: async def _github_latest_version(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]:
session = async_get_clientsession(hass) session = async_get_clientsession(hass)
headers = {
"Accept": "application/vnd.github+json",
"User-Agent": UA,
}
# 1) Latest release data, _ = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/releases/latest", headers=headers)
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): if isinstance(data, dict):
tag = data.get("tag_name") or data.get("name") tag = data.get("tag_name") or data.get("name")
if isinstance(tag, str) and tag.strip(): if isinstance(tag, str) and tag.strip():
return tag.strip(), "release" return tag.strip(), "release"
# 2) Latest tag (first item) data, _ = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/tags?per_page=1", headers=headers)
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: if isinstance(data, list) and data:
tag = data[0].get("name") tag = data[0].get("name")
if isinstance(tag, str) and tag.strip(): if isinstance(tag, str) and tag.strip():
@@ -86,23 +90,13 @@ async def _github_latest_version(hass: HomeAssistant, owner: str, repo: str) ->
async def _gitea_latest_version(hass: HomeAssistant, base: str, owner: str, repo: str) -> tuple[str | None, str | None]: async def _gitea_latest_version(hass: HomeAssistant, base: str, owner: str, repo: str) -> tuple[str | None, str | None]:
session = async_get_clientsession(hass) session = async_get_clientsession(hass)
# 1) Releases (first item) data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/releases?limit=1")
data, status = await _safe_json(
session,
f"{base}/api/v1/repos/{owner}/{repo}/releases?limit=1",
timeout=15,
)
if isinstance(data, list) and data: if isinstance(data, list) and data:
tag = data[0].get("tag_name") or data[0].get("name") tag = data[0].get("tag_name") or data[0].get("name")
if isinstance(tag, str) and tag.strip(): if isinstance(tag, str) and tag.strip():
return tag.strip(), "release" return tag.strip(), "release"
# 2) Tags (first item) data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/tags?limit=1")
data, status = await _safe_json(
session,
f"{base}/api/v1/repos/{owner}/{repo}/tags?limit=1",
timeout=15,
)
if isinstance(data, list) and data: if isinstance(data, list) and data:
tag = data[0].get("name") tag = data[0].get("name")
if isinstance(tag, str) and tag.strip(): if isinstance(tag, str) and tag.strip():
@@ -132,16 +126,14 @@ async def fetch_repo_info(hass: HomeAssistant, repo_url: str) -> RepoInfo:
try: try:
if provider == "github": if provider == "github":
api = f"https://api.github.com/repos/{owner}/{repo}" headers = {
data, status = await _safe_json( "Accept": "application/vnd.github+json",
session, "User-Agent": UA,
api, }
headers={"Accept": "application/vnd.github+json"}, data, _ = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}", headers=headers)
timeout=15,
)
if isinstance(data, dict): if isinstance(data, dict):
info.description = data.get("description") info.description = data.get("description")
info.repo_name = data.get("name") or repo info.repo_name = _normalize_repo_name(data.get("name")) or repo
info.default_branch = data.get("default_branch") or "main" info.default_branch = data.get("default_branch") or "main"
if isinstance(data.get("owner"), dict) and data["owner"].get("login"): if isinstance(data.get("owner"), dict) and data["owner"].get("login"):
info.owner = data["owner"]["login"] info.owner = data["owner"]["login"]
@@ -155,11 +147,10 @@ async def fetch_repo_info(hass: HomeAssistant, repo_url: str) -> RepoInfo:
u = urlparse(repo_url.rstrip("/")) u = urlparse(repo_url.rstrip("/"))
base = f"{u.scheme}://{u.netloc}" base = f"{u.scheme}://{u.netloc}"
api = f"{base}/api/v1/repos/{owner}/{repo}" data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}")
data, status = await _safe_json(session, api, timeout=15)
if isinstance(data, dict): if isinstance(data, dict):
info.description = data.get("description") info.description = data.get("description")
info.repo_name = data.get("name") or repo info.repo_name = _normalize_repo_name(data.get("name")) or repo
info.default_branch = data.get("default_branch") or "main" info.default_branch = data.get("default_branch") or "main"
if isinstance(data.get("owner"), dict) and data["owner"].get("login"): if isinstance(data.get("owner"), dict) and data["owner"].get("login"):
info.owner = data["owner"]["login"] info.owner = data["owner"]["login"]