brain-of-reese/scripts/git_sync.py

"""Git source sync for import_docs (phase 28).

clone_or_pull(url, dest) clones ``url`` into ``dest`` (shallow, depth 1)
the first time, or fast-forwards an existing checkout with ``git pull
--ff-only`` on subsequent runs.

Auth: nothing special — an ``https://…`` URL uses the OS credential
helper / prompts; a ``git@host:repo.git`` URL uses the machine's SSH key.
No credentials are stored here; whatever the URL/SSH config supplies is
used.

This module is the only place the ``git`` CLI is invoked (A11: stdlib
``subprocess`` only, no new packages) — every git command goes through
:func:`run_git`: the clone/pull in :func:`clone_or_pull`, the per-file
last-commit-date walk in :func:`file_commit_dates` (phase 106), and the
docs-push sequence in :mod:`app.core.docs_push` (phase 59).

Per-file last-commit dates (phase 106, D2/D10) — behavior verified
against scratch repos 2026-09-13:

* a LOCAL-PATH checkout made by :func:`clone_or_pull` keeps FULL
  history (``git clone --depth 1 /local/path`` prints "--depth is
  ignored in local clones" and does not shallow) →
  :func:`file_commit_dates` yields TRUE per-file last-commit dates;
* a URL-TRANSPORT checkout (https/ssh/``file://``) is shallow, and in a
  shallow clone git reports the TIP commit as every existing file's
  last commit (the shallow boundary is each file's history root) → a
  uniform per-repo tip date: no intra-repo distortion, a real
  cross-source signal, refreshed on every pull.
"""
from __future__ import annotations

import logging
import subprocess
from datetime import datetime
from pathlib import Path

logger = logging.getLogger(__name__)

__all__ = ["GitSyncError", "clone_or_pull", "file_commit_dates", "run_git"]


class GitSyncError(RuntimeError):
    """A git clone/pull failed (or git is missing); carries git's stderr."""


def clone_or_pull(url: str, dest: Path | str) -> Path:
    """Clone ``url`` into ``dest`` (shallow, first run) or fast-forward it.

    - dest without a ``.git`` (or absent) → ``git clone --depth 1 url dest``
      (shallow: the KB is re-imported incrementally anyway).
    - dest with a ``.git`` → ``git pull --ff-only`` (refuses to merge
      unrelated histories — a broken checkout fails loudly rather than
      producing a dirty index).

    Returns the destination path. Raises :class:`GitSyncError` when git is
    missing or a git invocation exits non-zero (with git's stderr in the
    message, so the caller can name the failing repo + reason).
    """
    dest = Path(dest)
    if not dest.exists() or not (dest / ".git").exists():
        dest.parent.mkdir(parents=True, exist_ok=True)
        run_git(["git", "clone", "--depth", "1", url, str(dest)], cwd=dest.parent)
    else:
        run_git(["git", "pull", "--ff-only"], cwd=dest)
    return dest


def _parse_commit_dates(output: str) -> dict[str, datetime]:
    """Parse ``git log --name-only --format=@@%cI`` output (newest first).

    A ``@@<ISO-8601>`` line starts a commit (``%cI`` is ISO-strict, so
    the date is always aware — parsed with ``datetime.fromisoformat``);
    the following non-empty, non-``@@`` lines are repo-relative paths.
    The FIRST sighting of a path wins (the walk is newest-first) — that
    is the file's last-commit date. Paths are split on whitespace (like
    name-only output), ``\\``-normalized to ``/``, and a leading ``/``
    is stripped. Raises ``ValueError`` on a malformed commit date or a
    path line before any commit header (the caller fails soft).
    """
    dates: dict[str, datetime] = {}
    commit: datetime | None = None
    for line in output.splitlines():
        line = line.strip()
        if not line:
            continue
        if line.startswith("@@"):
            commit = datetime.fromisoformat(line[2:])
            continue
        if commit is None:
            raise ValueError(f"path line before any commit header: {line!r}")
        for raw_path in line.split():
            path = raw_path.replace("\\", "/").lstrip("/")
            if path:
                dates.setdefault(path, commit)
    return dates


def file_commit_dates(dest: Path | str) -> dict[str, datetime]:
    """Per-file last-commit dates for one checkout (phase 106, D2).

    ONE ``git log --name-only --format=@@%cI`` walk through
    :func:`run_git` (the A11 single-invocation site, one git call per
    source per sync) → ``{repo-relative POSIX path: last-commit
    datetime}``, newest-first so the first sighting of a path wins.

    The checkout behavior is pinned (verified 2026-09-13 — see the
    module docstring): a local-path ``clone_or_pull`` checkout keeps
    FULL history → TRUE per-file dates; a URL-transport checkout is
    shallow → the repo's TIP-commit date for every working-tree file
    (D10: uniform within the repo, real across sources).

    Fail-soft (pinned): a missing/non-directory checkout, a git failure
    (:class:`GitSyncError`), or ANY parse anomaly logs a warning and
    returns ``{}`` — the importer falls back to file mtimes; a date
    walk must never break a sync.
    """
    dest = Path(dest)
    if not dest.is_dir():
        logger.warning(
            "file_commit_dates: %s is not a directory — no git dates "
            "(the importer will fall back to file mtimes)",
            dest,
        )
        return {}
    try:
        output = run_git(["git", "log", "--name-only", "--format=@@%cI"], cwd=dest)
    except GitSyncError as exc:
        logger.warning(
            "file_commit_dates: git log failed for %s: %s — the importer "
            "will fall back to file mtimes",
            dest,
            exc,
        )
        return {}
    try:
        return _parse_commit_dates(output)
    except ValueError as exc:
        logger.warning(
            "file_commit_dates: unparseable git log output for %s (%s) — the "
            "importer will fall back to file mtimes",
            dest,
            exc,
        )
        return {}


def run_git(argv: list[str], cwd: Path) -> str:
    """Run one git command, capturing output; raise GitSyncError on failure.

    The single ``git`` invocation point for the whole app (A11). Every
    step of :func:`clone_or_pull` and of the docs-push sequence
    (:mod:`app.core.docs_push`, phase 59) goes through here, so error
    handling stays uniform: captured stdout on success, and
    :class:`GitSyncError` carrying git's stderr on a non-zero exit (or
    when the git binary is missing from PATH).
    """
    try:
        proc = subprocess.run(argv, cwd=cwd, capture_output=True, text=True)
    except FileNotFoundError:
        raise GitSyncError("git was not found on PATH — install git and retry") from None
    if proc.returncode != 0:
        raise GitSyncError(
            f"git {' '.join(argv[1:])} failed (exit {proc.returncode}): "
            f"{proc.stderr.strip()}"
        )
    return proc.stdout