ifcurl/ifcurl/git.py

# IFC URL — resolve and render ifc:// URLs
# Copyright (C) 2026 Bruno Postle <bruno@postle.net>
#
# This file is part of IFC URL.
#
# IFC URL is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# IFC URL is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with IFC URL.  If not, see <http://www.gnu.org/licenses/>.

"""Fetch IFC file bytes from a git repository described by an IfcUrl.

Remote repositories are cloned as bare repos to a per-URL cache directory
under the OS cache directory.  Mutable refs (HEAD, branches) trigger a fetch
on each call to pick up upstream changes; immutable refs (commit hashes, tags)
reuse the cached clone without fetching.

Authentication
--------------
Pass a *token* string to inject it into HTTPS remote URLs as a credential.
SSH transport uses the platform key store and ignores the token.
See :mod:`ifcurl.auth` for loading tokens from the config file.
"""

from __future__ import annotations

import hashlib

# allows git import even if git executable isn't found during import
import logging
import os
import shutil
from pathlib import Path

_logger = logging.getLogger(__name__)

from platformdirs import user_cache_dir

os.environ.setdefault("GIT_PYTHON_REFRESH", "quiet")
try:
    import git
    import git.exc

    _HAS_GITPYTHON = True
except ImportError:
    _HAS_GITPYTHON = False

from ifcurl.url import IfcUrl


def fetch_ifc(ifc_url: IfcUrl, token: str | None = None) -> tuple[str, bytes, bool]:
    """Return ``(commit_hexsha, ifc_bytes, is_stale)`` for the file addressed by *ifc_url*.

    The commit hexsha is the resolved, immutable identifier for the ref —
    useful as a cache key even when the URL uses a mutable ref like a branch.

    *is_stale* is ``True`` when the URL uses a mutable ref (branch/HEAD) and
    the remote fetch failed — the returned bytes come from the local cache and
    may not reflect the latest commit.  A warning is logged in this case.

    :param ifc_url: A parsed :class:`IfcUrl`.
    :param token: Optional bearer token for HTTPS authentication.  Injected
        as ``https://<token>@host/path``.  Ignored for SSH and local repos.
    :raises ImportError: If GitPython is not installed.
    :raises ValueError: If ``ifc_url.path`` is unset, the repo cannot be
        reached, or the file is not found at the specified ref.
    """
    if not _HAS_GITPYTHON:
        raise ImportError(
            "GitPython is not installed.  Install with: pip install gitpython"
        )
    if ifc_url.path is None:
        raise ValueError("URL has no 'path' parameter — cannot fetch IFC file")

    repo, is_stale = _get_repo(ifc_url, token=token)
    hexsha, data = _read_commit_blob(repo, ifc_url.git_ref(), ifc_url.path)
    return hexsha, data, is_stale


def diff_text(base_url: IfcUrl, head_url: IfcUrl, token: str | None = None) -> str:
    """Return the raw ``git diff`` text between *base_url* and *head_url*.

    Both URLs must refer to the same IFC file path in the same repository.
    The diff is produced against the cached bare clone, so no additional
    network fetch is performed beyond what :func:`fetch_ifc` already did.

    :param base_url: The older (base) ref — typically the PR base commit.
    :param head_url: The newer (head) ref — typically the PR head commit.
    :param token: Optional HTTPS token for authentication.
    :raises ImportError: If GitPython is not installed.
    :raises ValueError: If the paths differ, a ref is not found, or the diff
        cannot be produced.
    """
    if not _HAS_GITPYTHON:
        raise ImportError(
            "GitPython is not installed.  Install with: pip install gitpython"
        )
    if base_url.path is None or head_url.path is None:
        raise ValueError("Both URLs must have a 'path' parameter")
    if base_url.path != head_url.path:
        raise ValueError(
            f"base and head URLs must refer to the same IFC path "
            f"(got {base_url.path!r} vs {head_url.path!r})"
        )

    repo = _get_repo(base_url, token=token)
    try:
        base_hexsha = repo.commit(base_url.git_ref()).hexsha
        head_hexsha = repo.commit(head_url.git_ref()).hexsha
    except (git.exc.BadName, git.exc.BadObject) as exc:
        raise ValueError(f"Ref not found: {exc}") from exc

    try:
        return repo.git.diff(base_hexsha, head_hexsha, "--", base_url.path)
    except git.exc.GitCommandError as exc:
        raise ValueError(f"git diff failed: {exc.stderr.strip()}") from exc


