| import re |
| from pathlib import Path |
|
|
| from langchain_community.document_loaders.git import GitLoader |
|
|
| from langflow.custom import Component |
| from langflow.io import MessageTextInput, Output |
| from langflow.schema import Data |
|
|
|
|
| class GitLoaderComponent(Component): |
| display_name = "GitLoader" |
| description = "Load files from a Git repository" |
| documentation = "https://python.langchain.com/v0.2/docs/integrations/document_loaders/git/" |
| trace_type = "tool" |
| icon = "GitLoader" |
| name = "GitLoader" |
|
|
| inputs = [ |
| MessageTextInput( |
| name="repo_path", |
| display_name="Repository Path", |
| required=False, |
| info="The local path to the Git repository.", |
| ), |
| MessageTextInput( |
| name="clone_url", |
| display_name="Clone URL", |
| required=False, |
| info="The URL to clone the Git repository from.", |
| ), |
| MessageTextInput( |
| name="branch", |
| display_name="Branch", |
| required=False, |
| value="main", |
| info="The branch to load files from. Defaults to 'main'.", |
| ), |
| MessageTextInput( |
| name="file_filter", |
| display_name="File Filter", |
| required=False, |
| advanced=True, |
| info="A list of patterns to filter files. Example to include only .py files: '*.py'. " |
| "Example to exclude .py files: '!*.py'. Multiple patterns can be separated by commas.", |
| ), |
| MessageTextInput( |
| name="content_filter", |
| display_name="Content Filter", |
| required=False, |
| advanced=True, |
| info="A regex pattern to filter files based on their content.", |
| ), |
| ] |
|
|
| outputs = [ |
| Output(name="data", display_name="Data", method="load_documents"), |
| ] |
|
|
| @staticmethod |
| def is_binary(file_path: str) -> bool: |
| """Check if a file is binary by looking for null bytes. |
| |
| This is necessary because when searches are performed using |
| the content_filter, binary files need to be ignored. |
| """ |
| with Path(file_path).open("rb") as file: |
| return b"\x00" in file.read(1024) |
|
|
| def build_gitloader(self) -> GitLoader: |
| file_filter_patterns = getattr(self, "file_filter", None) |
| content_filter_pattern = getattr(self, "content_filter", None) |
|
|
| file_filters = [] |
| if file_filter_patterns: |
| patterns = [pattern.strip() for pattern in file_filter_patterns.split(",")] |
|
|
| def file_filter(file_path: Path) -> bool: |
| if len(patterns) == 1 and patterns[0].startswith("!"): |
| return not file_path.match(patterns[0][1:]) |
| included = any(file_path.match(pattern) for pattern in patterns if not pattern.startswith("!")) |
| excluded = any(file_path.match(pattern[1:]) for pattern in patterns if pattern.startswith("!")) |
| return included and not excluded |
|
|
| file_filters.append(file_filter) |
|
|
| if content_filter_pattern: |
| content_regex = re.compile(content_filter_pattern) |
|
|
| def content_filter(file_path: Path) -> bool: |
| content = file_path.read_text(encoding="utf-8", errors="ignore") |
| return bool(content_regex.search(content)) |
|
|
| file_filters.append(content_filter) |
|
|
| def combined_filter(file_path: str) -> bool: |
| path = Path(file_path) |
| if self.is_binary(file_path): |
| return False |
| return all(f(path) for f in file_filters) |
|
|
| return GitLoader( |
| repo_path=self.repo_path, |
| clone_url=self.clone_url, |
| branch=self.branch, |
| file_filter=combined_filter, |
| ) |
|
|
| def load_documents(self) -> list[Data]: |
| gitloader = self.build_gitloader() |
| documents = list(gitloader.lazy_load()) |
| data = [Data.from_document(doc) for doc in documents] |
| self.status = data |
| return data |
|
|
