168 lines
4.8 KiB
Python
168 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
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" | None
|
|
name: str | None = None
|
|
description: str | None = None
|
|
category: str | None = None
|
|
author: str | None = None
|
|
maintainer: str | None = 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]:
|
|
u = urlparse(repo_url.rstrip("/"))
|
|
parts = [p for p in u.path.strip("/").split("/") if p]
|
|
if len(parts) < 2:
|
|
return None, None
|
|
owner = parts[0].strip() or None
|
|
repo = _normalize_repo_name(parts[1])
|
|
return owner, repo
|
|
|
|
|
|
def _is_github(repo_url: str) -> bool:
|
|
return "github.com" in urlparse(repo_url).netloc.lower()
|
|
|
|
|
|
def _is_gitlab(repo_url: str) -> bool:
|
|
return "gitlab" in urlparse(repo_url).netloc.lower()
|
|
|
|
|
|
def _is_gitea(repo_url: str) -> bool:
|
|
host = urlparse(repo_url).netloc.lower()
|
|
return host and ("github.com" not in host) and ("gitlab" not in host)
|
|
|
|
|
|
async def _fetch_text(hass: HomeAssistant, url: str) -> str | None:
|
|
session = async_get_clientsession(hass)
|
|
try:
|
|
async with session.get(url, timeout=20) as resp:
|
|
if resp.status != 200:
|
|
return None
|
|
return await resp.text()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _parse_meta_yaml(raw: str, source: str) -> RepoMetadata:
|
|
try:
|
|
data = ha_yaml.parse_yaml(raw)
|
|
if not isinstance(data, dict):
|
|
return RepoMetadata(source=source)
|
|
|
|
return RepoMetadata(
|
|
source=source,
|
|
name=data.get("name"),
|
|
description=data.get("description"),
|
|
category=data.get("category"),
|
|
author=data.get("author"),
|
|
maintainer=data.get("maintainer"),
|
|
)
|
|
except Exception:
|
|
return RepoMetadata(source=source)
|
|
|
|
|
|
def _parse_meta_hacs_json(raw: str) -> RepoMetadata:
|
|
try:
|
|
data = json.loads(raw)
|
|
if not isinstance(data, dict):
|
|
return RepoMetadata(source="hacs.json")
|
|
|
|
name = data.get("name")
|
|
description = data.get("description")
|
|
author = data.get("author")
|
|
maintainer = data.get("maintainer")
|
|
category = data.get("category") or data.get("type")
|
|
|
|
return RepoMetadata(
|
|
source="hacs.json",
|
|
name=name if isinstance(name, str) else None,
|
|
description=description if isinstance(description, str) else None,
|
|
category=category if isinstance(category, str) else None,
|
|
author=author if isinstance(author, str) else None,
|
|
maintainer=maintainer if isinstance(maintainer, str) else None,
|
|
)
|
|
except Exception:
|
|
return RepoMetadata(source="hacs.json")
|
|
|
|
|
|
async def fetch_repo_metadata(hass: HomeAssistant, repo_url: str, default_branch: str | None) -> RepoMetadata:
|
|
owner, repo = _split_owner_repo(repo_url)
|
|
if not owner or not repo:
|
|
return RepoMetadata()
|
|
|
|
branch = default_branch or "main"
|
|
|
|
# Priority:
|
|
# 1) bcs.yaml
|
|
# 2) hacs.yaml
|
|
# 3) hacs.json
|
|
filenames = ["bcs.yaml", "hacs.yaml", "hacs.json"]
|
|
|
|
candidates: list[tuple[str, str]] = []
|
|
|
|
if _is_github(repo_url):
|
|
base = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}"
|
|
for fn in filenames:
|
|
candidates.append((fn, f"{base}/{fn}"))
|
|
|
|
elif _is_gitlab(repo_url):
|
|
u = urlparse(repo_url.rstrip("/"))
|
|
root = f"{u.scheme}://{u.netloc}/{owner}/{repo}"
|
|
# GitLab raw format
|
|
# https://gitlab.com/<owner>/<repo>/-/raw/<branch>/<file>
|
|
for fn in filenames:
|
|
candidates.append((fn, f"{root}/-/raw/{branch}/{fn}"))
|
|
|
|
elif _is_gitea(repo_url):
|
|
u = urlparse(repo_url.rstrip("/"))
|
|
root = f"{u.scheme}://{u.netloc}/{owner}/{repo}"
|
|
|
|
bases = [
|
|
f"{root}/raw/branch/{branch}",
|
|
f"{root}/raw/{branch}",
|
|
]
|
|
for fn in filenames:
|
|
for b in bases:
|
|
candidates.append((fn, f"{b}/{fn}"))
|
|
|
|
else:
|
|
return RepoMetadata()
|
|
|
|
for fn, url in candidates:
|
|
raw = await _fetch_text(hass, url)
|
|
if not raw:
|
|
continue
|
|
|
|
if fn.endswith(".json"):
|
|
meta = _parse_meta_hacs_json(raw)
|
|
if meta.source:
|
|
return meta
|
|
continue
|
|
|
|
meta = _parse_meta_yaml(raw, fn)
|
|
if meta.source:
|
|
return meta
|
|
|
|
return RepoMetadata() |