rpa_vision_v3/agent_v0/server_v1/update_check.py

# agent_v0/server_v1/update_check.py
"""Logique PURE de décision de mise à jour du client Léa (DETTE-022 v2).

But : centraliser, SANS dépendance FastAPI, le cœur testable de la MAJ
silencieuse :

  - `parse_version()` (R3) : parse une version semver en tuple d'entiers, pour
    une comparaison correcte ("1.0.2" < "1.0.10" — le piège lexicographique
    classique). Tolérant : préfixe « v », espaces, et format invalide → fallback
    `(0,)` (la plus basse) SANS jamais lever.
  - `decide_update()` (R2) : compare la version courante à la dernière dispo,
    choisit l'`update_type` (`code-only` par défaut, ~500 Ko / `full` ~33 Mo
    rare) et construit la réponse
    `{update_available, latest_version, update_type, url}`.

Ce module est volontairement IMPORTABLE seul (aucun import lourd, pas de
`api_stream`) pour être testé sans démarrer le serveur (DETTE-013). Le
branchement HTTP (endpoint gated) vit dans `api_stream.py`.

⚠️ Cette brique ne fait QUE décider. Le swap réel des fichiers, l'édition de
Lea.bat et le redémarrage sont HORS de ce module (réservé révision humaine).

Branche feat/push-log-dgx.
"""

from __future__ import annotations

from typing import Optional, Tuple

# Niveaux de livraison valides (R2). `code-only` par défaut = 99 % des MAJ.
VALID_UPDATE_TYPES = ("code-only", "full")
DEFAULT_UPDATE_TYPE = "code-only"

# Fallback de version « la plus basse » pour une chaîne illisible : ainsi une
# version valide est toujours > à une version invalide, et une *latest* illisible
# ne déclenche jamais de MAJ douteuse.
_FALLBACK_VERSION: Tuple[int, ...] = (0,)


def parse_version(v) -> Tuple[int, ...]:
    """Parse une version semver en tuple d'entiers (R3).

    "1.0.2" → (1, 0, 2), "1.0.10" → (1, 0, 10), "v1.2.3" → (1, 2, 3).

    Tolérant et SANS exception : préfixe « v/V » et espaces tolérés ; tout
    format non numérique (vide, None, "abc", "1.x.3") retombe sur `(0,)`.

    Stratégie : `packaging.version` si présent (déjà dans le venv via
    setuptools/pip), sinon parse manuel. Aucune nouvelle dépendance.
    """
    if not isinstance(v, str):
        return _FALLBACK_VERSION
    s = v.strip().lstrip("vV").strip()
    if not s:
        return _FALLBACK_VERSION
    try:
        from packaging.version import Version

        return tuple(Version(s).release)
    except Exception:
        # packaging absent (python-embed minimal) OU version non-PEP440.
        pass
    try:
        return tuple(int(x) for x in s.split("."))
    except (ValueError, AttributeError):
        return _FALLBACK_VERSION


def is_newer(candidate: str, baseline: str) -> bool:
    """True si `candidate` est strictement plus récent que `baseline` (semver)."""
    return parse_version(candidate) > parse_version(baseline)


def _normalize_update_type(update_type: Optional[str]) -> str:
    """Normalise l'update_type sur un niveau valide (défaut code-only)."""
    if update_type in VALID_UPDATE_TYPES:
        return update_type
    return DEFAULT_UPDATE_TYPE


def build_download_url(
    machine_id: Optional[str],
    version: str,
    update_type: str,
) -> str:
    """Construit l'URL de téléchargement RELATIVE (R2, 2 niveaux).

    Forme alignée sur les endpoints fleet existants :
        /api/fleet/download/<machine_id>?type=<update_type>&version=<version>

    On garde une URL relative : le client la résout contre son SERVER_BASE.
    `machine_id` absent → segment « default » (rétrocompatible).
    """
    mid = (machine_id or "default").strip() or "default"
    return f"/api/fleet/download/{mid}?type={update_type}&version={version}"


def decide_update(
    current_version: str,
    latest_version: str,
    update_type: Optional[str] = None,
    machine_id: Optional[str] = None,
) -> dict:
    """Décision PURE de mise à jour (R2 + R3).

    Compare `current_version` à `latest_version` en semver. Si la dernière est
    strictement plus récente, construit une réponse d'update ; sinon réponse
    « à jour ». Aucune exception : versions illisibles → pas de MAJ (prudence).

    Returns:
        {
          "update_available": bool,
          "latest_version": str,
          "update_type": "code-only" | "full" | None,   # None si pas de MAJ
          "url": str | None,                              # None si pas de MAJ
        }
    """
    no_update = {
        "update_available": False,
        "latest_version": latest_version,
        "update_type": None,
        "url": None,
    }

    # latest illisible → on ne propose RIEN (pas de MAJ douteuse).
    if parse_version(latest_version) == _FALLBACK_VERSION:
        return no_update

    if not is_newer(latest_version, current_version):
        return no_update

    chosen_type = _normalize_update_type(update_type)
    return {
        "update_available": True,
        "latest_version": latest_version,
        "update_type": chosen_type,
        "url": build_download_url(machine_id, latest_version, chosen_type),
    }