def fetch_ifc_bytes(ifc_url: IfcUrl, token: str | None = None) -> bytes:
    """Return the raw bytes of the IFC file addressed by *ifc_url*.

    :param ifc_url: A parsed :class:`IfcUrl`.
    :param token: Optional bearer token for HTTPS authentication.
    :raises ImportError: If GitPython is not installed.
    :raises ValueError: If ``ifc_url.path`` is unset, the repo cannot be
        reached, or the file is not found at the specified ref.
    """
    _, data, _ = fetch_ifc(ifc_url, token=token)
    return data


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


def _get_repo(ifc_url: IfcUrl, token: str | None = None) -> tuple[git.Repo, bool]:
    """Open or fetch the repository for *ifc_url*.  Returns ``(repo, is_stale)``."""
    if ifc_url.transport == "local":
        return _open_local(ifc_url.repo_path), False
    return _open_remote(ifc_url.git_remote_url(), ifc_url.is_mutable_ref(), token=token)


def _touch_cache(cache_dir: Path) -> None:
    """Update mtime of *cache_dir* to now for LRU access tracking."""
    try:
        os.utime(str(cache_dir), None)
    except OSError:
        pass


def _dir_size(path: Path) -> int:
    """Return total byte size of all files under *path*."""
    return sum(f.stat().st_size for f in path.rglob("*") if f.is_file())


def _get_max_cache_bytes() -> int | None:
    """Return configured max disk cache size from IFCURL_CACHE_MAX_GB, or None."""
    val = os.environ.get("IFCURL_CACHE_MAX_GB")
    if val is None:
        return None
    try:
        return int(float(val) * 1024**3)
    except (ValueError, OverflowError):
        return None


def _repo_cache_entries() -> list[tuple[float, int, Path, str]]:
    """Return all cached repo dirs as ``(last_access, size_bytes, path, remote_url)``.

    Sorted oldest-first by last access.  Dirs without a ``repo.git``
    subdirectory are ignored.
    """
    base = Path(user_cache_dir("ifcurl"))
    if not base.exists():
        return []
    entries = []
    for entry in base.iterdir():
        if not entry.is_dir() or not (entry / "repo.git").exists():
            continue
        try:
            atime = entry.stat().st_mtime
            size = _dir_size(entry)
            url_file = entry / "remote_url"
            url = url_file.read_text().strip() if url_file.exists() else "?"
            entries.append((atime, size, entry, url))
        except OSError:
            pass
    entries.sort(key=lambda e: e[0])
    return entries


def _evict_if_needed() -> None:
    """Remove oldest cached repos until total disk usage is under the configured limit."""
    max_bytes = _get_max_cache_bytes()
    if max_bytes is None:
        return
    entries = _repo_cache_entries()
    total = sum(s for _, s, _, _ in entries)
    for _, size, cache_dir, _ in entries:
        if total <= max_bytes:
            break
        try:
            shutil.rmtree(str(cache_dir))
            total -= size
        except OSError:
            pass


def _open_local(repo_path: str) -> git.Repo:
    """Open a local git repository."""
    try:
        return git.Repo(repo_path, search_parent_directories=True)
    except git.exc.InvalidGitRepositoryError as exc:
        raise ValueError(f"Not a git repository: {repo_path!r}") from exc
    except git.exc.NoSuchPathError as exc:
        raise ValueError(f"Repository path does not exist: {repo_path!r}") from exc


def _cache_dir_for(remote_url: str) -> Path:
    """Return (and create) the per-URL cache directory for *remote_url*.

    Uses the OS-appropriate cache location:
      Linux   ~/.cache/ifcurl/<hash>
      macOS   ~/Library/Caches/ifcurl/<hash>
      Windows %LOCALAPPDATA%\\ifcurl\\Cache\\<hash>

    The cache key is derived from the clean URL without any token so the same
    cached clone is shared regardless of which credential was used to fetch it.
    """
    url_hash = hashlib.sha256(remote_url.encode()).hexdigest()[:24]
    cache_dir = Path(user_cache_dir("ifcurl")) / url_hash
    cache_dir.mkdir(parents=True, exist_ok=True)
    return cache_dir


