"""Gitea integration – push/pull between canonical repo and Gitea's on-disk repo.""" import logging import os import subprocess from pathlib import Path from typing import Dict, Optional from .mirror import Mirror logger = logging.getLogger(__name__) class GiteaError(Exception): pass def _get_gitea_repo_path(mirror: Mirror) -> Path: """Return the filesystem path to Gitea's bare repo for this mirror.""" try: return Path(mirror.config.gitea.repo_dir) except Exception as e: raise GiteaError(str(e)) # ─── Git subprocess helper ───────────────────────────────────── def _git(*args: str, timeout: int = 300, check: bool = True) -> str: """Run a git command (uses CWD for repo discovery).""" try: r = subprocess.run( ["git"] + list(args), capture_output=True, text=True, timeout=timeout, env={**os.environ, "LC_ALL": "C"}, ) except FileNotFoundError: raise GiteaError("git not found on PATH") except subprocess.TimeoutExpired: raise GiteaError(f"git timed out: {' '.join(args)}") if check and r.returncode != 0: raise GiteaError( f"git failed (exit {r.returncode}): " f"git {' '.join(args)}\n{r.stderr.strip()}" ) return r.stdout.strip() if r.returncode == 0 else "" # ─── Push canonical → Gitea ─────────────────────────────────── def push_to_gitea(mirror: Mirror) -> bool: """Sync canonical refs to Gitea's bare repo via direct ref updates. Uses ``git update-ref`` to bypass Gitea's pre-receive hook, then fires Gitea's ``post-receive`` hook manually so the UI refreshes. Returns True if anything changed. """ gitea_path = _get_gitea_repo_path(mirror) if not gitea_path.exists(): logger.warning("Gitea repo not found at %s – skipping push", gitea_path) return False canonical = str(mirror.canonical_dir) # Collect canonical refs out = _git("--git-dir", canonical, "for-each-ref", "--format=%(objectname) %(refname)", "refs/heads/", "refs/tags/", timeout=30, check=False) canonical_refs: Dict[str, str] = {} for line in (out or "").splitlines(): line = line.strip() if not line: continue obj_hash, ref = line.split(None, 1) canonical_refs[ref] = obj_hash changed: list = [] for ref, obj_hash in canonical_refs.items(): gitea_hash = _git("--git-dir", str(gitea_path), "rev-parse", "--verify", "--quiet", ref, timeout=15, check=False) if obj_hash != gitea_hash: if gitea_hash: _git("--git-dir", str(gitea_path), "update-ref", ref, obj_hash, gitea_hash, timeout=15) else: _git("--git-dir", str(gitea_path), "update-ref", ref, obj_hash, timeout=15) changed.append((gitea_hash or "0" * 40, obj_hash, ref)) logger.debug("Updated %s in Gitea", ref) # Prune refs in Gitea not in canonical gitea_out = _git("--git-dir", str(gitea_path), "for-each-ref", "--format=%(objectname) %(refname)", "refs/heads/", "refs/tags/", timeout=30, check=False) for line in (gitea_out or "").splitlines(): line = line.strip() if not line: continue obj_hash, ref = line.split(None, 1) if ref not in canonical_refs: _git("--git-dir", str(gitea_path), "update-ref", "-d", ref, obj_hash, timeout=15) changed.append((obj_hash, "0" * 40, ref)) logger.debug("Pruned %s from Gitea", ref) if not changed: logger.debug("No new commits to push to Gitea") return False logger.info("Synced %d ref(s) to Gitea", len(changed)) _trigger_gitea_post_receive(changed, mirror.config.gitea) return True def _trigger_gitea_post_receive(changed: list, gitea_cfg): """Fire Gitea's post-receive hook so the UI refreshes immediately.""" candidates = ("/usr/local/bin/gitea", "/usr/bin/gitea", "gitea") hook_bin = None for c in candidates: if subprocess.run(["which", c], capture_output=True, check=False).returncode == 0: hook_bin = c break if not hook_bin: logger.debug("gitea binary not found – skipping post-receive hook") return try: config_path = os.environ.get("GITEA_CONFIG", "/etc/gitea/app.ini") lines = "\n".join(f"{o} {n} {r}" for o, n, r in changed) r = subprocess.run( [hook_bin, "hook", "--config", config_path, "post-receive"], input=lines.encode(), capture_output=True, timeout=60, ) if r.returncode != 0: err = (r.stderr or b"").decode(errors="replace").strip() logger.warning("Gitea post-receive hook failed: %s", err) else: logger.debug("Gitea post-receive hook fired") except Exception as e: logger.warning("Could not fire Gitea post-receive hook: %s", e) # ─── Fetch Gitea → canonical ────────────────────────────────── def fetch_from_gitea(mirror: Mirror, ref: Optional[str] = None) -> bool: """Fetch refs from Gitea's bare repo into the canonical repo. If *ref* is given (e.g. ``refs/heads/master``), only that ref is fetched. Otherwise all heads are fetched. Returns True if anything new was fetched. """ gitea_path = _get_gitea_repo_path(mirror) if not gitea_path.exists(): raise GiteaError(f"Gitea repo not found at {gitea_path}") canonical = str(mirror.canonical_dir) if ref: refspec = f"+{ref}:refs/remotes/gitea/{ref.removeprefix('refs/heads/')}" else: refspec = "+refs/heads/*:refs/remotes/gitea/*" logger.info("Fetching %s → canonical …", gitea_path) _git( "--git-dir", canonical, "fetch", "--prune", str(gitea_path), refspec, timeout=300, ) logger.debug("Fetch from Gitea complete") # Fast-forward local branches from remote-tracking refs if not ref: _ff_local_branches(canonical) return True def _ff_local_branches(git_dir: str): """Fast-forward ``refs/heads/*`` from ``refs/remotes/gitea/*``.""" out = _git("--git-dir", git_dir, "for-each-ref", "--format=%(refname)", "refs/remotes/gitea/", timeout=30, check=False) for rt_ref in out.splitlines(): rt_ref = rt_ref.strip() if not rt_ref: continue branch = rt_ref.removeprefix("refs/remotes/gitea/") local_ref = f"refs/heads/{branch}" # Check if local branch exists local_hash = _git("--git-dir", git_dir, "rev-parse", "--verify", "--quiet", local_ref, timeout=15, check=False) if not local_hash: _git("--git-dir", git_dir, "update-ref", local_ref, rt_ref, timeout=15) logger.debug("Created new branch %s from Gitea", local_ref) continue # Fast-forward if possible rt_hash = _git("--git-dir", git_dir, "rev-parse", "--verify", rt_ref, timeout=15, check=False) if not rt_hash or rt_hash == local_hash: continue base = _git("--git-dir", git_dir, "merge-base", local_hash, rt_ref, timeout=15, check=False) if base == local_hash: _git("--git-dir", git_dir, "update-ref", local_ref, rt_ref, timeout=15) logger.debug("Fast-forward %s from Gitea", local_ref)