.
This commit is contained in:
@@ -9,6 +9,9 @@ from urllib.parse import quote_plus, urlparse
|
|||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||||
|
|
||||||
|
# Home Assistant includes "packaging"
|
||||||
|
from packaging.version import InvalidVersion, Version
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
UA = "BahmcloudStore (Home Assistant)"
|
UA = "BahmcloudStore (Home Assistant)"
|
||||||
@@ -97,6 +100,41 @@ def _extract_meta(html: str, *, prop: str | None = None, name: str | None = None
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _semver_key(tag: str) -> Version | None:
|
||||||
|
t = (tag or "").strip()
|
||||||
|
if not t:
|
||||||
|
return None
|
||||||
|
if t.startswith(("v", "V")):
|
||||||
|
t = t[1:]
|
||||||
|
try:
|
||||||
|
return Version(t)
|
||||||
|
except InvalidVersion:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _pick_highest_semver(candidates: list[str]) -> str | None:
|
||||||
|
"""Pick highest semantic version from a list of tag strings.
|
||||||
|
|
||||||
|
If no semver parseable tags exist, return None.
|
||||||
|
"""
|
||||||
|
parsed: list[tuple[Version, str]] = []
|
||||||
|
for t in candidates:
|
||||||
|
if not isinstance(t, str):
|
||||||
|
continue
|
||||||
|
ts = t.strip()
|
||||||
|
if not ts:
|
||||||
|
continue
|
||||||
|
v = _semver_key(ts)
|
||||||
|
if v is not None:
|
||||||
|
parsed.append((v, ts))
|
||||||
|
|
||||||
|
if not parsed:
|
||||||
|
return None
|
||||||
|
|
||||||
|
parsed.sort(key=lambda x: x[0], reverse=True)
|
||||||
|
return parsed[0][1]
|
||||||
|
|
||||||
|
|
||||||
async def _github_description_html(hass: HomeAssistant, owner: str, repo: str) -> str | None:
|
async def _github_description_html(hass: HomeAssistant, owner: str, repo: str) -> str | None:
|
||||||
session = async_get_clientsession(hass)
|
session = async_get_clientsession(hass)
|
||||||
url = f"https://github.com/{owner}/{repo}"
|
url = f"https://github.com/{owner}/{repo}"
|
||||||
@@ -161,21 +199,33 @@ async def _github_latest_version_api(hass: HomeAssistant, owner: str, repo: str)
|
|||||||
session = async_get_clientsession(hass)
|
session = async_get_clientsession(hass)
|
||||||
headers = {"Accept": "application/vnd.github+json", "User-Agent": UA}
|
headers = {"Accept": "application/vnd.github+json", "User-Agent": UA}
|
||||||
|
|
||||||
|
# 1) Preferred: GitHub latest release endpoint (already good)
|
||||||
data, status = 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=headers)
|
||||||
if isinstance(data, dict) and data.get("tag_name"):
|
if isinstance(data, dict) and data.get("tag_name"):
|
||||||
return str(data["tag_name"]), "release"
|
return str(data["tag_name"]), "release"
|
||||||
|
|
||||||
|
# 2) If no releases: fetch multiple tags and select highest semver
|
||||||
if status == 404:
|
if status == 404:
|
||||||
data, _ = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/tags?per_page=1", headers=headers)
|
tags_data, _ = await _safe_json(session, f"https://api.github.com/repos/{owner}/{repo}/tags?per_page=100", headers=headers)
|
||||||
if isinstance(data, list) and data:
|
tags: list[str] = []
|
||||||
t = data[0]
|
if isinstance(tags_data, list):
|
||||||
|
for t in tags_data:
|
||||||
if isinstance(t, dict) and t.get("name"):
|
if isinstance(t, dict) and t.get("name"):
|
||||||
return str(t["name"]), "tag"
|
tags.append(str(t["name"]))
|
||||||
|
|
||||||
|
best = _pick_highest_semver(tags)
|
||||||
|
if best:
|
||||||
|
return best, "tag"
|
||||||
|
|
||||||
|
# fallback: keep old behavior (first tag, if any)
|
||||||
|
if tags:
|
||||||
|
return tags[0], "tag"
|
||||||
|
|
||||||
return None, None
|
return None, None
|
||||||
|
|
||||||
|
|
||||||
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]:
|
||||||
|
# Redirect is very robust if releases exist
|
||||||
tag, src = await _github_latest_version_redirect(hass, owner, repo)
|
tag, src = await _github_latest_version_redirect(hass, owner, repo)
|
||||||
if tag:
|
if tag:
|
||||||
return tag, src
|
return tag, src
|
||||||
@@ -190,17 +240,34 @@ 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)
|
||||||
|
|
||||||
data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/releases?limit=1")
|
# Fetch multiple releases, pick highest semver by tag_name
|
||||||
if isinstance(data, list) and data:
|
rels, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/releases?limit=50")
|
||||||
r = data[0]
|
release_tags: list[str] = []
|
||||||
|
if isinstance(rels, list):
|
||||||
|
for r in rels:
|
||||||
if isinstance(r, dict) and r.get("tag_name"):
|
if isinstance(r, dict) and r.get("tag_name"):
|
||||||
return str(r["tag_name"]), "release"
|
release_tags.append(str(r["tag_name"]))
|
||||||
|
|
||||||
data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/tags?limit=1")
|
best_rel = _pick_highest_semver(release_tags)
|
||||||
if isinstance(data, list) and data:
|
if best_rel:
|
||||||
t = data[0]
|
return best_rel, "release"
|
||||||
|
if release_tags:
|
||||||
|
# fallback: first release tag
|
||||||
|
return release_tags[0], "release"
|
||||||
|
|
||||||
|
# No releases: fetch multiple tags, pick highest semver
|
||||||
|
tags_data, _ = await _safe_json(session, f"{base}/api/v1/repos/{owner}/{repo}/tags?limit=50")
|
||||||
|
tags: list[str] = []
|
||||||
|
if isinstance(tags_data, list):
|
||||||
|
for t in tags_data:
|
||||||
if isinstance(t, dict) and t.get("name"):
|
if isinstance(t, dict) and t.get("name"):
|
||||||
return str(t["name"]), "tag"
|
tags.append(str(t["name"]))
|
||||||
|
|
||||||
|
best = _pick_highest_semver(tags)
|
||||||
|
if best:
|
||||||
|
return best, "tag"
|
||||||
|
if tags:
|
||||||
|
return tags[0], "tag"
|
||||||
|
|
||||||
return None, None
|
return None, None
|
||||||
|
|
||||||
@@ -213,18 +280,35 @@ async def _gitlab_latest_version(
|
|||||||
|
|
||||||
project = quote_plus(f"{owner}/{repo}")
|
project = quote_plus(f"{owner}/{repo}")
|
||||||
|
|
||||||
data, _ = await _safe_json(session, f"{base}/api/v4/projects/{project}/releases?per_page=1", headers=headers)
|
# Fetch multiple releases, pick highest semver by tag_name
|
||||||
if isinstance(data, list) and data:
|
rels, _ = await _safe_json(session, f"{base}/api/v4/projects/{project}/releases?per_page=50", headers=headers)
|
||||||
r = data[0]
|
release_tags: list[str] = []
|
||||||
|
if isinstance(rels, list):
|
||||||
|
for r in rels:
|
||||||
if isinstance(r, dict) and r.get("tag_name"):
|
if isinstance(r, dict) and r.get("tag_name"):
|
||||||
return str(r["tag_name"]), "release"
|
release_tags.append(str(r["tag_name"]))
|
||||||
|
|
||||||
data, _ = await _safe_json(session, f"{base}/api/v4/projects/{project}/repository/tags?per_page=1", headers=headers)
|
best_rel = _pick_highest_semver(release_tags)
|
||||||
if isinstance(data, list) and data:
|
if best_rel:
|
||||||
t = data[0]
|
return best_rel, "release"
|
||||||
|
if release_tags:
|
||||||
|
return release_tags[0], "release"
|
||||||
|
|
||||||
|
# No releases: fetch multiple tags, pick highest semver
|
||||||
|
tags_data, _ = await _safe_json(session, f"{base}/api/v4/projects/{project}/repository/tags?per_page=50", headers=headers)
|
||||||
|
tags: list[str] = []
|
||||||
|
if isinstance(tags_data, list):
|
||||||
|
for t in tags_data:
|
||||||
if isinstance(t, dict) and t.get("name"):
|
if isinstance(t, dict) and t.get("name"):
|
||||||
return str(t["name"]), "tag"
|
tags.append(str(t["name"]))
|
||||||
|
|
||||||
|
best = _pick_highest_semver(tags)
|
||||||
|
if best:
|
||||||
|
return best, "tag"
|
||||||
|
if tags:
|
||||||
|
return tags[0], "tag"
|
||||||
|
|
||||||
|
# last fallback: atom (often newest by date, not semver)
|
||||||
atom, status = await _safe_text(session, f"{base}/{owner}/{repo}/-/tags?format=atom", headers=headers)
|
atom, status = await _safe_text(session, f"{base}/{owner}/{repo}/-/tags?format=atom", headers=headers)
|
||||||
if status == 200 and atom:
|
if status == 200 and atom:
|
||||||
try:
|
try:
|
||||||
|
|||||||
Reference in New Issue
Block a user