Git, GitHub, and Configuration¶

def get_git_status(directory: str) -> dict[str, str]:
    """Get Git status for files relative to a given directory.

    Runs ``git status --porcelain -z --untracked-files=all`` from the repository root
    and maps every changed/untracked path back to a path relative to *directory*,
    filtering out files that live outside of it.

    Status characters returned:

    - ``'U'``: Untracked (``??`` in porcelain output)
    - ``'M'``: Modified (working-tree or staged modification)
    - ``'A'``: Added / staged for the first time (includes renames)
    - ``'D'``: Deleted (working-tree or staged deletion)

    Args:
        directory: Absolute path to the directory being visualised. Must be inside a Git
            repository.

    Returns:
        ``{relative_path: status_char}`` where *relative_path* uses forward slashes
        regardless of OS, or an empty dict when Git is unavailable or the directory is
        not tracked.
    """
    import subprocess

    try:
        root_result = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],
            cwd=directory,
            capture_output=True,
            text=True,
        )
        if root_result.returncode != 0:
            return {}
        git_root = os.path.realpath(root_result.stdout.strip())
        directory = os.path.realpath(directory)

        status_result = subprocess.run(
            ["git", "status", "--porcelain", "-z", "--untracked-files=all"],
            cwd=git_root,
            capture_output=True,
            text=True,
        )
        if status_result.returncode != 0:
            return {}

        status_map: dict[str, str] = {}
        records = status_result.stdout.split("\0")
        i = 0
        while i < len(records):
            entry = records[i]
            i += 1
            if len(entry) < 4:
                continue
            xy = entry[:2]
            path = entry[3:]

            x, y = xy[0], xy[1]
            if x == "R" or x == "C":
                i += 1
            if x == "?" and y == "?":
                status = "U"
            elif x == "D" or y == "D":
                status = "D"
            elif x == "A" or x == "R":
                status = "A"
            else:
                status = "M"

            abs_file = os.path.normpath(
                os.path.join(git_root, path.replace("/", os.sep))
            )
            try:
                rel = os.path.relpath(abs_file, directory)
                outside = rel == os.pardir or rel.startswith(os.pardir + os.sep)
                if not outside:
                    status_map[rel.replace(os.sep, "/")] = status
            except ValueError:
                pass

        return status_map
    except Exception as e:
        logger.debug("Could not get git status for %s: %s", directory, e)
        return {}

def get_github_token() -> str | None:
    """Return a GitHub token from the environment, if configured.

    Looks up ``GITHUB_TOKEN`` first and then ``GH_TOKEN``. When set, the token is sent
    with archive and ref requests, raising rate limits and permitting access to private
    repositories.

    Returns:
        The token string, or ``None`` when neither variable is set.
    """
    for var in ("GITHUB_TOKEN", "GH_TOKEN"):
        token = os.environ.get(var)
        if token:
            return token.strip()
    return None

