coding-agent-gitea/gitea/workspace.py

import os
import logging
import subprocess
import base64
import shutil
from pathlib import Path
from urllib.parse import urlparse
from .config import GITEA_REPOS_ROOT, GITEA_URL, GITEA_TOKEN
from gitea.client import GiteaClient

logger: logging.Logger = logging.getLogger("gitea-workspace")


class WorkspaceManager:
    """Manages local workspace for Gitea repositories."""

    def __init__(self) -> None:
        self.root_dir: Path = Path(GITEA_REPOS_ROOT).expanduser().resolve()
        self.root_dir.mkdir(parents=True, exist_ok=True)

    def _configure_repo_user(self, repo_path: Path) -> None:
        try:
            client = GiteaClient()
            user = client.get_authenticated_user()
            if not user or not user.login:
                raise RuntimeError("No authenticated user found.")
            username: str = user.login

            auth_str: str = f"{username}:{GITEA_TOKEN}"
            auth_bytes: bytes = auth_str.encode("utf-8")
            auth_b64: str = base64.b64encode(auth_bytes).decode("utf-8")

            # Configure extraHeader locally for the repo
            subprocess.run(
                ["git", "-C", str(repo_path), "config", "http.extraHeader", f"Authorization: Basic {auth_b64}"],
                check=True, capture_output=True
            )

            name: str = user.full_name or user.login
            email: str = user.email or f"{user.login}@noreply.gitea"
            subprocess.run(
                ["git", "-C", str(repo_path), "config", "user.name", name],
                check=True, capture_output=True
            )
            subprocess.run(
                ["git", "-C", str(repo_path), "config", "user.email", email],
                check=True, capture_output=True
            )
        except Exception as e:
            logger.error(f"Error configuring local git user: {e}")
            raise

    def get_repo_path(self, repo_full_name: str) -> Path:
        parts: list[str] = repo_full_name.split("/")
        return self.root_dir / parts[0] / parts[1]

    def _get_authenticated_url(self, repo_full_name: str) -> str:
        parsed = urlparse(GITEA_URL.rstrip("/"))
        path = parsed.path.rstrip("/")
        return f"{parsed.scheme}://{parsed.netloc}{path}/{repo_full_name}.git"

    def sanitize_repo(self, repo_full_name: str, repo_path: Path) -> None:
        try:
            auth_url = self._get_authenticated_url(repo_full_name)
            subprocess.run(
                ["git", "-C", str(repo_path), "remote", "set-url", "origin", auth_url],
                check=True, capture_output=True,
            )
            self._configure_repo_user(repo_path)

            # Check for any uncommitted changes or untracked files
            status_res = subprocess.run(
                ["git", "-C", str(repo_path), "status", "--porcelain"],
                check=True, capture_output=True, text=True
            )
            if status_res.stdout.strip():
                logger.info(f"Uncommitted changes detected in {repo_path}. Stashing before sanitization.")
                subprocess.run(
                    ["git", "-C", str(repo_path), "stash", "push", "-u", "-m", "Auto-backup before agent sanitization"],
                    check=True, capture_output=True
                )

            subprocess.run(
                ["git", "-C", str(repo_path), "reset", "--hard", "HEAD"],
                check=True, capture_output=True,
            )
            subprocess.run(
                ["git", "-C", str(repo_path), "clean", "-fdx"],
                check=True, capture_output=True,
            )
            try:
                subprocess.run(
                    ["git", "-C", str(repo_path), "checkout", "main"],
                    check=True, capture_output=True,
                )
            except subprocess.CalledProcessError:
                subprocess.run(
                    ["git", "-C", str(repo_path), "checkout", "master"],
                    check=True, capture_output=True,
                )
            try:
                subprocess.run(
                    ["git", "-C", str(repo_path), "pull", "origin", "main"],
                    check=True, capture_output=True,
                )
            except subprocess.CalledProcessError:
                subprocess.run(
                    ["git", "-C", str(repo_path), "pull", "origin", "master"],
                    check=True, capture_output=True,
                )
        except Exception as e:
            logger.error(f"Error during sanitization: {e}", exc_info=True)
            raise RuntimeError(f"Failed to sanitize repository {repo_full_name} at {repo_path}: {e}") from e

    def clone_repo(self, repo_full_name: str, clone_url: str | None = None) -> Path:
        repo_path: Path = self.get_repo_path(repo_full_name)
        if repo_path.exists():
            if not (repo_path / ".git").exists():
                new_path: Path = repo_path.parent / f"{repo_full_name.replace('/', '_')}_old"
                if new_path.exists():
                    shutil.rmtree(new_path)
                repo_path.rename(new_path)
            return repo_path

        logger.info(f"Cloning repository {repo_full_name} to {repo_path}...")
        auth_url = self._get_authenticated_url(repo_full_name)

        client = GiteaClient()
        user = client.get_authenticated_user()
        if not user or not user.login:
            raise RuntimeError("No authenticated user found.")
        username: str = user.login

        auth_str: str = f"{username}:{GITEA_TOKEN}"
        auth_bytes: bytes = auth_str.encode("utf-8")
        auth_b64: str = base64.b64encode(auth_bytes).decode("utf-8")
        subprocess.run(
            ["git", "clone", "-c", f"http.extraHeader=Authorization: Basic {auth_b64}", auth_url, str(repo_path)],
            check=True, capture_output=True
        )
        self._configure_repo_user(repo_path)
        return repo_path