custom_components/bahmcloud_store/metadata.py hinzugefügt

custom_components/bahmcloud_store/metadata.py

@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from typing import Any
+from urllib.parse import urlparse
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.util import yaml as ha_yaml
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class RepoMetadata:
+    source: str | None = None  # "bcs.yaml" | "hacs.yaml" | "hacs.json"
+    name: str | None = None
+    description: str | None = None
+    category: str | None = None
+    author: str | None = None
+    maintainer: str | None = None
+
+
+def _clean_str(v: Any) -> str | None:
+    if v is None:
+        return None
+    if isinstance(v, (int, float)):
+        v = str(v)
+    if not isinstance(v, str):
+        return None
+    s = v.strip()
+    return s or None
+
+
+def _extract_common_fields(data: dict[str, Any]) -> RepoMetadata:
+    """
+    Best-effort extraction across possible schemas.
+    We keep this forgiving because third-party repos vary widely.
+    """
+    md = RepoMetadata()
+
+    # Common / preferred keys for BCS
+    md.name = _clean_str(data.get("name"))
+    md.description = _clean_str(data.get("description"))
+    md.category = _clean_str(data.get("category"))
+    md.author = _clean_str(data.get("author"))
+    md.maintainer = _clean_str(data.get("maintainer"))
+
+    # HACS compatibility fields
+    # Some repos use 'render_readme', 'content_in_root', etc. – ignored for now.
+    # Some use "authors" list or "maintainers" list:
+    if not md.author:
+        a = data.get("authors") or data.get("author")
+        if isinstance(a, list) and a:
+            md.author = _clean_str(a[0])
+        elif isinstance(a, str):
+            md.author = _clean_str(a)
+
+    if not md.maintainer:
+        m = data.get("maintainers") or data.get("maintainer")
+        if isinstance(m, list) and m:
+            md.maintainer = _clean_str(m[0])
+        elif isinstance(m, str):
+            md.maintainer = _clean_str(m)
+
+    # Some HACS style manifests use "documentation" or "info" as description-like
+    if not md.description:
+        md.description = _clean_str(data.get("info")) or _clean_str(data.get("documentation"))
+
+    return md
+
+
+def _is_github(repo_url: str) -> bool:
+    return "github.com" in urlparse(repo_url).netloc.lower()
+
+
+def _is_gitea(repo_url: str) -> bool:
+    # We treat self-hosted owner/repo as gitea in this project.
+    host = urlparse(repo_url).netloc.lower()
+    return host and "github.com" not in host and "gitlab.com" not in host
+
+
+def _split_owner_repo(repo_url: str) -> tuple[str | None, str | None]:
+    u = urlparse(repo_url.rstrip("/"))
+    parts = [p for p in u.path.strip("/").split("/") if p]
+    if len(parts) < 2:
+        return None, None
+    return parts[0], parts[1]
+
+
+async def _fetch_text(hass: HomeAssistant, url: str) -> str | None:
+    session = async_get_clientsession(hass)
+    try:
+        async with session.get(url, timeout=15) as resp:
+            if resp.status != 200:
+                return None
+            return await resp.text()
+    except Exception:
+        return None
+
+
+async def fetch_repo_metadata(hass: HomeAssistant, repo_url: str, default_branch: str | None) -> RepoMetadata:
+    """
+    Best-effort metadata resolution from repo root:
+    1) bcs.yaml
+    2) hacs.yaml
+    3) hacs.json
+
+    Works for:
+    - GitHub: raw.githubusercontent.com
+    - Gitea: /raw/branch/<branch>/
+    """
+    owner, repo = _split_owner_repo(repo_url)
+    if not owner or not repo:
+        return RepoMetadata()
+
+    branch = default_branch or "main"
+
+    candidates: list[tuple[str, str]] = []
+
+    if _is_github(repo_url):
+        base = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}"
+        candidates = [
+            ("bcs.yaml", f"{base}/bcs.yaml"),
+            ("hacs.yaml", f"{base}/hacs.yaml"),
+            ("hacs.json", f"{base}/hacs.json"),
+        ]
+    elif _is_gitea(repo_url):
+        u = urlparse(repo_url.rstrip("/"))
+        base = f"{u.scheme}://{u.netloc}/{owner}/{repo}/raw/branch/{branch}"
+        candidates = [
+            ("bcs.yaml", f"{base}/bcs.yaml"),
+            ("hacs.yaml", f"{base}/hacs.yaml"),
+            ("hacs.json", f"{base}/hacs.json"),
+        ]
+    else:
+        # Unsupported provider for metadata raw fetch in 0.3.2
+        return RepoMetadata()
+
+    for source, url in candidates:
+        text = await _fetch_text(hass, url)
+        if not text:
+            continue
+
+        try:
+            if source.endswith(".json"):
+                data = json.loads(text)
+                if not isinstance(data, dict):
+                    continue
+            else:
+                data = ha_yaml.parse_yaml(text)
+                if not isinstance(data, dict):
+                    continue
+
+            md = _extract_common_fields(data)
+            md.source = source
+            return md
+        except Exception as e:
+            _LOGGER.debug("Failed parsing %s for %s: %s", source, repo_url, e)
+            continue
+
+    return RepoMetadata()