def parse_github_url(text: str) -> GitHubTarget | None:
    """Parse a GitHub repository URL into a
    [`GitHubTarget`][recursivist.github.GitHubTarget].

    Accepts the common HTTPS forms (with or without scheme, ``www.`` or a trailing
    ``.git``), an optional ``/tree/<ref>[/<subpath>]`` or ``/blob/<ref>/<subpath>``
    selector, and the SSH form ``git@github.com:owner/repo.git``. Percent-encoded
    characters in the ref and subpath (e.g. ``%20``, ``%23``) are decoded, so URLs
    copied from a browser address bar resolve to the real names. The scheme and host are
    matched case-insensitively, as URL hosts are; owner, repository, ref and subpath are
    kept exactly as written.

    A form without a scheme (``github.com/owner/repo``) is also a valid relative
    filesystem path, e.g. a GOPATH-style ``src/github.com/golang/go`` checkout entered
    from ``src``. When such an argument names an existing local file or directory, it is
    treated as that local path and ``None`` is returned. A URL with an explicit
    ``http(s)://`` scheme, or the SSH form, is always treated as GitHub.

    The subpath is returned as written, whether it names a directory or a file: the two
    cannot be told apart from the URL alone. A subpath that turns out to be a file — as
    in the ``/blob/<ref>/<file>`` URL of a GitHub file page — is resolved to its
    containing directory by
    [`checkout_repository`][recursivist.github.checkout_repository].

    When a ``/tree`` or ``/blob`` selector is present, the segment immediately after it
    is taken as the ref and everything beyond it as the subpath. Refs that themselves
    contain slashes (e.g. ``feature/x``) therefore cannot be distinguished from a
    subpath by URL alone; pass such a repository without a selector, or pin the ref with
    a plain branch name.

    Args:
        text: The raw argument to parse.

    Returns:
        The parsed [`GitHubTarget`][recursivist.github.GitHubTarget], or ``None`` when
        *text* is not a recognizable GitHub URL.
    """
    if not text or "github.com" not in text.lower():
        return None
    text = text.strip()
    if _HAS_SCHEME_RE.match(text) is None and os.path.exists(text):
        return None
    ssh = _SSH_RE.match(text)
    if ssh:
        return GitHubTarget(
            owner=ssh.group("owner"), repo=_strip_git_suffix(ssh.group("repo"))
        )
    match = _HTTP_RE.match(text)
    if not match:
        return None
    ref = match.group("ref")
    if ref is not None:
        ref = urllib.parse.unquote(ref)
    subpath = urllib.parse.unquote(match.group("subpath") or "").strip("/")
    return GitHubTarget(
        owner=match.group("owner"),
        repo=_strip_git_suffix(match.group("repo")),
        ref=ref,
        subpath=subpath,
    )

def resolve_default_branch(target: GitHubTarget, token: str | None = None) -> str:
    """Resolve a repository's default branch without using the REST API.

    Reads the symbolic ``HEAD`` reference from the repository's Git smart-HTTP
    ``info/refs`` advertisement, which is not subject to the REST API's unauthenticated
    rate limit.

    Args:
        target: The repository whose default branch is wanted.
        token: Optional GitHub token for private repositories.

    Returns:
        The default branch name (e.g. ``"main"``).

    Raises:
        GitHubError: If the repository is missing or private without a valid token, or
            if the default branch cannot be determined.
    """
    payload = _fetch_refs_advertisement(target, token)
    match = re.search(rb"symref=HEAD:refs/heads/([^\x00 \n]+)", payload)
    if not match:
        raise GitHubError(
            f"Could not determine the default branch for '{target.slug}'."
        )
    return match.group(1).decode("utf-8", "replace")

def resolve_commit_shas(
    target: GitHubTarget,
    refs: list[str | None],
    token: str | None = None,
) -> list[str | None]:
    """Resolve each ref in *refs* to a commit SHA using one advertisement fetch.

    A single ``info/refs`` request is made and reused for every ref, so this is cheap
    even for several refs on the same repository. Each entry is resolved as follows:

    * ``None`` resolves to the commit the default branch (``HEAD``) points at.
    * A branch or tag name resolves to its tip commit; annotated tags resolve to the
      commit they dereference to.
    * A value that is not an advertised ref but looks like a commit SHA (7-40 hex
      characters) is returned as-is, lowercased, so explicit commit pins are supported.
    * Anything else resolves to ``None``.

    Args:
        target: The repository to resolve against.
        refs: The refs to resolve, in order. ``None`` means the default branch.
        token: Optional GitHub token for private repositories.

    Returns:
        A list the same length as *refs*, each a lowercase commit SHA or ``None`` when
        the ref could not be resolved.

    Raises:
        GitHubError: If the repository is missing, private, or unreachable.
    """
    advertised = _parse_advertised_refs(_fetch_refs_advertisement(target, token))
    resolved: list[str | None] = []
    for ref in refs:
        if ref is None:
            resolved.append(advertised.get("HEAD"))
            continue
        sha = (
            advertised.get(f"refs/heads/{ref}")
            or advertised.get(f"refs/tags/{ref}")
            or advertised.get(ref)
        )
        if sha is None and _SHA_RE.fullmatch(ref):
            sha = ref.lower()
        resolved.append(sha)
    return resolved

