165 lines
4.9 KiB
Python
165 lines
4.9 KiB
Python
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()
|