app/tools/github_tool.py

import concurrent.futures
import os
import re
import shutil
import subprocess  # nosec B404 — subprocess is required for git diff; all inputs are validated before use
import uuid
from pathlib import Path
from threading import Lock

from git import Repo

from app.core.config import get_settings
from app.core.logger import get_logger

CLONE_TIMEOUT_SECONDS = 60


settings = get_settings()
logger = get_logger(__name__)
_clone_lock = Lock()


def _get_dir_size_mb(path: Path) -> float:
    total = sum(f.stat().st_size for f in path.rglob("*") if f.is_file())
    return total / (1024 * 1024)


def get_changed_files(local_path: str, base_sha: str, head_sha: str) -> list[str]:
    """Return list of .py files changed between two commits."""

    result = subprocess.run(  # nosec B603 B607 — hardcoded executable, no shell, shas are validated by git itself
        ["git", "diff", "--name-only", base_sha, head_sha],
        cwd=local_path,
        capture_output=True,
        text=True,
        timeout=30,
    )

    if result.returncode != 0:
        return []

    return [
        f
        for f in result.stdout.splitlines()
        if f.endswith(".py") and os.path.exists(os.path.join(local_path, f))
    ]


GITHUB_URL_PATTERN = re.compile(
    r"^https://github\.com/[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+(\.git)?$"
)


def clone_repository(github_url: str) -> str:
    if not GITHUB_URL_PATTERN.match(github_url.strip()):
        raise ValueError(
            f"Invalid GitHub URL: '{github_url}'. "
            "Must be in the format https://github.com/owner/repo"
        )
    repo_name = github_url.rstrip("/").split("/")[-1]
    if repo_name.endswith(".git"):
        repo_name = repo_name[:-4]

    clone_dir = Path(settings.repo_clone_dir)
    clone_dir.mkdir(parents=True, exist_ok=True)

    with _clone_lock:
        local_path = clone_dir / f"{repo_name}_{uuid.uuid4().hex[:8]}"
    try:
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
            future = ex.submit(
                lambda: Repo.clone_from(github_url, str(local_path), depth=1)
            )
            try:
                future.result(timeout=CLONE_TIMEOUT_SECONDS)
            except concurrent.futures.TimeoutError:
                shutil.rmtree(local_path, ignore_errors=True)
                raise TimeoutError(f"Clone timed out after {CLONE_TIMEOUT_SECONDS}s")
    except TimeoutError:
        raise
    except Exception:
        shutil.rmtree(local_path, ignore_errors=True)
        raise
    size_mb = _get_dir_size_mb(local_path)
    if size_mb > settings.max_repo_size_mb:
        shutil.rmtree(local_path, ignore_errors=True)
        raise ValueError(
            f"Repository size {size_mb:.1f}MB exceeds limit of {settings.max_repo_size_mb}MB"
        )
    return str(local_path)


def delete_repository(local_path: str) -> None:
    path = Path(local_path)
    if path.exists():
        shutil.rmtree(path, ignore_errors=True)
        logger.info("Deleted cloned repository", extra={"path": str(local_path)})


def resolve_remote_head_sha(github_url: str) -> str | None:
    """
    Resolve the current HEAD commit SHA of a remote repo without cloning,
    using `git ls-remote`. Used as the cache key for analysis results.

    Returns None if the SHA can't be determined (e.g. network error,
    private repo without credentials) — callers should treat this as
    "cache not applicable" and fall back to the uncached pipeline.
    """
    try:
        result = subprocess.run(  # nosec B603 B607 — hardcoded executable, no shell, URL is validated by caller
            ["git", "ls-remote", github_url, "HEAD"],
            capture_output=True,
            text=True,
            timeout=15,
        )
    except Exception:
        logger.warning("Failed to resolve remote HEAD sha", extra={"url": github_url})
        return None

    if result.returncode != 0 or not result.stdout.strip():
        return None

    return result.stdout.split()[0]