def commit_shas_equal(sha1: str | None, sha2: str | None) -> bool:
    """Return whether two commit SHAs identify the same commit.

    Handles abbreviated SHAs (as short as 7 characters, Git's conventional minimum) by
    treating one as a match for the other when it is a case-insensitive prefix. ``None``
    never matches.

    Args:
        sha1: The first commit SHA, or ``None``.
        sha2: The second commit SHA, or ``None``.

    Returns:
        ``True`` if both are non-``None`` and identify the same commit.
    """
    if sha1 is None or sha2 is None:
        return False
    a, b = sha1.lower(), sha2.lower()
    if a == b:
        return True
    shorter, longer = (a, b) if len(a) <= len(b) else (b, a)
    return len(shorter) >= 7 and longer.startswith(shorter)

def same_github_target(
    target1: GitHubTarget,
    target2: GitHubTarget,
    token: str | None = None,
) -> bool:
    """Return whether two GitHub targets refer to the same scanned tree.

    Owner and repository names are compared case-insensitively because GitHub treats
    them that way, while the subpath is compared case-sensitively because file paths are
    case-sensitive.

    When both sides pin the same ref (or neither does, so both use the default branch),
    no network access is needed. Otherwise the two refs are resolved to the commits they
    point at and compared, so that distinct refs that name the same commit — a branch
    and a tag on the same tip, a branch and the default branch, or a branch and an
    explicit commit SHA — are recognized as the same. If either ref cannot be resolved
    (repository missing, private, unreachable, or the ref does not exist), the targets
    are treated as *not* the same so the normal comparison flow can surface the real
    error rather than a misleading "compare with itself" message.

    Subpaths are compared as given. A subpath that names a file is only resolved to its
    containing directory by
    [`checkout_repository`][recursivist.github.checkout_repository], so two URLs for
    different files in one directory are recognized as the same tree only when the
    targets passed here are the checked-out ones (`RepoCheckout.target`).

    Args:
        target1: The first GitHub target.
        target2: The second GitHub target.
        token: Optional GitHub token used for the ref lookups.

    Returns:
        ``True`` if both targets resolve to the same repository, commit and subtree,
        else ``False``.
    """
    if (
        target1.owner.lower() != target2.owner.lower()
        or target1.repo.lower() != target2.repo.lower()
        or target1.subpath != target2.subpath
    ):
        return False
    if target1.ref == target2.ref:
        return True
    try:
        sha1, sha2 = resolve_commit_shas(target1, [target1.ref, target2.ref], token)
    except GitHubError:
        return False
    return commit_shas_equal(sha1, sha2)

@contextmanager
def checkout_repository(
    target: GitHubTarget, token: str | None = None
) -> Generator[RepoCheckout]:
    """Download and extract a GitHub repository into a temporary directory.

    Resolves the ref (using the default branch when the target does not pin one),
    downloads the source archive, and safely extracts it. The extracted files are
    removed when the context exits.

    When the target's subpath names a file rather than a directory, the directory
    containing that file is checked out instead, and the yielded checkout's ``target``
    carries that directory as its subpath.

    Args:
        target: The repository (and optional subtree) to check out.
        token: Optional GitHub token; defaults to
            [`get_github_token`][recursivist.github.get_github_token].

    Yields:
        A [`RepoCheckout`][recursivist.github.RepoCheckout] describing the local
        extraction.

    Raises:
        GitHubError: If the repository cannot be resolved, downloaded, or extracted, or
            the requested subpath does not exist in it.
    """
    if token is None:
        token = get_github_token()
    ref = target.ref or resolve_default_branch(target, token)
    temp_dir = tempfile.mkdtemp(prefix="recursivist-gh-")
    try:
        archive_path = os.path.join(temp_dir, "archive.tar.gz")
        logger.debug("Downloading %s at ref '%s'", target.slug, ref)
        _download_archive(target, ref, token, archive_path)
        extract_dir = os.path.join(temp_dir, "extracted")
        os.makedirs(extract_dir, exist_ok=True)
        _safe_extract(archive_path, extract_dir)
        os.remove(archive_path)
        local_root, subpath = _locate_root(extract_dir, target)
        if subpath != target.subpath:
            logger.info(
                "'%s' is a file in '%s'; scanning its containing directory instead",
                target.subpath,
                target.slug,
            )
            target = replace(target, subpath=subpath)
        yield RepoCheckout(
            target=target,
            local_root=local_root,
            ref=ref,
            root_name=target.display_name,
        )
    finally:
        shutil.rmtree(temp_dir, ignore_errors=True)

