353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""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 os
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from .config import Game
|
|
from .metadata import GameMeta, icon_path, load_meta, save_meta
|
|
from .paths import game_dir, metadata_dir, repo_dir
|
|
|
|
LogFn = Callable[[str], None]
|
|
|
|
|
|
# --- Tolerancia a repos offline / inalcanzables ----------------------------
|
|
# Dos mecanismos complementarios para que git no se cuelgue:
|
|
# 1) low-speed abort: si una transferencia baja de `stall_limit` bytes/s durante
|
|
# `stall_time` segundos, git la aborta (transferencias que se estancan a media
|
|
# descarga).
|
|
# 2) techo duro: timeout en subprocess.run que mata git pase lo que pase (cubre
|
|
# cuelgues de conexión TCP/DNS cuando el host está offline, donde aún no fluyen
|
|
# bytes y el low-speed no llega a dispararse).
|
|
# Los valores son configurables desde settings.json (ver settings.Settings).
|
|
@dataclass(frozen=True)
|
|
class NetConfig:
|
|
fetch_timeout: float = 60 # techo para fetch / comprobar update (operación ligera)
|
|
clone_timeout: float = 900 # techo para clone (puede traer un repo grande)
|
|
http_timeout: float = 15 # techo para la API de Gitea (urllib)
|
|
stall_limit: int = 1000 # bytes/s por debajo de los cuales se considera estancado
|
|
stall_time: int = 20 # segundos por debajo del límite -> abortar
|
|
|
|
|
|
DEFAULT_NET = NetConfig()
|
|
|
|
# Windows: evita que cada `git` (aplicación de consola) abra una ventana negra
|
|
# cuando el lanzador corre como GUI sin consola (el .exe compilado con Nuitka
|
|
# --windows-console-mode=disable). En el resto de SO es 0 (sin efecto).
|
|
_NO_WINDOW = subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
|
|
|
|
|
|
def _noop(_: str) -> None:
|
|
pass
|
|
|
|
|
|
def _force_rmtree(path: Path, log: LogFn = _noop) -> None:
|
|
"""Esborra un arbre de fitxers, fins i tot amb fitxers de només-lectura.
|
|
|
|
A Windows els objectes de ``.git`` es creen com a només-lectura i fan que
|
|
``shutil.rmtree`` falli amb ``PermissionError``; cal llevar el bit d'escriptura
|
|
i reintentar. A diferència de ``ignore_errors=True``, aquí els errors que no
|
|
puguem resoldre es registren al log en lloc d'empassar-se en silenci.
|
|
"""
|
|
if not path.exists():
|
|
return
|
|
|
|
def handle(func: Callable, p: str, _exc: object) -> None:
|
|
try:
|
|
os.chmod(p, stat.S_IWRITE)
|
|
func(p) # reintenta l'operació (unlink/rmdir) que havia fallat
|
|
except OSError as exc:
|
|
log(f"No s'ha pogut esborrar {p}: {exc}")
|
|
|
|
# Python 3.12 va renombrar el paràmetre `onerror` a `onexc`; suportem tots dos.
|
|
if sys.version_info >= (3, 12):
|
|
shutil.rmtree(path, onexc=handle)
|
|
else:
|
|
shutil.rmtree(path, onerror=handle)
|
|
|
|
|
|
def _auth_args(token: str) -> list[str]:
|
|
"""Args -c para autenticar git ante Gitea con un token, sin tocar .git/config."""
|
|
if not token:
|
|
return []
|
|
return ["-c", f"http.extraHeader=Authorization: token {token}"]
|
|
|
|
|
|
def _net_args(net: NetConfig | None) -> list[str]:
|
|
"""Args -c para abortar transferencias estancadas (solo afectan al transporte http)."""
|
|
if net is None:
|
|
return []
|
|
return [
|
|
"-c",
|
|
f"http.lowSpeedLimit={net.stall_limit}",
|
|
"-c",
|
|
f"http.lowSpeedTime={net.stall_time}",
|
|
]
|
|
|
|
|
|
def _run_git(
|
|
args: list[str],
|
|
cwd: Path | None,
|
|
log: LogFn,
|
|
token: str = "",
|
|
timeout: float | None = None,
|
|
net: NetConfig | None = None,
|
|
) -> str:
|
|
"""Ejecuta git capturando salida; lanza RuntimeError si falla o agota `timeout`.
|
|
|
|
Si se pasa `token`, inyecta la cabecera de autorización de Gitea (para clone/fetch
|
|
de repos privados) y la redacta en el log para no filtrarla. Si se pasa `net`
|
|
añade el low-speed abort; `timeout` impone un techo duro (mata git si se cuelga).
|
|
"""
|
|
cmd = ["git", *_auth_args(token), *_net_args(net), *args]
|
|
|
|
def emit(line: str) -> None:
|
|
log(line.replace(token, "***") if token else line)
|
|
|
|
# Echo del comando: nunca mostramos los args -c con el token.
|
|
emit("$ " + " ".join(["git", *args]))
|
|
# GIT_TERMINAL_PROMPT=0: si falta auth en un repo privado, falla rápido
|
|
# en vez de colgarse esperando credenciales (no hay terminal en la GUI).
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
cwd=str(cwd) if cwd else None,
|
|
capture_output=True,
|
|
text=True,
|
|
env={**os.environ, "GIT_TERMINAL_PROMPT": "0"},
|
|
timeout=timeout,
|
|
creationflags=_NO_WINDOW,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
emit(
|
|
f"git {' '.join(args)} ha excedit el temps ({timeout:.0f}s): "
|
|
"el repositori no respon o no és accessible."
|
|
)
|
|
raise RuntimeError(
|
|
f"git {' '.join(args)} ha excedit el temps ({timeout:.0f}s)"
|
|
) from None
|
|
if proc.stdout:
|
|
emit(proc.stdout.rstrip())
|
|
if proc.stderr:
|
|
emit(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 check_update(
|
|
root: Path,
|
|
game: Game,
|
|
log: LogFn = _noop,
|
|
token: str = "",
|
|
net: NetConfig = DEFAULT_NET,
|
|
) -> bool:
|
|
"""Hace fetch y devuelve True si el clone local está por detrás del remoto.
|
|
|
|
No modifica el árbol de trabajo: solo cuenta los commits de origin/<rama> que no
|
|
están en HEAD. Devuelve False si el juego no está descargado.
|
|
"""
|
|
repo = repo_dir(root, game.id)
|
|
if not (repo / ".git").exists():
|
|
return False
|
|
_run_git(
|
|
["fetch", "origin", "--prune"],
|
|
repo,
|
|
log,
|
|
token=token,
|
|
timeout=net.fetch_timeout,
|
|
net=net,
|
|
)
|
|
target = (
|
|
load_meta(root, game.id).default_branch
|
|
or _detect_origin_head(repo, log)
|
|
or "HEAD"
|
|
)
|
|
try:
|
|
behind = _run_git(
|
|
["rev-list", "--count", f"HEAD..origin/{target}"], repo, log
|
|
)
|
|
except RuntimeError:
|
|
return False
|
|
return int(behind or "0") > 0
|
|
|
|
|
|
def delete_local(root: Path, game: Game, log: LogFn = _noop) -> None:
|
|
"""Esborra la descàrrega local del joc (clon + metadata), sense tocar el TOML."""
|
|
target = game_dir(root, game.id)
|
|
if not target.exists():
|
|
log(f"{game.name}: no hi ha res a esborrar")
|
|
return
|
|
log(f"Esborrant la descàrrega local de {game.name}…")
|
|
_force_rmtree(target, log)
|
|
|
|
|
|
def download(
|
|
root: Path,
|
|
game: Game,
|
|
log: LogFn = _noop,
|
|
token: str = "",
|
|
net: NetConfig = DEFAULT_NET,
|
|
) -> 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, token, net)
|
|
|
|
if (repo / ".git").exists():
|
|
log(f"Actualitzant {game.name} (forçat, descartant canvis locals)…")
|
|
_run_git(
|
|
["fetch", "origin", "--prune"],
|
|
repo,
|
|
log,
|
|
token=token,
|
|
timeout=net.fetch_timeout,
|
|
net=net,
|
|
)
|
|
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"Clonant {game.name}…")
|
|
if repo.exists(): # carpeta a medias sin .git: limpiarla
|
|
_force_rmtree(repo, log)
|
|
_run_git(
|
|
["clone", game.clone_url, str(repo)],
|
|
None,
|
|
log,
|
|
token=token,
|
|
timeout=net.clone_timeout,
|
|
net=net,
|
|
)
|
|
|
|
return refresh_metadata(root, game, branch, log, token, net)
|
|
|
|
|
|
def refresh_metadata(
|
|
root: Path,
|
|
game: Game,
|
|
branch: str | None = None,
|
|
log: LogFn = _noop,
|
|
token: str = "",
|
|
net: NetConfig = DEFAULT_NET,
|
|
) -> 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 + topics + fecha de creación desde la API de Gitea (best-effort).
|
|
api = _fetch_gitea_info(game, log, token, net)
|
|
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)
|
|
meta.topics = list(api.get("topics") or [])
|
|
meta.created_at = api.get("created_at", meta.created_at) or meta.created_at
|
|
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"(sense icona a {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"Icona actualitzada des de {game.icon_rel}")
|
|
except OSError as exc:
|
|
log(f"No s'ha pogut copiar la icona: {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,
|
|
creationflags=_NO_WINDOW,
|
|
)
|
|
except (OSError, subprocess.SubprocessError) as exc:
|
|
log(f"version_cmd ha fallat: {exc}")
|
|
return ""
|
|
out = (proc.stdout or "").strip()
|
|
return out.splitlines()[0] if out else ""
|
|
|
|
|
|
def _fetch_gitea_info(
|
|
game: Game, log: LogFn, token: str = "", net: NetConfig = DEFAULT_NET
|
|
) -> dict | None:
|
|
if not game.info_url:
|
|
return None
|
|
headers = {"Accept": "application/json"}
|
|
if token:
|
|
headers["Authorization"] = f"token {token}"
|
|
try:
|
|
req = urllib.request.Request(game.info_url, headers=headers)
|
|
with urllib.request.urlopen(req, timeout=net.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 s'ha pogut llegir la info de Gitea ({game.info_url}): {exc}")
|
|
return None
|
|
|
|
|
|
def _fetch_default_branch(
|
|
game: Game, log: LogFn, token: str = "", net: NetConfig = DEFAULT_NET
|
|
) -> str | None:
|
|
info = _fetch_gitea_info(game, log, token, net)
|
|
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
|