git_ops.py

"""Confirm-friendly git helpers for the scanned repo (status, stash, fetch, rebase, history).

Never force-pushes. Callers must confirm destructive-ish actions in the UI.
"""

from __future__ import annotations

import re
import subprocess
import time
from pathlib import Path
from typing import Any

_SAFE_REF = re.compile(r"^[A-Za-z0-9._/\-@{}]+$")


def _run(
    repo: Path,
    args: list[str],
    *,
    timeout: int = 120,
) -> tuple[int, str, str]:
    try:
        proc = subprocess.run(
            ["git", *args],
            cwd=str(repo),
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=timeout,
        )
        return proc.returncode, proc.stdout or "", proc.stderr or ""
    except (OSError, subprocess.TimeoutExpired) as exc:
        return 1, "", str(exc)


def _ok_ref(ref: str) -> bool:
    ref = (ref or "").strip()
    if not ref or len(ref) > 200:
        return False
    if ref.startswith("-"):
        return False
    return bool(_SAFE_REF.match(ref))


def rev_parse(repo: Path, rev: str = "HEAD") -> str:
    code, out, _ = _run(repo, ["rev-parse", rev], timeout=15)
    return (out or "").strip() if code == 0 else ""


def git_ref(repo: Path) -> dict[str, str]:
    """Current branch (or detached HEAD) + full commit sha."""
    code_b, out_b, _ = _run(repo, ["rev-parse", "--abbrev-ref", "HEAD"], timeout=15)
    branch = (out_b or "").strip() if code_b == 0 else ""
    if branch == "HEAD":
        branch = ""
    commit = rev_parse(repo, "HEAD")
    return {"branch": branch, "commit": commit}


def status(repo: Path) -> dict[str, Any]:
    ref = git_ref(repo)
    code_d, out_d, err_d = _run(repo, ["status", "--porcelain", "-b"], timeout=20)
    if code_d != 0 and not ref.get("commit"):
        return {
            "ok": False,
            "error": (err_d or "not a git repo").strip(),
            "branch": "",
            "commit": "",
            "short": "",
            "dirty": False,
            "upstream": "",
            "summary": "",
        }
    lines = [ln for ln in (out_d or "").splitlines() if ln.strip()]
    header = lines[0] if lines and lines[0].startswith("##") else ""
    porcelain = [ln for ln in lines if not ln.startswith("##")]
    upstream = ""
    if header:
        # ## main...origin/main [ahead 1]
        parts = header[2:].strip().split()
        if parts:
            head = parts[0]
            if "..." in head:
                upstream = head.split("...", 1)[1]
    short = (ref.get("commit") or "")[:7]
    return {
        "ok": True,
        "branch": ref.get("branch") or "",
        "commit": ref.get("commit") or "",
        "short": short,
        "dirty": bool(porcelain),
        "upstream": upstream,
        "summary": header[2:].strip() if header else "",
        "changed": len(porcelain),
    }


def stash(
    repo: Path,
    *,
    message: str | None = None,
    include_untracked: bool = True,
) -> dict[str, Any]:
    msg = (message or "codebase-graph stash").strip()[:200] or "codebase-graph stash"
    args = ["stash", "push"]
    if include_untracked:
        args.append("-u")
    args.extend(["-m", msg])
    code, out, err = _run(repo, args, timeout=120)
    st = status(repo)
    if code != 0:
        return {
            "ok": False,
            "error": (err or out or "git stash failed").strip(),
            **{k: st.get(k) for k in ("branch", "commit", "short", "dirty")},
        }
    return {
        "ok": True,
        "stdout": (out or err or "").strip(),
        **{k: st.get(k) for k in ("branch", "commit", "short", "dirty", "upstream", "summary")},
    }


def fetch(repo: Path, *, prune: bool = True) -> dict[str, Any]:
    args = ["fetch", "--all"]
    if prune:
        args.append("--prune")
    code, out, err = _run(repo, args, timeout=300)
    st = status(repo)
    if code != 0:
        return {
            "ok": False,
            "error": (err or out or "git fetch failed").strip(),
            **{k: st.get(k) for k in ("branch", "commit", "short", "dirty")},
        }
    return {
        "ok": True,
        "stdout": (out or err or "fetch ok").strip(),
        **{k: st.get(k) for k in ("branch", "commit", "short", "dirty", "upstream", "summary")},
    }


def rebase(repo: Path, onto: str | None = None) -> dict[str, Any]:
    """Rebase onto upstream (default) or an explicit ref. Never --force."""
    target = (onto or "").strip() or "@{upstream}"
    if target != "@{upstream}" and not _ok_ref(target):
        return {"ok": False, "error": "invalid rebase target"}
    code, out, err = _run(repo, ["rebase", target], timeout=300)
    st = status(repo)
    text = (err or out or "").strip()
    if code != 0:
        conflict = "conflict" in text.lower() or "could not apply" in text.lower()
        return {
            "ok": False,
            "error": text or "git rebase failed",
            "conflict": conflict,
            "hint": "Resolve conflicts, then `git rebase --continue`, or `git rebase --abort`.",
            **{k: st.get(k) for k in ("branch", "commit", "short", "dirty")},
        }
    return {
        "ok": True,
        "stdout": text or "rebase ok",
        "onto": target,
        **{k: st.get(k) for k in ("branch", "commit", "short", "dirty", "upstream", "summary")},
    }