def apply_github_urls(structure: Directory, checkout: RepoCheckout) -> Directory:
    """Rewrite each file's display path to its GitHub blob URL, in place.

    Used when ``--full-path`` is requested for a GitHub input: it walks *structure* and
    replaces every [`FileEntry`][recursivist._models.FileEntry] ``path`` with the file's
    canonical blob URL, so the renderers and exporters display GitHub URLs instead of
    temporary filesystem paths.

    Args:
        structure: A scanned structure produced by
            [`recursivist.scanner.get_directory_structure`][recursivist.scanner.get_directory_structure]
            for the checkout's `RepoCheckout.local_root`.
        checkout: The checkout the structure was scanned from, supplying the owner,
            repo, ref, and subpath used to build URLs.

    Returns:
        The same *structure* object, with file paths rewritten.
    """
    target = checkout.target
    base_prefix = target.subpath.strip("/")

    def _walk(node: Directory, rel_dir: str) -> None:
        rewritten: list[FileEntry] = []
        for entry in node.files:
            rel_file = f"{rel_dir}/{entry.name}" if rel_dir else entry.name
            url = target.blob_url(checkout.ref, rel_file)
            rewritten.append(entry._replace(path=url))
        node.files = rewritten
        for name, content in node.subdirectories.items():
            next_dir = f"{rel_dir}/{name}" if rel_dir else name
            _walk(content, next_dir)

    _walk(structure, base_prefix)
    return structure

def accepts_value(key: str, value: Any) -> bool:
    """Return whether *value* is one the configuration *key* accepts.

    A key with a set of choices accepts exactly those strings; any other key accepts a
    string that is not empty or made of whitespace only. A key in `LIST_KEYS` accepts a
    list of one or more such strings, and every other key a single one.

    Args:
        key: A key of `CONFIG_KEYS`, in its underscored form.
        value: The value to check, of any type.

    Returns:
        ``True`` if *key* accepts *value*, ``False`` otherwise.
    """
    choices = CONFIG_KEYS[key]

    def accepts_string(item: Any) -> bool:
        if not isinstance(item, str):
            return False
        return bool(item.strip()) if choices is None else item in choices

    if key in LIST_KEYS:
        return (
            isinstance(value, list)
            and bool(value)
            and all(accepts_string(item) for item in value)
        )
    return accepts_string(value)

def describe_accepted_values(key: str) -> str:
    """Return a phrase naming the values the configuration *key* accepts.

    Args:
        key: A key of `CONFIG_KEYS`, in its underscored form.

    Returns:
        A phrase that completes the sentence "Use ...": the quoted choices of a key
        that has them (``'emoji' or 'nerd'``), ``a list of one or more non-empty
        strings`` for a key in `LIST_KEYS`, and ``a non-empty string`` for any other
        key.
    """
    choices = CONFIG_KEYS[key]
    if choices is not None:
        return " or ".join(f"'{v}'" for v in choices)
    if key in LIST_KEYS:
        return "a list of one or more non-empty strings"
    return "a non-empty string"

def get_config_path() -> Path:
    """Return the path to the configuration file.

    The file is ``config.json`` inside the directory given by `typer.get_app_dir`, so
    its location follows each platform's convention for application data. Nothing is
    created on disk: the file and its directory may not exist until `save_config`
    writes them.
    """
    return Path(typer.get_app_dir(APP_NAME)) / "config.json"

def read_config_file() -> dict[str, Any]:
    """Read the user configuration file exactly as it is stored.

    Nothing is validated, so the mapping may hold keys and values that recursivist does
    not recognize. It is what ``recursivist config set`` and ``config unset`` edit, so
    that changing one setting leaves the rest of the file untouched. Use `load_config`
    for settings that are safe to act on.

    A file that cannot be read, is not valid JSON, or does not hold a JSON object is
    reported with a warning and treated as empty. Reading never writes: a missing
    configuration directory is left missing.

    Returns:
        The stored mapping, or an empty one when the file is missing or unusable.
    """
    config_path = get_config_path()
    if not os.path.isfile(config_path):
        return {}
    try:
        with open(config_path, encoding="utf-8") as f:
            config = json.load(f)
    except (OSError, ValueError) as e:
        logger.warning("Ignoring user configuration %s: %s", config_path, e)
        return {}
    if not isinstance(config, dict):
        logger.warning(
            "Ignoring user configuration %s: expected a JSON object", config_path
        )
        return {}
    return config

