refactor: rename clone to clone_repo and consolidate schema & utility modules (#237)

* refactor: rename clone to clone_repo and consolidate schema & utility modules

Author: filipchristiansen

src/gitingest/__init__.py

@@ -1,8 +1,8 @@
 """Gitingest: A package for ingesting data from Git repositories."""
 
-from gitingest.cloning import clone
+from gitingest.cloning import clone_repo
 from gitingest.entrypoint import ingest, ingest_async
 from gitingest.ingestion import ingest_query
 from gitingest.query_parsing import parse_query
 
-__all__ = ["ingest_query", "clone", "parse_query", "ingest", "ingest_async"]
+__all__ = ["ingest_query", "clone_repo", "parse_query", "ingest", "ingest_async"]

src/gitingest/cloning.py

@@ -1,18 +1,18 @@
 """This module contains functions for cloning a Git repository to a local path."""
 
-import asyncio
 import os
 from pathlib import Path
-from typing import List, Optional, Tuple
+from typing import Optional
 
-from gitingest.ingestion_schema import CloneConfig
+from gitingest.schemas import CloneConfig
+from gitingest.utils.git_utils import check_repo_exists, ensure_git_installed, run_command
 from gitingest.utils.timeout_wrapper import async_timeout
 
 TIMEOUT: int = 60
 
 
 @async_timeout(TIMEOUT)
-async def clone(config: CloneConfig) -> None:
+async def clone_repo(config: CloneConfig) -> None:
     """
     Clone a repository to a local path based on the provided configuration.
 
@@ -47,7 +47,7 @@ async def clone(config: CloneConfig) -> None:
         raise OSError(f"Failed to create parent directory {parent_dir}: {exc}") from exc
 
     # Check if the repository exists
-    if not await _check_repo_exists(url):
+    if not await check_repo_exists(url):
         raise ValueError("Repository not found, make sure it is public")
 
     clone_cmd = ["git", "clone", "--single-branch"]
@@ -64,7 +64,8 @@ async def clone(config: CloneConfig) -> None:
     clone_cmd += [url, local_path]
 
     # Clone the repository
-    await _run_command(*clone_cmd)
+    await ensure_git_installed()
+    await run_command(*clone_cmd)
 
     if commit or partial_clone:
         checkout_cmd = ["git", "-C", local_path]
@@ -81,148 +82,4 @@ async def clone(config: CloneConfig) -> None:
             checkout_cmd += ["checkout", commit]
 
         # Check out the specific commit and/or subpath
-        await _run_command(*checkout_cmd)
-
-
-async def _check_repo_exists(url: str) -> bool:
-    """
-    Check if a Git repository exists at the provided URL.
-
-    Parameters
-    ----------
-    url : str
-        The URL of the Git repository to check.
-    Returns
-    -------
-    bool
-        True if the repository exists, False otherwise.
-
-    Raises
-    ------
-    RuntimeError
-        If the curl command returns an unexpected status code.
-    """
-    proc = await asyncio.create_subprocess_exec(
-        "curl",
-        "-I",
-        url,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-    )
-    stdout, _ = await proc.communicate()
-
-    if proc.returncode != 0:
-        return False
-
-    response = stdout.decode()
-    status_code = _get_status_code(response)
-
-    if status_code in (200, 301):
-        return True
-
-    if status_code in (404, 302):
-        return False
-
-    raise RuntimeError(f"Unexpected status code: {status_code}")
-
-
-async def fetch_remote_branch_list(url: str) -> List[str]:
-    """
-    Fetch the list of branches from a remote Git repository.
-    Parameters
-    ----------
-    url : str
-        The URL of the Git repository to fetch branches from.
-    Returns
-    -------
-    List[str]
-        A list of branch names available in the remote repository.
-    """
-    fetch_branches_command = ["git", "ls-remote", "--heads", url]
-    stdout, _ = await _run_command(*fetch_branches_command)
-    stdout_decoded = stdout.decode()
-
-    return [
-        line.split("refs/heads/", 1)[1]
-        for line in stdout_decoded.splitlines()
-        if line.strip() and "refs/heads/" in line
-    ]
-
-
-async def _run_command(*args: str) -> Tuple[bytes, bytes]:
-    """
-    Execute a command asynchronously and captures its output.
-
-    Parameters
-    ----------
-    *args : str
-        The command and its arguments to execute.
-
-    Returns
-    -------
-    Tuple[bytes, bytes]
-        A tuple containing the stdout and stderr of the command.
-
-    Raises
-    ------
-    RuntimeError
-        If command exits with a non-zero status.
-    """
-    await check_git_installed()
-
-    # Execute the requested command
-    proc = await asyncio.create_subprocess_exec(
-        *args,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-    )
-    stdout, stderr = await proc.communicate()
-    if proc.returncode != 0:
-        error_message = stderr.decode().strip()
-        raise RuntimeError(f"Command failed: {' '.join(args)}\nError: {error_message}")
-
-    return stdout, stderr
-
-
-async def check_git_installed() -> None:
-    """
-    Check if Git is installed and accessible on the system.
-
-    Raises
-    ------
-    RuntimeError
-        If Git is not installed or if the Git command exits with a non-zero status.
-    """
-    try:
-        proc = await asyncio.create_subprocess_exec(
-            "git",
-            "--version",
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        _, stderr = await proc.communicate()
-        if proc.returncode != 0:
-            error_message = stderr.decode().strip() if stderr else "Git command not found"
-            raise RuntimeError(f"Git is not installed or not accessible: {error_message}")
-
-    except FileNotFoundError as exc:
-        raise RuntimeError("Git is not installed. Please install Git before proceeding.") from exc
-
-
-def _get_status_code(response: str) -> int:
-    """
-    Extract the status code from an HTTP response.
-
-    Parameters
-    ----------
-    response : str
-        The HTTP response string.
-
-    Returns
-    -------
-    int
-        The status code of the response
-    """
-    status_line = response.splitlines()[0].strip()
-    status_code = int(status_line.split(" ", 2)[1])
-    return status_code
+        await run_command(*checkout_cmd)

src/gitingest/entrypoint.py

@@ -5,7 +5,7 @@
 import shutil
 from typing import Optional, Set, Tuple, Union
 
-from gitingest.cloning import clone
+from gitingest.cloning import clone_repo
 from gitingest.config import TMP_BASE_PATH
 from gitingest.ingestion import ingest_query
 from gitingest.query_parsing import IngestionQuery, parse_query
@@ -53,7 +53,7 @@ async def ingest_async(
     Raises
     ------
     TypeError
-        If `clone` does not return a coroutine, or if the `source` is of an unsupported type.
+        If `clone_repo` does not return a coroutine, or if the `source` is of an unsupported type.
     """
     repo_cloned = False
 
@@ -71,15 +71,15 @@ async def ingest_async(
             query.branch = selected_branch
 
             clone_config = query.extract_clone_config()
-            clone_coroutine = clone(clone_config)
+            clone_coroutine = clone_repo(clone_config)
 
             if inspect.iscoroutine(clone_coroutine):
                 if asyncio.get_event_loop().is_running():
                     await clone_coroutine
                 else:
                     asyncio.run(clone_coroutine)
             else:
-                raise TypeError("clone did not return a coroutine as expected.")
+                raise TypeError("clone_repo did not return a coroutine as expected.")
 
             repo_cloned = True
 

src/gitingest/ingestion.py

@@ -5,9 +5,9 @@
 from typing import Tuple
 
 from gitingest.config import MAX_DIRECTORY_DEPTH, MAX_FILES, MAX_TOTAL_SIZE_BYTES
-from gitingest.filesystem_schema import FileSystemNode, FileSystemNodeType, FileSystemStats
 from gitingest.output_formatters import format_node
 from gitingest.query_parsing import IngestionQuery
+from gitingest.schemas import FileSystemNode, FileSystemNodeType, FileSystemStats
 from gitingest.utils.ingestion_utils import _should_exclude, _should_include
 from gitingest.utils.path_utils import _is_safe_symlink
 

src/gitingest/output_formatters.py

@@ -4,8 +4,8 @@
 
 import tiktoken
 
-from gitingest.filesystem_schema import FileSystemNode, FileSystemNodeType
 from gitingest.query_parsing import IngestionQuery
+from gitingest.schemas import FileSystemNode, FileSystemNodeType
 
 
 def format_node(node: FileSystemNode, query: IngestionQuery) -> Tuple[str, str, str]:

src/gitingest/query_parsing.py

@@ -7,10 +7,10 @@
 from typing import List, Optional, Set, Union
 from urllib.parse import unquote, urlparse
 
-from gitingest.cloning import _check_repo_exists, fetch_remote_branch_list
 from gitingest.config import TMP_BASE_PATH
-from gitingest.exceptions import InvalidPatternError
-from gitingest.ingestion_schema import IngestionQuery
+from gitingest.schemas import IngestionQuery
+from gitingest.utils.exceptions import InvalidPatternError
+from gitingest.utils.git_utils import check_repo_exists, fetch_remote_branch_list
 from gitingest.utils.ignore_patterns import DEFAULT_IGNORE_PATTERNS
 from gitingest.utils.query_parser_utils import (
     KNOWN_GIT_HOSTS,
@@ -308,6 +308,6 @@ async def try_domains_for_user_and_repo(user_name: str, repo_name: str) -> str:
     """
     for domain in KNOWN_GIT_HOSTS:
         candidate = f"https://{domain}/{user_name}/{repo_name}"
-        if await _check_repo_exists(candidate):
+        if await check_repo_exists(candidate):
             return domain
     raise ValueError(f"Could not find a valid repository host for '{user_name}/{repo_name}'.")

src/gitingest/schemas/__init__.py

@@ -0,0 +1,6 @@
+"""This module contains the schemas for the Gitingest package."""
+
+from gitingest.schemas.filesystem_schema import FileSystemNode, FileSystemNodeType, FileSystemStats
+from gitingest.schemas.ingestion_schema import CloneConfig, IngestionQuery
+
+__all__ = ["FileSystemNode", "FileSystemNodeType", "FileSystemStats", "CloneConfig", "IngestionQuery"]

src/gitingest/filesystem_schema.py renamed to src/gitingest/schemas/filesystem_schema.py

@@ -7,9 +7,8 @@
 from enum import Enum, auto
 from pathlib import Path
 
-from gitingest.utils.ingestion_utils import _get_encoding_list
+from gitingest.utils.file_utils import get_preferred_encodings, is_text_file
 from gitingest.utils.notebook_utils import process_notebook
-from gitingest.utils.textfile_checker_utils import is_textfile
 
 SEPARATOR = "=" * 48  # Tiktoken, the tokenizer openai uses, counts 2 tokens if we have more than 48
 
@@ -117,7 +116,7 @@ def content(self) -> str:  # pylint: disable=too-many-return-statements
         if self.type == FileSystemNodeType.DIRECTORY:
             raise ValueError("Cannot read content of a directory node")
 
-        if not is_textfile(self.path):
+        if not is_text_file(self.path):
             return "[Non-text file]"
 
         if self.path.suffix == ".ipynb":
@@ -127,7 +126,7 @@ def content(self) -> str:  # pylint: disable=too-many-return-statements
                 return f"Error processing notebook: {exc}"
 
         # Try multiple encodings
-        for encoding in _get_encoding_list():
+        for encoding in get_preferred_encodings():
             try:
                 with self.path.open(encoding=encoding) as f:
                     return f.read()

src/gitingest/ingestion_schema.py renamed to src/gitingest/schemas/ingestion_schema.py

@@ -0,0 +1,72 @@
+"""Utility functions for working with files and directories."""
+
+import locale
+import platform
+from pathlib import Path
+from typing import List
+
+try:
+    locale.setlocale(locale.LC_ALL, "")
+except locale.Error:
+    locale.setlocale(locale.LC_ALL, "C")
+
+
+def get_preferred_encodings() -> List[str]:
+    """
+    Get list of encodings to try, prioritized for the current platform.
+
+    Returns
+    -------
+    List[str]
+        List of encoding names to try in priority order, starting with the
+        platform's default encoding followed by common fallback encodings.
+    """
+    encodings = [locale.getpreferredencoding(), "utf-8", "utf-16", "utf-16le", "utf-8-sig", "latin"]
+    if platform.system() == "Windows":
+        encodings += ["cp1252", "iso-8859-1"]
+    return encodings
+
+
+def is_text_file(path: Path) -> bool:
+    """
+    Determine if the file is likely a text file by trying to decode a small chunk
+    with multiple encodings, and checking for common binary markers.
+
+    Parameters
+    ----------
+    path : Path
+        The path to the file to check.
+
+    Returns
+    -------
+    bool
+        True if the file is likely textual; False if it appears to be binary.
+    """
+
+    # Attempt to read a portion of the file in binary mode
+    try:
+        with path.open("rb") as f:
+            chunk = f.read(1024)
+    except OSError:
+        return False
+
+    # If file is empty, treat as text
+    if not chunk:
+        return True
+
+    # Check obvious binary bytes
+    if b"\x00" in chunk or b"\xff" in chunk:
+        return False
+
+    # Attempt multiple encodings
+    for enc in get_preferred_encodings():
+        try:
+            with path.open(encoding=enc) as f:
+                f.read()
+                return True
+        except UnicodeDecodeError:
+            continue
+        except OSError:
+            return False
+
+    return False