Panneau MAJ simplifié : dépôt public, plus de configuration Gitea — patchnote du dernier tag
This commit is contained in:
@@ -114,23 +114,3 @@ async def set_sonarr_config(url: str, apikey: str) -> None:
|
||||
await set_setting(SONARR_APIKEY_KEY, apikey.strip())
|
||||
logger.info("Configuration Sonarr enregistrée (%s)", url)
|
||||
|
||||
# ---------------------------------------------------------------- mise à jour logicielle
|
||||
|
||||
UPDATE_GITEA_URL_KEY = "update:gitea_url"
|
||||
UPDATE_REPO_KEY = "update:repo"
|
||||
UPDATE_TOKEN_KEY = "update:token"
|
||||
|
||||
|
||||
async def get_update_config() -> dict[str, str]:
|
||||
return {
|
||||
"gitea_url": await get_setting(UPDATE_GITEA_URL_KEY, "https://git.lanro.eu"),
|
||||
"repo": await get_setting(UPDATE_REPO_KEY, "Roman/ohm_streaming"),
|
||||
"token": await get_setting(UPDATE_TOKEN_KEY, ""),
|
||||
}
|
||||
|
||||
|
||||
async def set_update_config(gitea_url: str, repo: str, token: str) -> None:
|
||||
await set_setting(UPDATE_GITEA_URL_KEY, gitea_url.rstrip("/"))
|
||||
await set_setting(UPDATE_REPO_KEY, repo.strip().strip("/"))
|
||||
await set_setting(UPDATE_TOKEN_KEY, token.strip())
|
||||
logger.info("Configuration de mise à jour enregistrée (%s)", repo)
|
||||
|
||||
+18
-21
@@ -1,6 +1,7 @@
|
||||
"""Mises à jour logicielles.
|
||||
|
||||
- Détection : dernier tag semver du dépôt Gitea via son API (jeton requis, repo privé).
|
||||
- Détection : dernier tag semver du dépôt Gitea public via son API (lecture anonyme),
|
||||
avec le message du tag comme patchnote.
|
||||
- Application : POST à Watchtower (compagnon docker-compose) qui tire la nouvelle
|
||||
image et recrée le conteneur — quelques secondes d'indisponibilité.
|
||||
"""
|
||||
@@ -12,7 +13,6 @@ import time
|
||||
import httpx
|
||||
|
||||
from app.config import get_settings
|
||||
from app.services.settings import get_update_config
|
||||
from app.version import get_version
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -20,11 +20,11 @@ logger = logging.getLogger(__name__)
|
||||
_TAG_RE = re.compile(r"^v?(\d+)\.(\d+)\.(\d+)$")
|
||||
_CACHE_TTL = 300.0 # secondes
|
||||
|
||||
_cache: dict[str, object] = {"checked_at": 0.0, "latest": None}
|
||||
_cache: dict[str, object] = {"checked_at": 0.0, "latest": None, "notes": None}
|
||||
|
||||
|
||||
class UpdateError(Exception):
|
||||
"""Erreur de mise à jour (config absente, Gitea injoignable, Watchtower KO)."""
|
||||
"""Erreur de mise à jour (Gitea injoignable, Watchtower KO)."""
|
||||
|
||||
|
||||
def parse_tag(tag: str) -> tuple[int, int, int] | None:
|
||||
@@ -41,51 +41,48 @@ def is_newer(latest: str, current: str) -> bool:
|
||||
|
||||
|
||||
async def fetch_latest_version(*, force: bool = False) -> str | None:
|
||||
"""Dernier tag semver du dépôt (cache 5 min). None si non configuré ou erreur."""
|
||||
"""Dernier tag semver du dépôt + patchnote (cache 5 min). None si erreur."""
|
||||
now = time.monotonic()
|
||||
latest_cache = _cache["latest"]
|
||||
if not force and latest_cache and now - float(_cache["checked_at"]) < _CACHE_TTL:
|
||||
return str(latest_cache) # type: ignore[arg-type]
|
||||
|
||||
config = await get_update_config()
|
||||
if not config["repo"]:
|
||||
return None
|
||||
|
||||
url = f"{config['gitea_url']}/api/v1/repos/{config['repo']}/tags?limit=20"
|
||||
# Jeton requis uniquement pour un dépôt privé (public en lecture : inutile)
|
||||
headers = {"Authorization": f"token {config['token']}"} if config["token"] else {}
|
||||
settings = get_settings()
|
||||
url = f"{settings.gitea_url}/api/v1/repos/{settings.gitea_repo}/tags?limit=20"
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client:
|
||||
resp = await client.get(url, headers=headers)
|
||||
resp = await client.get(url)
|
||||
resp.raise_for_status()
|
||||
tags = [t["name"] for t in resp.json() if parse_tag(t.get("name", ""))]
|
||||
tags = [
|
||||
(t["name"], (t.get("message") or "").strip())
|
||||
for t in resp.json()
|
||||
if parse_tag(t.get("name", ""))
|
||||
]
|
||||
except (httpx.HTTPError, ValueError, KeyError) as exc:
|
||||
logger.warning("Vérification de mise à jour impossible : %s", exc)
|
||||
return None
|
||||
|
||||
latest = max(tags, key=parse_tag) if tags else None # type: ignore[arg-type]
|
||||
_cache.update(checked_at=now, latest=latest)
|
||||
latest, notes = max(tags, key=lambda t: parse_tag(t[0])) if tags else (None, None)
|
||||
_cache.update(checked_at=now, latest=latest, notes=notes or None)
|
||||
if latest and is_newer(latest, get_version()):
|
||||
logger.info("Nouvelle version disponible : %s (courante %s)", latest, get_version())
|
||||
return latest
|
||||
|
||||
|
||||
def invalidate_cache() -> None:
|
||||
_cache.update(checked_at=0.0, latest=None)
|
||||
_cache.update(checked_at=0.0, latest=None, notes=None)
|
||||
|
||||
|
||||
async def status() -> dict[str, object]:
|
||||
"""État complet : version courante, dernière dispo, config."""
|
||||
config = await get_update_config()
|
||||
"""État complet : version courante, dernière dispo, patchnote."""
|
||||
current = get_version()
|
||||
latest = await fetch_latest_version()
|
||||
return {
|
||||
"current": current,
|
||||
"latest": latest,
|
||||
"notes": _cache["notes"],
|
||||
"update_available": bool(latest and is_newer(latest, current)),
|
||||
"configured": bool(config["repo"]),
|
||||
"docker": bool(get_settings().watchtower_url),
|
||||
"config": config,
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user