def load_config() -> dict[str, Any]:
    """Load the valid settings of the user configuration file.

    The file is validated as it is loaded, so a hand-edited mistake never reaches the
    rest of the program: an unknown key or an unacceptable value is reported with a
    warning and left out. The warning for an unknown key gives the ``recursivist config
    unset`` command that removes it. Keys may be written with dashes or underscores.
    Reading never writes, so the file itself is left as it is.

    Returns:
        The valid settings, keyed in their underscored form. Settings the file does not
        set, or sets wrongly, are absent; the mapping is empty when the file is missing
        or unusable. Use `resolve_config` for a mapping with every key present.
    """
    return _validate_settings(read_config_file(), get_config_path(), unset_hint=True)

def save_config(config: dict[str, Any]) -> None:
    """Write the user configuration to disk as indented JSON.

    Creates the configuration directory if it does not already exist.

    Args:
        config: Configuration mapping to persist. Overwrites any existing file at the
            configuration path.
    """
    config_path = get_config_path()
    config_path.parent.mkdir(parents=True, exist_ok=True)
    with open(config_path, "w") as f:
        json.dump(config, f, indent=4)

def delete_config_file() -> bool:
    """Delete the user configuration file.

    The file is removed whatever it holds, including one that cannot be read as
    configuration, so every setting falls back to the layers below. The directory that
    holds the file is left in place, and nothing is created.

    Returns:
        ``True`` if the file is removed, ``False`` if there is no file to remove.

    Raises:
        OSError: If the file exists but cannot be removed.
    """
    config_path = get_config_path()
    if not os.path.isfile(config_path):
        return False
    os.remove(config_path)
    return True

def resolve_config_layers(
    project_dir: Path | None = None,
) -> dict[str, list[ConfigLayer]]:
    """Return what every configuration layer holds for each setting.

    The layers of a setting are listed from the highest precedence to the lowest: the
    project configuration that applies to *project_dir*, the user configuration file,
    then the built-in default. The first layer that has a value is the one in effect;
    a setting whose built-in default is ``None`` has no value when neither file sets
    it. Both files are validated as they are loaded, so every value is one its key
    accepts, and a layer whose file sets a key wrongly counts as not setting it. Reading
    never writes.

    Args:
        project_dir: Directory whose project configuration should apply, normally the
            one being scanned. ``None`` leaves the project layer out altogether.

    Returns:
        A mapping from every key in `CONFIG_KEYS` to its layers, in precedence order.
    """
    user_path = get_config_path()
    user_settings = load_config()
    layers: list[tuple[str, dict[str, Any], Path | None]] = [
        (LAYER_USER, user_settings, user_path if os.path.isfile(user_path) else None),
        (LAYER_DEFAULT, DEFAULT_CONFIG, None),
    ]
    if project_dir is not None:
        source, settings = _find_project_config(project_dir) or (None, {})
        layers.insert(0, (LAYER_PROJECT, settings, source))
    return {
        key: [
            ConfigLayer(name, settings.get(key), source)
            for name, settings, source in layers
        ]
        for key in CONFIG_KEYS
    }

def resolve_config(project_dir: Path | None = None) -> dict[str, Any]:
    """Return the effective configuration for a run.

    Layers are applied in order, each overriding the previous one: the built-in
    defaults, the user configuration file, then the project configuration that applies
    to *project_dir*. Both files are validated as they are loaded, so every value in the
    result is one its key accepts, or ``None`` for a setting that no layer gives a
    value.

    Args:
        project_dir: Directory whose project configuration should apply, normally the
            one being scanned. ``None`` skips the project layer, leaving the user
            configuration over the defaults.

    Returns:
        The merged configuration mapping, with every key in `CONFIG_KEYS` present.
    """
    return {
        key: next((layer.value for layer in layers if layer.value is not None), None)
        for key, layers in resolve_config_layers(project_dir).items()
    }