custom_components/bahmcloud_store/providers.py aktualisiert
This commit is contained in:
@@ -51,12 +51,7 @@ def detect_provider(repo_url: str) -> str:
|
||||
return "github"
|
||||
if "gitlab" in host:
|
||||
return "gitlab"
|
||||
|
||||
owner, repo = _split_owner_repo(repo_url)
|
||||
if owner and repo:
|
||||
return "gitea"
|
||||
|
||||
return "generic"
|
||||
return "gitea"
|
||||
|
||||
|
||||
async def _safe_json(session, url: str, *, headers: dict | None = None, timeout: int = 20):
|
||||
@@ -81,179 +76,41 @@ async def _safe_text(session, url: str, *, headers: dict | None = None, timeout:
|
||||
return None, None
|
||||
|
||||
|
||||
def _extract_tag_from_github_url(url: str) -> str | None:
|
||||
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
|
||||
|
||||
|
||||
def _strip_html(s: str) -> str:
|
||||
# minimal HTML entity cleanup for meta descriptions
|
||||
out = (
|
||||
s.replace("&", "&")
|
||||
.replace(""", '"')
|
||||
.replace("'", "'")
|
||||
.replace("<", "<")
|
||||
.replace(">", ">")
|
||||
)
|
||||
return re.sub(r"\s+", " ", out).strip()
|
||||
|
||||
|
||||
def _extract_meta(html: str, *, prop: str | None = None, name: str | None = None) -> str | None:
|
||||
# Extract <meta property="og:description" content="...">
|
||||
# or <meta name="description" content="...">
|
||||
if prop:
|
||||
# property="..." content="..."
|
||||
m = re.search(
|
||||
r'<meta[^>]+property=["\']' + re.escape(prop) + r'["\'][^>]+content=["\']([^"\']+)["\']',
|
||||
html,
|
||||
flags=re.IGNORECASE,
|
||||
)
|
||||
if m:
|
||||
return _strip_html(m.group(1))
|
||||
m = re.search(
|
||||
r'<meta[^>]+content=["\']([^"\']+)["\'][^>]+property=["\']' + re.escape(prop) + r'["\']',
|
||||
html,
|
||||
flags=re.IGNORECASE,
|
||||
)
|
||||
if m:
|
||||
return _strip_html(m.group(1))
|
||||
|
||||
if name:
|
||||
m = re.search(
|
||||
r'<meta[^>]+name=["\']' + re.escape(name) + r'["\'][^>]+content=["\']([^"\']+)["\']',
|
||||
html,
|
||||
flags=re.IGNORECASE,
|
||||
)
|
||||
if m:
|
||||
return _strip_html(m.group(1))
|
||||
m = re.search(
|
||||
r'<meta[^>]+content=["\']([^"\']+)["\'][^>]+name=["\']' + re.escape(name) + r'["\']',
|
||||
html,
|
||||
flags=re.IGNORECASE,
|
||||
)
|
||||
if m:
|
||||
return _strip_html(m.group(1))
|
||||
|
||||
return None
|
||||
|
||||
|
||||
async def _github_description_html(hass: HomeAssistant, owner: str, repo: str) -> str | None:
|
||||
"""
|
||||
GitHub API may be rate-limited; fetch public HTML and read meta description.
|
||||
"""
|
||||
session = async_get_clientsession(hass)
|
||||
headers = {
|
||||
"User-Agent": UA,
|
||||
"Accept": "text/html,application/xhtml+xml",
|
||||
}
|
||||
|
||||
html, status = await _safe_text(session, f"https://github.com/{owner}/{repo}", headers=headers)
|
||||
if not html or status != 200:
|
||||
url = f"https://github.com/{owner}/{repo}"
|
||||
html, status = await _safe_text(session, url, headers={"User-Agent": UA})
|
||||
if status != 200 or not html:
|
||||
return None
|
||||
|
||||
desc = _extract_meta(html, prop="og:description")
|
||||
if desc:
|
||||
return desc
|
||||
# Try to locate the repository description meta tags
|
||||
m = re.search(r'<meta\s+property="og:description"\s+content="([^"]+)"', html)
|
||||
if m:
|
||||
return m.group(1).strip()
|
||||
|
||||
desc = _extract_meta(html, name="description")
|
||||
if desc:
|
||||
return desc
|
||||
m = re.search(r'<meta\s+name="description"\s+content="([^"]+)"', html)
|
||||
if m:
|
||||
return m.group(1).strip()
|
||||
|
||||
return None
|
||||
|
||||
|
||||
async def _github_latest_version_atom(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]:
|
||||
session = async_get_clientsession(hass)
|
||||
headers = {"User-Agent": UA, "Accept": "application/atom+xml,text/xml;q=0.9,*/*;q=0.8"}
|
||||
|
||||
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
|
||||
|
||||
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]:
|
||||
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]:
|
||||
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"
|
||||
|
||||
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")
|
||||
if isinstance(tag, str) and tag.strip():
|
||||
return tag.strip(), "tag"
|
||||
|
||||
return None, None
|
||||
|
||||
|
||||
async def _github_latest_version(hass: HomeAssistant, owner: str, repo: str) -> tuple[str | None, str | None]:
|
||||
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)
|
||||
headers = {"Accept": "application/vnd.github+json", "User-Agent": UA}
|
||||
|
||||
data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/releases?limit=1")
|
||||
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"
|
||||
# Prefer releases
|
||||
data, status = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/releases/latest", headers=headers)
|
||||
if isinstance(data, dict) and data.get("tag_name"):
|
||||
return str(data["tag_name"]), "release"
|
||||
|
||||
data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/tags?limit=1")
|
||||
if isinstance(data, list) and data:
|
||||
tag = data[0].get("name")
|
||||
if isinstance(tag, str) and tag.strip():
|
||||
return tag.strip(), "tag"
|
||||
if status == 404:
|
||||
# No releases -> try tags
|
||||
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:
|
||||
t = data[0]
|
||||
if isinstance(t, dict) and t.get("name"):
|
||||
return str(t["name"]), "tag"
|
||||
|
||||
return None, None
|
||||
|
||||
@@ -261,27 +118,36 @@ 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]:
|
||||
session = async_get_clientsession(hass)
|
||||
headers = {"User-Agent": UA}
|
||||
|
||||
project = quote_plus(f"{owner}/{repo}")
|
||||
|
||||
data, _ = await _safe_json(
|
||||
session,
|
||||
f"{base}/api/v4/projects/{project}/releases?per_page=1&order_by=released_at&sort=desc",
|
||||
headers=headers,
|
||||
)
|
||||
# Releases
|
||||
data, _ = await _safe_json(session, f"{base}/api/v4/projects/{project}/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"
|
||||
r = data[0]
|
||||
if isinstance(r, dict) and r.get("tag_name"):
|
||||
return str(r["tag_name"]), "release"
|
||||
|
||||
data, _ = await _safe_json(
|
||||
session,
|
||||
f"{base}/api/v4/projects/{project}/repository/tags?per_page=1&order_by=updated&sort=desc",
|
||||
headers=headers,
|
||||
)
|
||||
# Tags
|
||||
data, _ = await _safe_json(session, f"{base}/api/v4/projects/{project}/repository/tags?per_page=1", headers=headers)
|
||||
if isinstance(data, list) and data:
|
||||
tag = data[0].get("name")
|
||||
if isinstance(tag, str) and tag.strip():
|
||||
return tag.strip(), "tag"
|
||||
t = data[0]
|
||||
if isinstance(t, dict) and t.get("name"):
|
||||
return str(t["name"]), "tag"
|
||||
|
||||
# Atom feed fallback (public instances)
|
||||
atom, status = await _safe_text(session, f"{base}/{owner}/{repo}/-/tags?format=atom", headers=headers)
|
||||
if status == 200 and atom:
|
||||
try:
|
||||
root = ET.fromstring(atom)
|
||||
ns = {"a": "http://www.w3.org/2005/Atom"}
|
||||
entry = root.find("a:entry", ns)
|
||||
if entry is not None:
|
||||
title = entry.find("a:title", ns)
|
||||
if title is not None and title.text:
|
||||
return title.text.strip(), "atom"
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return None, None
|
||||
|
||||
@@ -363,16 +229,4 @@ async def fetch_repo_info(hass: HomeAssistant, repo_url: str) -> RepoInfo:
|
||||
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"]
|
||||
|
||||
ver, src = await _gitea_latest_version(hass, base, owner, repo)
|
||||
info.latest_version = ver
|
||||
info.latest_version_source = src
|
||||
return info
|
||||
|
||||
return info
|
||||
|
||||
except Exception as e:
|
||||
_LOGGER.debug("Provider fetch failed for %s: %s", repo_url, e)
|
||||
return info
|
||||
if isinstance(data.get("owner"), dict) and data["owner"].get("lo
|
||||
|
||||
Reference in New Issue
Block a user