legal-ai/web/gitea_client.py

"""Gitea REST API client — create repos in the 'cases' org and push."""

from __future__ import annotations

import logging
import os
import subprocess
from pathlib import Path

import httpx

logger = logging.getLogger(__name__)

GITEA_ORG = "cases"


def _host() -> str:
    return os.environ.get("GITEA_HOST", "https://gitea.nautilus.marcusgroup.org")


def _token() -> str:
    return os.environ.get("GITEA_ACCESS_TOKEN") or os.environ.get("GITEA_TOKEN", "")


async def create_repo(case_number: str, title: str, description: str = "") -> dict:
    """Create a private repo in the 'cases' org on Gitea."""
    repo_name = case_number  # e.g. "1130-25"
    async with httpx.AsyncClient(verify=False, timeout=30) as client:
        resp = await client.post(
            f"{_host()}/api/v1/orgs/{GITEA_ORG}/repos",
            headers={"Authorization": f"token {_token()}"},
            json={
                "name": repo_name,
                "description": f"ערר {case_number} — {title}"[:255],
                "private": True,
                "auto_init": False,
            },
        )
        if resp.status_code == 409:
            # Repo already exists — fetch it
            resp2 = await client.get(
                f"{_host()}/api/v1/repos/{GITEA_ORG}/{repo_name}",
                headers={"Authorization": f"token {_token()}"},
            )
            resp2.raise_for_status()
            return resp2.json()
        resp.raise_for_status()
        return resp.json()


def _git_env() -> dict:
    return {
        "GIT_AUTHOR_NAME": "Ezer Mishpati",
        "GIT_AUTHOR_EMAIL": "legal@local",
        "GIT_COMMITTER_NAME": "Ezer Mishpati",
        "GIT_COMMITTER_EMAIL": "legal@local",
        "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
        "GIT_TERMINAL_PROMPT": "0",
    }


def _refresh_remote_url(case_dir: Path, env: dict) -> bool:
    """Re-inject current token into existing origin URL.

    Handles the case where the token in Infisical was rotated but the
    case repo's origin URL still has the old token embedded.
    Returns True if origin exists (and was refreshed), False otherwise.
    """
    result = subprocess.run(
        ["git", "remote", "get-url", "origin"],
        cwd=case_dir, capture_output=True, text=True,
    )
    if result.returncode != 0:
        return False
    current_url = result.stdout.strip()
    # Strip any embedded credentials (https://user:token@host -> https://host)
    if "@" in current_url and current_url.startswith("https://"):
        bare_url = "https://" + current_url.split("@", 1)[1]
    else:
        bare_url = current_url
    auth_url = bare_url.replace("https://", f"https://chaim:{_token()}@")
    if auth_url != current_url:
        subprocess.run(
            ["git", "remote", "set-url", "origin", auth_url],
            cwd=case_dir, capture_output=True, env=env,
        )
    return True


def commit_and_push(case_dir: str | Path, message: str) -> bool:
    """Stage, commit, refresh remote URL with current token, and push.

    Best-effort: returns False on any failure, but logs the reason at
    WARNING level so issues are visible (the previous code swallowed
    push failures silently).
    """
    case_dir = Path(case_dir)
    if not (case_dir / ".git").exists():
        return False

    env = _git_env()

    subprocess.run(["git", "add", "."], cwd=case_dir, capture_output=True, env=env)
    commit = subprocess.run(
        ["git", "commit", "-m", message],
        cwd=case_dir, capture_output=True, text=True, env=env,
    )
    if commit.returncode != 0 and "nothing to commit" not in commit.stdout:
        logger.warning("Git commit failed in %s: %s", case_dir, commit.stderr or commit.stdout)
        # Continue anyway — maybe there were no new changes but we still
        # want to push earlier unpushed commits.

    if not _refresh_remote_url(case_dir, env):
        logger.warning("No origin remote configured in %s — skipping push", case_dir)
        return False

    push = subprocess.run(
        ["git", "push"],
        cwd=case_dir, capture_output=True, text=True, env=env,
    )
    if push.returncode != 0:
        logger.warning("Git push failed in %s: %s", case_dir, push.stderr)
        return False
    return True


def setup_remote_and_push(case_dir: str | Path, repo_clone_url: str) -> bool:
    """Add Gitea as remote 'origin' (or update it) and push.

    Used on initial case creation. For routine commits + push on
    existing repos, prefer ``commit_and_push``.
    """
    case_dir = Path(case_dir)
    if not (case_dir / ".git").exists():
        return False

    env = _git_env()

    auth_url = repo_clone_url.replace("https://", f"https://chaim:{_token()}@")

    result = subprocess.run(
        ["git", "remote", "get-url", "origin"],
        cwd=case_dir, capture_output=True, text=True,
    )
    if result.returncode == 0:
        subprocess.run(
            ["git", "remote", "set-url", "origin", auth_url],
            cwd=case_dir, capture_output=True, env=env,
        )
    else:
        subprocess.run(
            ["git", "remote", "add", "origin", auth_url],
            cwd=case_dir, capture_output=True, env=env,
        )

    push_result = subprocess.run(
        ["git", "push", "-u", "origin", "main"],
        cwd=case_dir, capture_output=True, text=True, env=env,
    )
    if push_result.returncode != 0:
        push_result = subprocess.run(
            ["git", "push", "-u", "origin", "master"],
            cwd=case_dir, capture_output=True, text=True, env=env,
        )
    if push_result.returncode != 0:
        logger.warning("Git push failed: %s", push_result.stderr)
        return False
    return True