govoplan/tools/release/govoplan_release/git_state.py

"""Read-only Git and version metadata collection."""

from __future__ import annotations

import json
from pathlib import Path
import re
import subprocess
import tomllib

from .model import RepositorySnapshot, RepositorySpec, VersionSnapshot
from .workspace import resolve_repo_path


def collect_repository_snapshot(
    spec: RepositorySpec,
    *,
    workspace_root: Path,
    target_tag: str | None,
    online: bool = False,
) -> RepositorySnapshot:
    path = resolve_repo_path(spec, workspace_root)
    if not path.exists():
        return RepositorySnapshot(spec=spec, absolute_path=str(path), exists=False, is_git=False, errors=("repository directory is missing",))
    if not (path / ".git").exists():
        return RepositorySnapshot(spec=spec, absolute_path=str(path), exists=True, is_git=False, errors=("not a git repository",))

    errors: list[str] = []
    status_result = git(path, "status", "--porcelain", timeout=10)
    if status_result.returncode != 0:
        message = git_error(status_result)
        if is_dubious_ownership_error(message):
            return RepositorySnapshot(
                spec=spec,
                absolute_path=str(path),
                exists=True,
                is_git=True,
                safe_directory_required=True,
                safe_directory_command=safe_directory_command(path),
                errors=(compact_git_error(message),),
            )
        return RepositorySnapshot(
            spec=spec,
            absolute_path=str(path),
            exists=True,
            is_git=True,
            errors=(compact_git_error(message),),
        )

    branch = git_text(path, "branch", "--show-current")
    has_head = git_success(path, "rev-parse", "--verify", "HEAD")
    head = git_text(path, "rev-parse", "--short", "HEAD") if has_head else None
    upstream = git_text(path, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{upstream}") if has_head else None
    ahead: int | None = None
    behind: int | None = None
    if upstream:
        counts = git_text(path, "rev-list", "--left-right", "--count", "@{upstream}...HEAD")
        parts = counts.split()
        if len(parts) == 2 and all(part.isdigit() for part in parts):
            behind = int(parts[0])
            ahead = int(parts[1])

    local_target_tag_exists: bool | None = None
    remote_target_tag_exists: bool | None = None
    if target_tag:
        local_target_tag_exists = git_success(path, "rev-parse", "-q", "--verify", f"refs/tags/{target_tag}")
        if online:
            remote_target_tag_exists = git_success(path, "ls-remote", "--exit-code", "--tags", "origin", f"refs/tags/{target_tag}", timeout=8)

    try:
        versions = collect_versions(path)
    except Exception as exc:  # noqa: BLE001 - surfaced in the dashboard.
        errors.append(str(exc))
        versions = VersionSnapshot()

    return RepositorySnapshot(
        spec=spec,
        absolute_path=str(path),
        exists=True,
        is_git=True,
        has_head=has_head,
        head=head,
        branch=branch or None,
        upstream=upstream or None,
        ahead=ahead,
        behind=behind,
        dirty_entries=tuple(line for line in status_result.stdout.splitlines() if line.strip()),
        versions=versions,
        local_target_tag_exists=local_target_tag_exists,
        remote_target_tag_exists=remote_target_tag_exists,
        remote_checked=online,
        errors=tuple(errors),
    )


def collect_versions(path: Path) -> VersionSnapshot:
    return VersionSnapshot(
        pyproject=read_pyproject_version(path),
        package=read_json_version(path / "package.json"),
        webui_package=read_json_version(path / "webui" / "package.json"),
        manifests=read_manifest_versions(path),
    )


def read_pyproject_version(path: Path) -> str | None:
    pyproject = path / "pyproject.toml"
    if not pyproject.exists():
        return None
    with pyproject.open("rb") as handle:
        payload = tomllib.load(handle)
    project = payload.get("project")
    if isinstance(project, dict) and isinstance(project.get("version"), str):
        return project["version"]
    return None


def read_json_version(path: Path) -> str | None:
    if not path.exists():
        return None
    payload = json.loads(path.read_text(encoding="utf-8"))
    version = payload.get("version") if isinstance(payload, dict) else None
    return version if isinstance(version, str) else None


def read_manifest_versions(path: Path) -> tuple[str, ...]:
    src = path / "src"
    if not src.exists():
        return ()
    versions: list[str] = []
    for manifest in sorted(src.glob("**/backend/manifest.py")):
        text = manifest.read_text(encoding="utf-8")
        match = re.search(r'(?m)^\s*version\s*=\s*["\']([^"\']+)["\']', text)
        if match is None:
            match = re.search(r'(?m)^MODULE_VERSION\s*=\s*["\']([^"\']+)["\']', text)
        if match is not None:
            versions.append(match.group(1))
    return tuple(versions)


def git_text(path: Path, *args: str, timeout: int = 10) -> str:
    result = git(path, *args, timeout=timeout)
    return result.stdout.strip() if result.returncode == 0 else ""


def git_success(path: Path, *args: str, timeout: int = 10) -> bool:
    return git(path, *args, timeout=timeout).returncode == 0


def git(path: Path, *args: str, timeout: int = 10) -> subprocess.CompletedProcess[str]:
    try:
        return subprocess.run(
            ["git", *args],
            cwd=path,
            check=False,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired as exc:
        return subprocess.CompletedProcess(["git", *args], 124, exc.stdout or "", exc.stderr or "git command timed out")


def git_error(result: subprocess.CompletedProcess[str]) -> str:
    return "\n".join(part for part in (result.stderr.strip(), result.stdout.strip()) if part)


def is_dubious_ownership_error(value: str) -> bool:
    return "detected dubious ownership in repository" in value or "safe.directory" in value


def compact_git_error(value: str) -> str:
    lines = [line.strip() for line in value.splitlines() if line.strip()]
    return lines[0] if lines else "git command failed"


def safe_directory_command(path: Path) -> str:
    return f"git config --global --add safe.directory {path}"