111 lines
4.0 KiB
Python
111 lines
4.0 KiB
Python
"""Mises à jour logicielles.
|
|
|
|
- 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é.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
import time
|
|
|
|
import httpx
|
|
|
|
from app.config import get_settings
|
|
from app.version import get_version
|
|
|
|
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, "notes": None}
|
|
|
|
|
|
class UpdateError(Exception):
|
|
"""Erreur de mise à jour (Gitea injoignable, Watchtower KO)."""
|
|
|
|
|
|
def parse_tag(tag: str) -> tuple[int, int, int] | None:
|
|
"""'v0.2.1' → (0, 2, 1) ; None si le tag n'est pas un semver strict."""
|
|
m = _TAG_RE.match(tag.strip())
|
|
return tuple(int(g) for g in m.groups()) if m else None # type: ignore[return-value]
|
|
|
|
|
|
def is_newer(latest: str, current: str) -> bool:
|
|
a, b = parse_tag(latest), parse_tag(current)
|
|
if a is None or b is None:
|
|
return False
|
|
return a > b
|
|
|
|
|
|
async def fetch_latest_version(*, force: bool = False) -> str | None:
|
|
"""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]
|
|
|
|
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)
|
|
resp.raise_for_status()
|
|
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, 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, notes=None)
|
|
|
|
|
|
async def status() -> dict[str, object]:
|
|
"""É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)),
|
|
"docker": bool(get_settings().watchtower_url),
|
|
}
|
|
|
|
|
|
async def trigger_update() -> dict[str, str]:
|
|
"""Demande à Watchtower de recréer le conteneur avec la dernière image.
|
|
|
|
Le conteneur courant (donc cette requête) disparaît quelques secondes après :
|
|
la réponse est renvoyée immédiatement, le frontend gère la reconnexion.
|
|
"""
|
|
settings = get_settings()
|
|
if not settings.watchtower_url:
|
|
raise UpdateError(
|
|
"Watchtower non configuré — mise à jour disponible uniquement en déploiement Docker"
|
|
)
|
|
headers = {}
|
|
if settings.watchtower_token:
|
|
headers["Authorization"] = f"Bearer {settings.watchtower_token}"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.post(f"{settings.watchtower_url}/v1/update", headers=headers)
|
|
resp.raise_for_status()
|
|
except httpx.HTTPError as exc:
|
|
raise UpdateError(f"Watchtower injoignable : {exc}") from exc
|
|
logger.info("Mise à jour déclenchée via Watchtower (version courante %s)", get_version())
|
|
return {"status": "started"}
|