codekingpro/portable-devtools

1import os2from typing import Callable, Iterator, Optional3 4from langchain_core.documents import Document5 6from langchain_community.document_loaders.base import BaseLoader7 8 9class GitLoader(BaseLoader):10    """Load `Git` repository files.11 12    The Repository can be local on disk available at `repo_path`,13    or remote at `clone_url` that will be cloned to `repo_path`.14    Currently, supports only text files.15 16    Each document represents one file in the repository. The `path` points to17    the local Git repository, and the `branch` specifies the branch to load18    files from. By default, it loads from the `main` branch.19    """20 21    def __init__(22        self,23        repo_path: str,24        clone_url: Optional[str] = None,25        branch: Optional[str] = "main",26        file_filter: Optional[Callable[[str], bool]] = None,27    ):28        """29 30        Args:31            repo_path: The path to the Git repository.32            clone_url: Optional. The URL to clone the repository from.33            branch: Optional. The branch to load files from. Defaults to `main`.34            file_filter: Optional. A function that takes a file path and returns35              a boolean indicating whether to load the file. Defaults to None.36        """37        self.repo_path = repo_path38        self.clone_url = clone_url39        self.branch = branch40        self.file_filter = file_filter41 42    def lazy_load(self) -> Iterator[Document]:43        try:44            from git import Blob, Repo45        except ImportError as ex:46            raise ImportError(47                "Could not import git python package. "48                "Please install it with `pip install GitPython`."49            ) from ex50 51        if not os.path.exists(self.repo_path) and self.clone_url is None:52            raise ValueError(f"Path {self.repo_path} does not exist")53        elif self.clone_url:54            # If the repo_path already contains a git repository, verify that it's the55            # same repository as the one we're trying to clone.56            if os.path.isdir(os.path.join(self.repo_path, ".git")):57                repo = Repo(self.repo_path)58                # If the existing repository is not the same as the one we're trying to59                # clone, raise an error.60                if repo.remotes.origin.url != self.clone_url:61                    raise ValueError(62                        "A different repository is already cloned at this path."63                    )64            else:65                repo = Repo.clone_from(self.clone_url, self.repo_path)66            repo.git.checkout(self.branch)67        else:68            repo = Repo(self.repo_path)69            repo.git.checkout(self.branch)70 71        for item in repo.tree().traverse():72            if not isinstance(item, Blob):73                continue74 75            file_path = os.path.join(self.repo_path, item.path)76 77            ignored_files = repo.ignored([file_path])78            if len(ignored_files):79                continue80 81            # uses filter to skip files82            if self.file_filter and not self.file_filter(file_path):83                continue84 85            rel_file_path = os.path.relpath(file_path, self.repo_path)86            try:87                with open(file_path, "rb") as f:88                    content = f.read()89                    file_type = os.path.splitext(item.name)[1]90 91                    # loads only text files92                    try:93                        text_content = content.decode("utf-8")94                    except UnicodeDecodeError:95                        continue96 97                    metadata = {98                        "source": rel_file_path,99                        "file_path": rel_file_path,100                        "file_name": item.name,101                        "file_type": file_type,102                    }103                    yield Document(page_content=text_content, metadata=metadata)104            except Exception as e:105                print(f"Error reading file {file_path}: {e}")  # noqa: T201106