def switch(
    repo: Path,
    branch: str,
    *,
    stash_if_dirty: bool = False,
) -> dict[str, Any]:
    ref = str(branch or "").strip()
    if not ref or not _ok_ref(ref):
        return {"ok": False, "error": "branch required"}
    before = status(repo)
    stashed = False
    if before.get("dirty") and stash_if_dirty:
        s = stash(repo, message=f"codebase-graph auto-stash before switch to {ref}")
        if not s.get("ok"):
            return {**s, "aborted": True}
        stashed = True
    elif before.get("dirty") and not stash_if_dirty:
        # Still attempt switch — git may refuse; surface the error.
        pass

    code, out, err = _run(repo, ["switch", ref], timeout=60)
    after = status(repo)
    if code != 0:
        return {
            "ok": False,
            "error": (err or out or "git switch failed").strip(),
            "stashed": stashed,
            **{k: after.get(k) for k in ("branch", "commit", "short", "dirty")},
        }
    return {
        "ok": True,
        "stashed": stashed,
        "from": {"branch": before.get("branch"), "commit": before.get("commit")},
        "stdout": (out or "").strip(),
        **{k: after.get(k) for k in ("branch", "commit", "short", "dirty", "upstream", "summary")},
    }


def file_log(repo: Path, path: str, *, limit: int = 20) -> dict[str, Any]:
    rel = str(path or "").replace("\\", "/").lstrip("/")
    if not rel or ".." in rel.split("/"):
        return {"ok": False, "error": "invalid path", "commits": []}
    lim = max(1, min(100, int(limit or 20)))
    code, out, err = _run(
        repo,
        [
            "log",
            "--follow",
            f"-n{lim}",
            "--pretty=format:%H%x09%h%x09%an%x09%ad%x09%s",
            "--date=short",
            "--",
            rel,
        ],
        timeout=60,
    )
    if code != 0:
        return {"ok": False, "error": (err or out or "git log failed").strip(), "commits": [], "path": rel}
    commits = []
    for line in (out or "").splitlines():
        parts = line.split("\t")
        if len(parts) < 5:
            continue
        commits.append(
            {
                "commit": parts[0],
                "short": parts[1],
                "author": parts[2],
                "date": parts[3],
                "subject": "\t".join(parts[4:]),
            }
        )
    return {"ok": True, "path": rel, "commits": commits}


def activity(
    repo: Path,
    path: str = "",
    *,
    days: int = 30,
    buckets: int = 12,
) -> dict[str, Any]:
    """Commit counts + unique authors for path (or whole repo) over the last N days."""
    rel = str(path or "").replace("\\", "/").lstrip("/")
    if ".." in rel.split("/"):
        return {
            "ok": False,
            "error": "invalid path",
            "path": rel,
            "days": days,
            "commits": 0,
            "authors": 0,
            "buckets": [],
        }
    days_n = max(1, min(365, int(days or 30)))
    bucket_n = max(1, min(52, int(buckets or 12)))
    cmd = [
        "log",
        f"--since={days_n} days ago",
        "--pretty=format:%at%x09%an",
    ]
    if rel:
        cmd.extend(["--", rel])
    code, out, err = _run(repo, cmd, timeout=90)
    if code != 0:
        return {
            "ok": False,
            "error": (err or out or "git log failed").strip(),
            "path": rel,
            "days": days_n,
            "commits": 0,
            "authors": 0,
            "buckets": [{"count": 0} for _ in range(bucket_n)],
        }

    now = time.time()
    window = float(days_n) * 86400.0
    start = now - window
    bucket_size = window / float(bucket_n)
    counts = [0] * bucket_n
    authors: set[str] = set()
    for line in (out or "").splitlines():
        parts = line.split("\t", 1)
        if len(parts) < 2:
            continue
        try:
            ts = float(parts[0])
        except ValueError:
            continue
        author = (parts[1] or "").strip()
        if author:
            authors.add(author)
        if ts < start:
            continue
        idx = int((ts - start) / bucket_size) if bucket_size > 0 else bucket_n - 1
        if idx < 0:
            idx = 0
        elif idx >= bucket_n:
            idx = bucket_n - 1
        counts[idx] += 1

    return {
        "ok": True,
        "path": rel,
        "days": days_n,
        "commits": sum(counts),
        "authors": len(authors),
        "buckets": [{"count": c} for c in counts],
    }


def file_show(repo: Path, path: str, commit: str) -> dict[str, Any]:
    rel = str(path or "").replace("\\", "/").lstrip("/")
    sha = str(commit or "").strip()
    if not rel or ".." in rel.split("/") or not _ok_ref(sha):
        return {"ok": False, "error": "invalid path or commit"}
    code, out, err = _run(repo, ["show", f"{sha}:{rel}"], timeout=60)
    if code != 0:
        return {"ok": False, "error": (err or out or "git show failed").strip(), "path": rel, "commit": sha}
    # Cap payload for Monaco
    text = out
    truncated = False
    if len(text.encode("utf-8", errors="replace")) > 1_500_000:
        text = text[:1_200_000]
        truncated = True
    return {
        "ok": True,
        "path": rel,
        "commit": sha,
        "short": sha[:7],
        "content": text,
        "truncated": truncated,
    }