Suport de repos privats: token de Gitea global configurable des del menú

This commit is contained in:
2026-05-29 22:30:11 +02:00
parent bfa01f31e3
commit 667eade660
4 changed files with 79 additions and 25 deletions
+46 -19
View File
@@ -8,6 +8,7 @@ from __future__ import annotations
import datetime as _dt
import json
import os
import shutil
import subprocess
import urllib.error
@@ -28,20 +29,39 @@ 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))
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 _run_git(args: list[str], cwd: Path | None, log: LogFn, token: str = "") -> str:
"""Ejecuta git capturando salida; lanza RuntimeError si falla.
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.
"""
cmd = ["git", *_auth_args(token), *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).
proc = subprocess.run(
cmd,
cwd=str(cwd) if cwd else None,
capture_output=True,
text=True,
env={**os.environ, "GIT_TERMINAL_PROMPT": "0"},
)
if proc.stdout:
log(proc.stdout.rstrip())
emit(proc.stdout.rstrip())
if proc.stderr:
log(proc.stderr.rstrip())
emit(proc.stderr.rstrip())
if proc.returncode != 0:
raise RuntimeError(f"git {' '.join(args)} falló (código {proc.returncode})")
return proc.stdout.strip()
@@ -52,7 +72,7 @@ def is_installed(root: Path, game: Game) -> bool:
return (repo_dir(root, game.id) / ".git").exists()
def check_update(root: Path, game: Game, log: LogFn = _noop) -> bool:
def check_update(root: Path, game: Game, log: LogFn = _noop, token: str = "") -> 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
@@ -61,7 +81,7 @@ def check_update(root: Path, game: Game, log: LogFn = _noop) -> bool:
repo = repo_dir(root, game.id)
if not (repo / ".git").exists():
return False
_run_git(["fetch", "origin", "--prune"], repo, log)
_run_git(["fetch", "origin", "--prune"], repo, log, token=token)
target = (
load_meta(root, game.id).default_branch
or _detect_origin_head(repo, log)
@@ -86,7 +106,7 @@ def delete_local(root: Path, game: Game, log: LogFn = _noop) -> None:
shutil.rmtree(target, ignore_errors=True)
def download(root: Path, game: Game, log: LogFn = _noop) -> GameMeta:
def download(root: Path, game: Game, log: LogFn = _noop, token: str = "") -> 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.
@@ -94,11 +114,11 @@ def download(root: Path, game: Game, log: LogFn = _noop) -> GameMeta:
repo = repo_dir(root, game.id)
repo.parent.mkdir(parents=True, exist_ok=True)
branch = _fetch_default_branch(game, log)
branch = _fetch_default_branch(game, log, token)
if (repo / ".git").exists():
log(f"Actualitzant {game.name} (forçat, descartant canvis locals)…")
_run_git(["fetch", "origin", "--prune"], repo, log)
_run_git(["fetch", "origin", "--prune"], repo, log, token=token)
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)
@@ -106,20 +126,24 @@ def download(root: Path, game: Game, log: LogFn = _noop) -> GameMeta:
log(f"Clonant {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)
_run_git(["clone", game.clone_url, str(repo)], None, log, token=token)
return refresh_metadata(root, game, branch, log)
return refresh_metadata(root, game, branch, log, token)
def refresh_metadata(
root: Path, game: Game, branch: str | None = None, log: LogFn = _noop
root: Path,
game: Game,
branch: str | None = None,
log: LogFn = _noop,
token: str = "",
) -> 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)
api = _fetch_gitea_info(game, log, token)
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)
@@ -170,11 +194,14 @@ def _read_version(game: Game, repo: Path, log: LogFn) -> str:
return out.splitlines()[0] if out else ""
def _fetch_gitea_info(game: Game, log: LogFn) -> dict | None:
def _fetch_gitea_info(game: Game, log: LogFn, token: str = "") -> 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={"Accept": "application/json"})
req = urllib.request.Request(game.info_url, headers=headers)
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:
@@ -182,8 +209,8 @@ def _fetch_gitea_info(game: Game, log: LogFn) -> dict | None:
return None
def _fetch_default_branch(game: Game, log: LogFn) -> str | None:
info = _fetch_gitea_info(game, log)
def _fetch_default_branch(game: Game, log: LogFn, token: str = "") -> str | None:
info = _fetch_gitea_info(game, log, token)
if info:
return info.get("default_branch")
return None
+2
View File
@@ -15,6 +15,7 @@ SETTINGS_NAME = "settings.json"
class Settings:
hide_not_downloaded: bool = False
updates_pending: list[str] = field(default_factory=list) # ids con update pendiente
gitea_token: str = "" # token personal de Gitea para repos privados (no se versiona)
def settings_path() -> Path:
@@ -32,6 +33,7 @@ def load_settings() -> Settings:
return Settings(
hide_not_downloaded=bool(data.get("hide_not_downloaded", False)),
updates_pending=list(data.get("updates_pending", [])),
gitea_token=str(data.get("gitea_token", "")),
)
+25 -2
View File
@@ -7,6 +7,8 @@ from pathlib import Path
from PySide6.QtCore import QThreadPool, Qt
from PySide6.QtGui import QAction
from PySide6.QtWidgets import (
QInputDialog,
QLineEdit,
QMainWindow,
QMessageBox,
QPlainTextEdit,
@@ -101,6 +103,27 @@ class MainWindow(QMainWindow):
self.action_delete.toggled.connect(self._set_delete_mode)
menu.addAction(self.action_delete)
menu.addSeparator()
self.action_token = QAction("Configura el token de Gitea…", self)
self.action_token.triggered.connect(self._configure_token)
menu.addAction(self.action_token)
def _configure_token(self) -> None:
token, ok = QInputDialog.getText(
self,
"Token de Gitea",
"Token personal d'accés (per a repos privats).\n"
"Es guarda local a settings.json (no es versiona).",
QLineEdit.Password,
self.settings.gitea_token,
)
if not ok:
return
self.settings.gitea_token = token.strip()
save_settings(self.settings)
estat = "configurat" if self.settings.gitea_token else "esborrat"
self._log(f"Token de Gitea {estat}.")
def _on_toggle_hide(self, checked: bool) -> None:
self.settings.hide_not_downloaded = checked
save_settings(self.settings)
@@ -116,7 +139,7 @@ class MainWindow(QMainWindow):
def _check_updates(self) -> None:
self.action_check.setEnabled(False)
self._log("=== Comprovant actualitzacions ===")
worker = CheckUpdatesWorker(self.root, self.config.games)
worker = CheckUpdatesWorker(self.root, self.config.games, self.settings.gitea_token)
worker.signals.log.connect(self._log)
worker.signals.result.connect(self._mark_update)
worker.signals.finished.connect(self._check_done)
@@ -185,7 +208,7 @@ class MainWindow(QMainWindow):
row.set_busy(True, "Descarregant…")
self._log(f"=== Descàrrega: {game.name} ===")
worker = DownloadWorker(self.root, game)
worker = DownloadWorker(self.root, game, self.settings.gitea_token)
worker.signals.log.connect(self._log)
worker.signals.finished.connect(lambda _meta, g=game: self._download_done(g))
worker.signals.error.connect(lambda msg, g=game: self._op_error(g, msg))
+6 -4
View File
@@ -25,16 +25,17 @@ class _Signals(QObject):
class DownloadWorker(QRunnable):
"""Clona o actualiza (forzado) y refresca la metadata de un juego."""
def __init__(self, root: Path, game: Game) -> None:
def __init__(self, root: Path, game: Game, token: str = "") -> None:
super().__init__()
self.root = root
self.game = game
self.token = token
self.signals = _Signals()
def run(self) -> None: # noqa: D401 - API de QRunnable
try:
meta: GameMeta = gitops.download(
self.root, self.game, log=self.signals.log.emit
self.root, self.game, log=self.signals.log.emit, token=self.token
)
except Exception as exc: # noqa: BLE001 - reportar a la UI
self.signals.error.emit(str(exc))
@@ -66,10 +67,11 @@ class CheckUpdatesWorker(QRunnable):
Emite `result(game_id, has_update)` por cada juego instalado y `finished` al final.
"""
def __init__(self, root: Path, games: list[Game]) -> None:
def __init__(self, root: Path, games: list[Game], token: str = "") -> None:
super().__init__()
self.root = root
self.games = games
self.token = token
self.signals = _Signals()
def run(self) -> None: # noqa: D401 - API de QRunnable
@@ -79,7 +81,7 @@ class CheckUpdatesWorker(QRunnable):
continue
try:
has_update = gitops.check_update(
self.root, game, log=self.signals.log.emit
self.root, game, log=self.signals.log.emit, token=self.token
)
except Exception as exc: # noqa: BLE001 - no abortar el resto
self.signals.log.emit(f"check {game.id}: {exc}")