def _auth_url(remote_url: str, token: str | None) -> str:
    """Return *remote_url* with the token injected, or the original URL."""
    if token and (
        remote_url.startswith("https://") or remote_url.startswith("http://")
    ):
        from ifcurl.auth import inject_token

        return inject_token(remote_url, token)
    return remote_url


def _http_fallback(url: str) -> str | None:
    """Return the http:// equivalent of an https:// URL, or None."""
    if url.startswith("https://"):
        return "http" + url[5:]
    return None


def _clone_bare(auth_url: str, git_dir: Path, remote_url: str) -> git.Repo:
    """Clone *auth_url* as a bare repo to *git_dir*, falling back to http
    when the server does not speak HTTPS (common for local dev instances).
    """
    try:
        return git.Repo.clone_from(auth_url, str(git_dir), bare=True)
    except git.exc.GitCommandError as exc:
        fallback = _http_fallback(auth_url)
        if fallback:
            try:
                return git.Repo.clone_from(fallback, str(git_dir), bare=True)
            except git.exc.GitCommandError:
                pass  # report the original error
        raise ValueError(
            f"Failed to clone {remote_url!r}: {exc.stderr.strip()}"
        ) from exc


def _open_remote(
    remote_url: str, is_mutable: bool, token: str | None = None
) -> tuple[git.Repo, bool]:
    """Return ``(repo, is_stale)`` for *remote_url*, cloning if necessary.

    Bare clones are stored under the OS cache dir keyed on the clean URL.
    For mutable refs (branches, HEAD) the remote is fetched on every call.
    Immutable refs (commit hashes, tags) skip the fetch.

    *is_stale* is ``True`` when a mutable-ref fetch was attempted but all
    network attempts failed — the caller receives cached data that may be
    outdated.  A warning is logged in this case.

    If *token* is provided it is injected into the URL for clone and fetch
    operations but is never written to the on-disk git config.

    HTTPS URLs automatically fall back to HTTP when the server does not
    speak TLS — this handles local Gitea instances running on HTTP.
    """
    cache_dir = _cache_dir_for(remote_url)
    git_dir = cache_dir / "repo.git"
    auth = _auth_url(remote_url, token)
    is_stale = False

    if not git_dir.exists():
        repo = _clone_bare(auth, git_dir, remote_url)
        (cache_dir / "remote_url").write_text(remote_url)
        _evict_if_needed()
    else:
        repo = git.Repo(str(git_dir))
        if is_mutable:
            fetch_ok = False
            try:
                if auth != remote_url:
                    # Fetch directly from the authenticated URL so the token
                    # is never stored in the repo config.
                    fetch_url = auth
                    fallback = _http_fallback(auth)
                    try:
                        repo.git.fetch(fetch_url, "+refs/*:refs/*")
                        fetch_ok = True
                    except git.exc.GitCommandError:
                        if fallback:
                            try:
                                repo.git.fetch(fallback, "+refs/*:refs/*")
                                fetch_ok = True
                            except git.exc.GitCommandError:
                                pass
                else:
                    repo.git.fetch("origin")
                    fetch_ok = True
            except git.exc.GitCommandError:
                pass  # offline — fall through to stale check

            if not fetch_ok:
                is_stale = True
                _logger.warning(
                    "fetch failed for %r — serving stale cached data", remote_url
                )
            _evict_if_needed()

    _touch_cache(cache_dir)
    return repo, is_stale


def _read_commit_blob(
    repo: git.Repo, git_ref: str, file_path: str
) -> tuple[str, bytes]:
    """Return ``(commit_hexsha, bytes)`` for *file_path* at *git_ref* in *repo*."""
    try:
        commit = repo.commit(git_ref)
    except (git.exc.BadName, git.exc.BadObject) as exc:
        raise ValueError(f"Ref {git_ref!r} not found in repository") from exc

    try:
        obj = commit.tree
        for part in file_path.strip("/").split("/"):
            obj = obj[part]
    except KeyError as exc:
        raise ValueError(f"File {file_path!r} not found at ref {git_ref!r}") from exc

    if obj.type != "blob":
        raise ValueError(f"{file_path!r} is a directory, not a file")

    return commit.hexsha, obj.data_stream.read()