jail-launcher/jlauncher/gitops.py

"""Operaciones git + refresco de metadata desde Gitea.

Todo lo que toca disco/red vive aquí; los workers (QThread) lo invocan en segundo plano.
Las funciones aceptan un callback ``log(str)`` opcional para emitir progreso a la UI.
"""

from __future__ import annotations

import datetime as _dt
import json
import shutil
import subprocess
import urllib.error
import urllib.request
from collections.abc import Callable
from pathlib import Path

from .config import Game
from .metadata import GameMeta, icon_path, load_meta, save_meta
from .paths import metadata_dir, repo_dir

LogFn = Callable[[str], None]

_HTTP_TIMEOUT = 15


def _noop(_: str) -> None:
    pass


def _run_git(args: list[str], cwd: Path | None, log: LogFn) -> str:
    """Ejecuta git capturando salida; lanza RuntimeError si falla."""
    cmd = ["git", *args]
    log("$ " + " ".join(cmd))
    proc = subprocess.run(
        cmd,
        cwd=str(cwd) if cwd else None,
        capture_output=True,
        text=True,
    )
    if proc.stdout:
        log(proc.stdout.rstrip())
    if proc.stderr:
        log(proc.stderr.rstrip())
    if proc.returncode != 0:
        raise RuntimeError(f"git {' '.join(args)} falló (código {proc.returncode})")
    return proc.stdout.strip()


def is_installed(root: Path, game: Game) -> bool:
    """Un juego está instalado si su clone existe y tiene un .git."""
    return (repo_dir(root, game.id) / ".git").exists()


def download(root: Path, game: Game, log: LogFn = _noop) -> GameMeta:
    """Clona (si no existe) o trae el remoto forzado (descartando cambios locales).

    Luego refresca la metadata (descripción Gitea + versión + icono) y la devuelve.
    """
    repo = repo_dir(root, game.id)
    repo.parent.mkdir(parents=True, exist_ok=True)

    branch = _fetch_default_branch(game, log)

    if (repo / ".git").exists():
        log(f"Actualizando {game.name} (forzado, descartando cambios locales)…")
        _run_git(["fetch", "origin", "--prune"], repo, log)
        target = branch or _detect_origin_head(repo, log) or "HEAD"
        _run_git(["reset", "--hard", f"origin/{target}"], repo, log)
        _run_git(["clean", "-fd"], repo, log)
    else:
        log(f"Clonando {game.name}…")
        if repo.exists():  # carpeta a medias sin .git: limpiarla
            shutil.rmtree(repo, ignore_errors=True)
        _run_git(["clone", game.clone_url, str(repo)], None, log)

    return refresh_metadata(root, game, branch, log)


def refresh_metadata(
    root: Path, game: Game, branch: str | None = None, log: LogFn = _noop
) -> GameMeta:
    """Reconstruye info.json + icon.png a partir del repo clonado y la API Gitea."""
    repo = repo_dir(root, game.id)
    meta = load_meta(root, game.id)

    # Descripción + rama por defecto desde la API de Gitea (best-effort).
    api = _fetch_gitea_info(game, log)
    if api is not None:
        meta.description = api.get("description", meta.description) or meta.description
        meta.default_branch = api.get("default_branch", meta.default_branch)
    elif branch:
        meta.default_branch = branch

    # Versión vía version_cmd dentro del repo (best-effort).
    meta.version = _read_version(game, repo, log) or meta.version

    meta.updated_at = _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds")

    # Copiar icono desde el repo a la cache.
    _copy_icon(root, game, repo, log)

    save_meta(root, game.id, meta)
    return meta


def _copy_icon(root: Path, game: Game, repo: Path, log: LogFn) -> None:
    src = repo / game.icon_rel
    if not src.exists():
        log(f"(sin icono en {game.icon_rel})")
        return
    metadata_dir(root, game.id).mkdir(parents=True, exist_ok=True)
    try:
        shutil.copyfile(src, icon_path(root, game.id))
        log(f"Icono actualizado desde {game.icon_rel}")
    except OSError as exc:
        log(f"No se pudo copiar el icono: {exc}")


def _read_version(game: Game, repo: Path, log: LogFn) -> str:
    if not game.version_cmd:
        return ""
    try:
        proc = subprocess.run(
            game.version_cmd,
            cwd=str(repo),
            shell=True,
            capture_output=True,
            text=True,
            timeout=20,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        log(f"version_cmd falló: {exc}")
        return ""
    out = (proc.stdout or "").strip()
    return out.splitlines()[0] if out else ""


def _fetch_gitea_info(game: Game, log: LogFn) -> dict | None:
    if not game.info_url:
        return None
    try:
        req = urllib.request.Request(game.info_url, headers={"Accept": "application/json"})
        with urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except (urllib.error.URLError, OSError, json.JSONDecodeError, ValueError) as exc:
        log(f"No se pudo leer la info de Gitea ({game.info_url}): {exc}")
        return None


def _fetch_default_branch(game: Game, log: LogFn) -> str | None:
    info = _fetch_gitea_info(game, log)
    if info:
        return info.get("default_branch")
    return None


def _detect_origin_head(repo: Path, log: LogFn) -> str | None:
    """Fallback: deduce la rama por defecto desde origin/HEAD."""
    try:
        ref = _run_git(
            ["symbolic-ref", "--short", "refs/remotes/origin/HEAD"], repo, log
        )
    except RuntimeError:
        return None
    # ref tiene forma "origin/main"
    return ref.split("/", 1)[1] if "/" in ref else ref or None