132 lines
4.1 KiB
Python
132 lines
4.1 KiB
Python
"""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 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:
|
||
"""Push all heads and tags from the canonical repo to Gitea's bare repo.
|
||
|
||
Returns True if anything was pushed.
|
||
"""
|
||
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)
|
||
|
||
# Check if there's anything to push by comparing refs
|
||
out = _git("--git-dir", canonical, "for-each-ref",
|
||
"--format=%(objectname) %(refname)",
|
||
"refs/heads/", "refs/tags/",
|
||
timeout=30, check=False)
|
||
if out:
|
||
has_new = False
|
||
for line in out.splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
obj_hash, ref = line.split(None, 1)
|
||
gitea_hash = _git("--git-dir", str(gitea_path), "rev-parse",
|
||
"--verify", "--quiet", ref,
|
||
timeout=30, check=False)
|
||
if obj_hash != gitea_hash:
|
||
has_new = True
|
||
break
|
||
if not has_new:
|
||
logger.debug("No new commits to push to Gitea")
|
||
return False
|
||
|
||
logger.info("Pushing canonical → Gitea (%s) …", gitea_path)
|
||
# Use `git push --prune` with explicit refspecs to mirror branches and tags
|
||
_git(
|
||
"--git-dir", canonical, "push", "--prune",
|
||
str(gitea_path),
|
||
"+refs/heads/*:refs/heads/*",
|
||
"+refs/tags/*:refs/tags/*",
|
||
timeout=300,
|
||
)
|
||
logger.info("Push to Gitea complete")
|
||
return True
|
||
|
||
|
||
# ─── 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}:{ref}"
|
||
else:
|
||
refspec = "+refs/heads/*:refs/heads/*"
|
||
|
||
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")
|
||
return True
|