Github

GitHubRepositoryCollaboratorsReader

Bases: BaseReader

GitHub存储库协作者阅读器。

检索GitHub存储库的协作者列表，并返回一个文档列表。

示例： >>> reader = GitHubRepositoryCollaboratorsReader("owner", "repo") >>> colabs = reader.load_data() >>> print(colabs)

Source code in llama_index/readers/github/collaborators/base.py

class GitHubRepositoryCollaboratorsReader(BaseReader):
    """GitHub存储库协作者阅读器。

检索GitHub存储库的协作者列表，并返回一个文档列表。

示例：
    >>> reader = GitHubRepositoryCollaboratorsReader("owner", "repo")
    >>> colabs = reader.load_data()
    >>> print(colabs)"""

    class FilterType(enum.Enum):
        """过滤器类型。

用于确定过滤器是包含还是排除。"""

        EXCLUDE = enum.auto()
        INCLUDE = enum.auto()

    def __init__(
        self,
        github_client: BaseGitHubCollaboratorsClient,
        owner: str,
        repo: str,
        verbose: bool = False,
    ):
        """初始化参数。

Args:
    - github_client (BaseGitHubCollaboratorsClient): GitHub客户端。
    - owner (str): 仓库所有者。
    - repo (str): 仓库名称。
    - verbose (bool): 是否打印详细信息。

抛出:
    - `ValueError`: 如果未提供github_token并且未设置GITHUB_TOKEN环境变量。
"""
        super().__init__()

        self._owner = owner
        self._repo = repo
        self._verbose = verbose

        # Set up the event loop
        try:
            self._loop = asyncio.get_running_loop()
        except RuntimeError:
            # If there is no running loop, create a new one
            self._loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._loop)

        self._github_client = github_client

    def load_data(
        self,
    ) -> List[Document]:
        """
        GitHub repository collaborators reader.

        Retrieves the list of collaborators in a GitHub repository and converts them to documents.

        Each collaborator is converted to a document by doing the following:

            - The text of the document is the login.
            - The title of the document is also the login.
            - The extra_info of the document is a dictionary with the following keys:
                - login: str, the login of the user
                - type: str, the type of user e.g. "User"
                - site_admin: bool, whether the user has admin permissions
                - role_name: str, e.g. "admin"
                - name: str, the name of the user, if available
                - email: str, the email of the user, if available
                - permissions: str, the permissions of the user, if available


        :return: list of documents
        """
        documents = []
        page = 1
        # Loop until there are no more collaborators
        while True:
            collaborators: Dict = self._loop.run_until_complete(
                self._github_client.get_collaborators(
                    self._owner, self._repo, page=page
                )
            )

            if len(collaborators) == 0:
                print_if_verbose(self._verbose, "No more collaborators found, stopping")

                break
            print_if_verbose(
                self._verbose,
                f"Found {len(collaborators)} collaborators in the repo page {page}",
            )
            page += 1
            for collab in collaborators:
                extra_info = {
                    "login": collab["login"],
                    "type": collab["type"],
                    "site_admin": collab["site_admin"],
                    "role_name": collab["role_name"],
                }
                if collab.get("name") is not None:
                    extra_info["name"] = collab["name"]
                if collab.get("email") is not None:
                    extra_info["email"] = collab["email"]
                if collab.get("permissions") is not None:
                    extra_info["permissions"] = collab["permissions"]
                document = Document(
                    doc_id=str(collab["login"]),
                    text=str(collab["login"]),  # unsure for this
                    extra_info=extra_info,
                )
                documents.append(document)

            print_if_verbose(self._verbose, f"Resulted in {len(documents)} documents")

        return documents

FilterType

Bases: Enum

过滤器类型。

用于确定过滤器是包含还是排除。

Source code in llama_index/readers/github/collaborators/base.py

    class FilterType(enum.Enum):
        """过滤器类型。

用于确定过滤器是包含还是排除。"""

        EXCLUDE = enum.auto()
        INCLUDE = enum.auto()

load_data

load_data() -> List[Document]

GitHub repository collaborators reader.

Retrieves the list of collaborators in a GitHub repository and converts them to documents.

Each collaborator is converted to a document by doing the following:

- The text of the document is the login.
- The title of the document is also the login.
- The extra_info of the document is a dictionary with the following keys:
    - login: str, the login of the user
    - type: str, the type of user e.g. "User"
    - site_admin: bool, whether the user has admin permissions
    - role_name: str, e.g. "admin"
    - name: str, the name of the user, if available
    - email: str, the email of the user, if available
    - permissions: str, the permissions of the user, if available

:return: list of documents

Source code in llama_index/readers/github/collaborators/base.py

def load_data(
    self,
) -> List[Document]:
    """
    GitHub repository collaborators reader.

    Retrieves the list of collaborators in a GitHub repository and converts them to documents.

    Each collaborator is converted to a document by doing the following:

        - The text of the document is the login.
        - The title of the document is also the login.
        - The extra_info of the document is a dictionary with the following keys:
            - login: str, the login of the user
            - type: str, the type of user e.g. "User"
            - site_admin: bool, whether the user has admin permissions
            - role_name: str, e.g. "admin"
            - name: str, the name of the user, if available
            - email: str, the email of the user, if available
            - permissions: str, the permissions of the user, if available


    :return: list of documents
    """
    documents = []
    page = 1
    # Loop until there are no more collaborators
    while True:
        collaborators: Dict = self._loop.run_until_complete(
            self._github_client.get_collaborators(
                self._owner, self._repo, page=page
            )
        )

        if len(collaborators) == 0:
            print_if_verbose(self._verbose, "No more collaborators found, stopping")

            break
        print_if_verbose(
            self._verbose,
            f"Found {len(collaborators)} collaborators in the repo page {page}",
        )
        page += 1
        for collab in collaborators:
            extra_info = {
                "login": collab["login"],
                "type": collab["type"],
                "site_admin": collab["site_admin"],
                "role_name": collab["role_name"],
            }
            if collab.get("name") is not None:
                extra_info["name"] = collab["name"]
            if collab.get("email") is not None:
                extra_info["email"] = collab["email"]
            if collab.get("permissions") is not None:
                extra_info["permissions"] = collab["permissions"]
            document = Document(
                doc_id=str(collab["login"]),
                text=str(collab["login"]),  # unsure for this
                extra_info=extra_info,
            )
            documents.append(document)

        print_if_verbose(self._verbose, f"Resulted in {len(documents)} documents")

    return documents

GitHubRepositoryIssuesReader

Bases: BaseReader

GitHub仓库问题阅读器。

获取GitHub仓库的问题列表，并返回一个文档列表。

示例

reader = GitHubRepositoryIssuesReader("owner", "repo") issues = reader.load_data() print(issues)

Source code in llama_index/readers/github/issues/base.py

class GitHubRepositoryIssuesReader(BaseReader):
    """GitHub仓库问题阅读器。

获取GitHub仓库的问题列表，并返回一个文档列表。

示例:
    >>> reader = GitHubRepositoryIssuesReader("owner", "repo")
    >>> issues = reader.load_data()
    >>> print(issues)"""

    class IssueState(enum.Enum):
        """问题类型。

用于决定要检索哪些问题。

属性:
- OPEN: 仅打开的问题。这是默认值。
- CLOSED: 仅关闭的问题。
- ALL: 所有问题，包括打开和关闭的。"""

        OPEN = "open"
        CLOSED = "closed"
        ALL = "all"

    class FilterType(enum.Enum):
        """过滤器类型。

用于确定过滤器是包含还是排除。"""

        EXCLUDE = enum.auto()
        INCLUDE = enum.auto()

    def __init__(
        self,
        github_client: BaseGitHubIssuesClient,
        owner: str,
        repo: str,
        verbose: bool = False,
    ):
        """初始化参数。

Args:
    - github_client（BaseGitHubIssuesClient）：GitHub客户端。
    - owner（str）：仓库所有者。
    - repo（str）：仓库名称。
    - verbose（bool）：是否打印详细信息。

引发：
    - `ValueError`：如果未提供github_token且未设置GITHUB_TOKEN环境变量。
"""
        super().__init__()

        self._owner = owner
        self._repo = repo
        self._verbose = verbose

        # Set up the event loop
        try:
            self._loop = asyncio.get_running_loop()
        except RuntimeError:
            # If there is no running loop, create a new one
            self._loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._loop)

        self._github_client = github_client

    def load_data(
        self,
        state: Optional[IssueState] = IssueState.OPEN,
        labelFilters: Optional[List[Tuple[str, FilterType]]] = None,
    ) -> List[Document]:
        """从存储库中加载问题并将其转换为文档。

通过以下方式将每个问题转换为文档：

- 文档的文本是问题的标题和正文的连接。
- 文档的标题是问题的标题。
- 文档的doc_id是问题编号。
- 文档的extra_info是一个包含以下键的字典：
    - state：问题的状态。
    - created_at：问题创建的日期。
    - closed_at：问题关闭的日期。仅在问题关闭时出现。
    - url：问题的URL。
    - assignee：分配给问题的用户的登录名。仅在问题已分配时出现。
- 文档的嵌入是None。
- 文档的doc_hash是None。

Args:
    - state（IssueState）：要检索的问题的状态。默认为IssueState.OPEN。
    - labelFilters：要应用于基于标签的问题列表的可选过滤器列表。

Returns:文档列表
"""
        documents = []
        page = 1
        # Loop until there are no more issues
        while True:
            issues: Dict = self._loop.run_until_complete(
                self._github_client.get_issues(
                    self._owner, self._repo, state=state.value, page=page
                )
            )

            if len(issues) == 0:
                print_if_verbose(self._verbose, "No more issues found, stopping")

                break
            print_if_verbose(
                self._verbose, f"Found {len(issues)} issues in the repo page {page}"
            )
            page += 1
            filterCount = 0
            for issue in issues:
                if not self._must_include(labelFilters, issue):
                    filterCount += 1
                    continue
                title = issue["title"]
                body = issue["body"]
                document = Document(
                    doc_id=str(issue["number"]),
                    text=f"{title}\n{body}",
                )
                extra_info = {
                    "state": issue["state"],
                    "created_at": issue["created_at"],
                    # url is the API URL
                    "url": issue["url"],
                    # source is the HTML URL, more convenient for humans
                    "source": issue["html_url"],
                }
                if issue["closed_at"] is not None:
                    extra_info["closed_at"] = issue["closed_at"]
                if issue["assignee"] is not None:
                    extra_info["assignee"] = issue["assignee"]["login"]
                if issue["labels"] is not None:
                    extra_info["labels"] = [label["name"] for label in issue["labels"]]
                document.extra_info = extra_info
                documents.append(document)

            print_if_verbose(self._verbose, f"Resulted in {len(documents)} documents")
            if labelFilters is not None:
                print_if_verbose(self._verbose, f"Filtered out {filterCount} issues")

        return documents

    def _must_include(self, labelFilters, issue):
        if labelFilters is None:
            return True
        labels = [label["name"] for label in issue["labels"]]
        for labelFilter in labelFilters:
            label = labelFilter[0]
            filterType = labelFilter[1]
            # Only include issues with the label and value
            if filterType == self.FilterType.INCLUDE:
                return label in labels
            elif filterType == self.FilterType.EXCLUDE:
                return label not in labels

        return True

IssueState

Bases: Enum

问题类型。

用于决定要检索哪些问题。

属性: - OPEN: 仅打开的问题。这是默认值。 - CLOSED: 仅关闭的问题。 - ALL: 所有问题，包括打开和关闭的。

Source code in llama_index/readers/github/issues/base.py

    class IssueState(enum.Enum):
        """问题类型。

用于决定要检索哪些问题。

属性:
- OPEN: 仅打开的问题。这是默认值。
- CLOSED: 仅关闭的问题。
- ALL: 所有问题，包括打开和关闭的。"""

        OPEN = "open"
        CLOSED = "closed"
        ALL = "all"

FilterType

Bases: Enum

过滤器类型。

用于确定过滤器是包含还是排除。

Source code in llama_index/readers/github/issues/base.py

    class FilterType(enum.Enum):
        """过滤器类型。

用于确定过滤器是包含还是排除。"""

        EXCLUDE = enum.auto()
        INCLUDE = enum.auto()

load_data

load_data(
    state: Optional[IssueState] = IssueState.OPEN,
    labelFilters: Optional[
        List[Tuple[str, FilterType]]
    ] = None,
) -> List[Document]

从存储库中加载问题并将其转换为文档。

通过以下方式将每个问题转换为文档：

• 文档的文本是问题的标题和正文的连接。
• 文档的标题是问题的标题。
• 文档的doc_id是问题编号。
• 文档的extra_info是一个包含以下键的字典：
  • state：问题的状态。
  • created_at：问题创建的日期。
  • closed_at：问题关闭的日期。仅在问题关闭时出现。
  • url：问题的URL。
  • assignee：分配给问题的用户的登录名。仅在问题已分配时出现。
• 文档的嵌入是None。
• 文档的doc_hash是None。

Returns:文档列表

Source code in llama_index/readers/github/issues/base.py

    def load_data(
        self,
        state: Optional[IssueState] = IssueState.OPEN,
        labelFilters: Optional[List[Tuple[str, FilterType]]] = None,
    ) -> List[Document]:
        """从存储库中加载问题并将其转换为文档。

通过以下方式将每个问题转换为文档：

- 文档的文本是问题的标题和正文的连接。
- 文档的标题是问题的标题。
- 文档的doc_id是问题编号。
- 文档的extra_info是一个包含以下键的字典：
    - state：问题的状态。
    - created_at：问题创建的日期。
    - closed_at：问题关闭的日期。仅在问题关闭时出现。
    - url：问题的URL。
    - assignee：分配给问题的用户的登录名。仅在问题已分配时出现。
- 文档的嵌入是None。
- 文档的doc_hash是None。

Args:
    - state（IssueState）：要检索的问题的状态。默认为IssueState.OPEN。
    - labelFilters：要应用于基于标签的问题列表的可选过滤器列表。

Returns:文档列表
"""
        documents = []
        page = 1
        # Loop until there are no more issues
        while True:
            issues: Dict = self._loop.run_until_complete(
                self._github_client.get_issues(
                    self._owner, self._repo, state=state.value, page=page
                )
            )

            if len(issues) == 0:
                print_if_verbose(self._verbose, "No more issues found, stopping")

                break
            print_if_verbose(
                self._verbose, f"Found {len(issues)} issues in the repo page {page}"
            )
            page += 1
            filterCount = 0
            for issue in issues:
                if not self._must_include(labelFilters, issue):
                    filterCount += 1
                    continue
                title = issue["title"]
                body = issue["body"]
                document = Document(
                    doc_id=str(issue["number"]),
                    text=f"{title}\n{body}",
                )
                extra_info = {
                    "state": issue["state"],
                    "created_at": issue["created_at"],
                    # url is the API URL
                    "url": issue["url"],
                    # source is the HTML URL, more convenient for humans
                    "source": issue["html_url"],
                }
                if issue["closed_at"] is not None:
                    extra_info["closed_at"] = issue["closed_at"]
                if issue["assignee"] is not None:
                    extra_info["assignee"] = issue["assignee"]["login"]
                if issue["labels"] is not None:
                    extra_info["labels"] = [label["name"] for label in issue["labels"]]
                document.extra_info = extra_info
                documents.append(document)

            print_if_verbose(self._verbose, f"Resulted in {len(documents)} documents")
            if labelFilters is not None:
                print_if_verbose(self._verbose, f"Filtered out {filterCount} issues")

        return documents

GithubRepositoryReader

Bases: BaseReader

    # Github仓库阅读器。

    # 获取Github仓库的内容，并返回文档列表。
    # 这些文档可以是仓库中文件的内容，也可以是使用解析器从文件中提取的文本。

    # 示例:
    >>> client = github_client = GithubClient(
    ...    github_token=os.environ["GITHUB_TOKEN"],
    ...    verbose=True
    ... )
    >>> reader = GithubRepositoryReader(
    ...    github_client=github_client,
    ...    owner="run-llama",
    ...    repo="llama_index",
    ... )
    >>> branch_documents = reader.load_data(branch="branch")
    >>> commit_documents = reader.load_data(commit_sha="commit_sha")

Source code in llama_index/readers/github/repository/base.py

class GithubRepositoryReader(BaseReader):
    """```python
    # Github仓库阅读器。

    # 获取Github仓库的内容，并返回文档列表。
    # 这些文档可以是仓库中文件的内容，也可以是使用解析器从文件中提取的文本。

    # 示例:
    >>> client = github_client = GithubClient(
    ...    github_token=os.environ["GITHUB_TOKEN"],
    ...    verbose=True
    ... )
    >>> reader = GithubRepositoryReader(
    ...    github_client=github_client,
    ...    owner="run-llama",
    ...    repo="llama_index",
    ... )
    >>> branch_documents = reader.load_data(branch="branch")
    >>> commit_documents = reader.load_data(commit_sha="commit_sha")
```"""

    class FilterType(enum.Enum):
        """        过滤器类型。

        用于确定过滤器是包含还是排除的。

        属性：
            - EXCLUDE: 排除目录中的文件或具有特定扩展名的文件。
            - INCLUDE: 仅包含目录中的文件或具有特定扩展名的文件。"""

        EXCLUDE = enum.auto()
        INCLUDE = enum.auto()

    def __init__(
        self,
        github_client: BaseGithubClient,
        owner: str,
        repo: str,
        use_parser: bool = False,
        verbose: bool = False,
        concurrent_requests: int = 5,
        timeout: Optional[int] = 5,
        retries: int = 0,
        filter_directories: Optional[Tuple[List[str], FilterType]] = None,
        filter_file_extensions: Optional[Tuple[List[str], FilterType]] = None,
    ):
        """初始化参数。

Args:
    - github_client（BaseGithubClient）：Github客户端。
    - owner（str）：仓库所有者。
    - repo（str）：仓库名称。
    - use_parser（bool）：是否使用解析器来提取文件中的文本。
    - verbose（bool）：是否打印详细信息。
    - concurrent_requests（int）：向Github API发出的并发请求数量。
    - timeout（int或None）：对Github API的请求超时时间。默认为5。
    - retries（int）：向Github API发出的请求重试次数。默认为0。
    - filter_directories（Optional[Tuple[List[str], FilterType]]）：包含目录列表和FilterType的元组。如果FilterType为INCLUDE，则只包括列表中目录中的文件。如果FilterType为EXCLUDE，则排除列表中目录中的文件。
    - filter_file_extensions（Optional[Tuple[List[str], FilterType]]）：包含文件扩展名列表和FilterType的元组。如果FilterType为INCLUDE，则只包括列表中扩展名的文件。如果FilterType为EXCLUDE，则排除列表中扩展名的文件。

Raises:
    - `ValueError`：如果未提供github_token并且未设置GITHUB_TOKEN环境变量。
"""
        super().__init__()

        self._owner = owner
        self._repo = repo
        self._use_parser = use_parser
        self._verbose = verbose
        self._concurrent_requests = concurrent_requests
        self._timeout = timeout
        self._retries = retries
        self._filter_directories = filter_directories
        self._filter_file_extensions = filter_file_extensions

        # Set up the event loop
        try:
            self._loop = asyncio.get_running_loop()
        except RuntimeError:
            # If there is no running loop, create a new one
            self._loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._loop)

        self._github_client = github_client

        self._file_readers: Dict[str, BaseReader] = {}
        self._supported_suffix = list(DEFAULT_FILE_READER_CLS.keys())

    def _check_filter_directories(self, tree_obj_path: str) -> bool:
        """检查根据目录是否应允许树对象。

:param `tree_obj_path`: 树对象的路径，例如 'llama_index/readers'

:return: 如果应允许树对象，则为True，否则为False
"""
        if self._filter_directories is None:
            return True
        filter_directories, filter_type = self._filter_directories
        print_if_verbose(
            self._verbose,
            f"Checking {tree_obj_path} whether to {filter_type} it"
            + f" based on the filter directories: {filter_directories}",
        )

        if filter_type == self.FilterType.EXCLUDE:
            print_if_verbose(
                self._verbose,
                f"Checking if {tree_obj_path} is not a subdirectory of any of the"
                " filter directories",
            )
            return not any(
                tree_obj_path.startswith(directory) for directory in filter_directories
            )
        if filter_type == self.FilterType.INCLUDE:
            print_if_verbose(
                self._verbose,
                f"Checking if {tree_obj_path} is a subdirectory of any of the filter"
                " directories",
            )
            return any(
                tree_obj_path.startswith(directory)
                or directory.startswith(tree_obj_path)
                for directory in filter_directories
            )
        raise ValueError(
            f"Unknown filter type: {filter_type}. "
            "Please use either 'INCLUDE' or 'EXCLUDE'."
        )

    def _check_filter_file_extensions(self, tree_obj_path: str) -> bool:
        """检查树对象是否应该根据文件扩展名允许。

:param `tree_obj_path`: 树对象的路径，例如'llama_index/indices'

:return: 如果应该允许树对象，则返回True，否则返回False
"""
        if self._filter_file_extensions is None:
            return True
        filter_file_extensions, filter_type = self._filter_file_extensions
        print_if_verbose(
            self._verbose,
            f"Checking {tree_obj_path} whether to {filter_type} it"
            + f" based on the filter file extensions: {filter_file_extensions}",
        )

        if filter_type == self.FilterType.EXCLUDE:
            return get_file_extension(tree_obj_path) not in filter_file_extensions
        if filter_type == self.FilterType.INCLUDE:
            return get_file_extension(tree_obj_path) in filter_file_extensions
        raise ValueError(
            f"Unknown filter type: {filter_type}. "
            "Please use either 'INCLUDE' or 'EXCLUDE'."
        )

    def _allow_tree_obj(self, tree_obj_path: str, tree_obj_type: str) -> bool:
        """检查树对象是否应该被允许。

:param `tree_obj_path`: 树对象的路径

:return: 如果树对象应该被允许，则返回True，否则返回False
"""
        if self._filter_directories is not None and tree_obj_type == "tree":
            return self._check_filter_directories(tree_obj_path)

        if self._filter_file_extensions is not None and tree_obj_type == "blob":
            return self._check_filter_directories(
                tree_obj_path
            ) and self._check_filter_file_extensions(tree_obj_path)

        return True

    def _load_data_from_commit(self, commit_sha: str) -> List[Document]:
        """从提交或分支加载数据。

从特定的提交sha或分支名称加载github存储库数据。

:param `commit`: 提交sha
:param `branch`: 分支名称

:return: 文档列表
"""
        commit_response: GitCommitResponseModel = self._loop.run_until_complete(
            self._github_client.get_commit(
                self._owner, self._repo, commit_sha, timeout=self._timeout
            )
        )

        tree_sha = commit_response.commit.tree.sha
        blobs_and_paths = self._loop.run_until_complete(self._recurse_tree(tree_sha))

        print_if_verbose(self._verbose, f"got {len(blobs_and_paths)} blobs")

        return self._loop.run_until_complete(
            self._generate_documents(blobs_and_paths=blobs_and_paths, id=commit_sha)
        )

    def _load_data_from_branch(self, branch: str) -> List[Document]:
        """从一个分支加载数据。

从特定分支加载github仓库的数据。

:param `branch`: 分支名称

:return: 文档列表
"""
        branch_data: GitBranchResponseModel = self._loop.run_until_complete(
            self._github_client.get_branch(
                self._owner, self._repo, branch, timeout=self._timeout
            )
        )

        tree_sha = branch_data.commit.commit.tree.sha
        blobs_and_paths = self._loop.run_until_complete(self._recurse_tree(tree_sha))

        print_if_verbose(self._verbose, f"got {len(blobs_and_paths)} blobs")

        return self._loop.run_until_complete(
            self._generate_documents(blobs_and_paths=blobs_and_paths, id=branch)
        )

    def load_data(
        self,
        commit_sha: Optional[str] = None,
        branch: Optional[str] = None,
    ) -> List[Document]:
        """
        Load data from a commit or a branch.

        Loads github repository data from a specific commit sha or a branch.

        :param `commit`: commit sha
        :param `branch`: branch name

        :return: list of documents
        """
        if commit_sha is not None and branch is not None:
            raise ValueError("You can only specify one of commit or branch.")

        if commit_sha is None and branch is None:
            raise ValueError("You must specify one of commit or branch.")

        if commit_sha is not None:
            return self._load_data_from_commit(commit_sha)

        if branch is not None:
            return self._load_data_from_branch(branch)

        raise ValueError("You must specify one of commit or branch.")

    async def _recurse_tree(
        self,
        tree_sha: str,
        current_path: str = "",
        current_depth: int = 0,
        max_depth: int = -1,
    ) -> Any:
        """递归获取树中的所有blob树对象。

并构建它们相对于存储库根目录的完整路径。
（有关更多信息，请参见github_api_client.py中的GitTreeResponseModel.GitTreeObject）

:param `tree_sha`: 要递归的树的sha
:param `current_path`: 树的当前路径
:param `current_depth`: 树的当前深度
:return: 元组列表，包括
    （树对象，文件相对于存储库根目录的完整路径）
"""
        if max_depth != -1 and current_depth > max_depth:
            return []

        blobs_and_full_paths: List[Tuple[GitTreeResponseModel.GitTreeObject, str]] = []
        print_if_verbose(
            self._verbose,
            "\t" * current_depth + f"current path: {current_path}",
        )

        tree_data: GitTreeResponseModel = await self._github_client.get_tree(
            self._owner, self._repo, tree_sha, timeout=self._timeout
        )
        print_if_verbose(
            self._verbose, "\t" * current_depth + f"tree data: {tree_data}"
        )
        print_if_verbose(
            self._verbose, "\t" * current_depth + f"processing tree {tree_sha}"
        )
        for tree_obj in tree_data.tree:
            file_path = os.path.join(current_path, tree_obj.path)
            if not self._allow_tree_obj(file_path, tree_obj.type):
                print_if_verbose(
                    self._verbose,
                    "\t" * current_depth + f"ignoring {tree_obj.path} due to filter",
                )
                continue

            print_if_verbose(
                self._verbose,
                "\t" * current_depth + f"tree object: {tree_obj}",
            )

            if tree_obj.type == "tree":
                print_if_verbose(
                    self._verbose,
                    "\t" * current_depth + f"recursing into {tree_obj.path}",
                )

                blobs_and_full_paths.extend(
                    await self._recurse_tree(
                        tree_obj.sha, file_path, current_depth + 1, max_depth
                    )
                )
            elif tree_obj.type == "blob":
                print_if_verbose(
                    self._verbose,
                    "\t" * current_depth + f"found blob {tree_obj.path}",
                )

                blobs_and_full_paths.append((tree_obj, file_path))

            print_if_verbose(
                self._verbose,
                "\t" * current_depth + f"blob and full paths: {blobs_and_full_paths}",
            )
        return blobs_and_full_paths

    async def _generate_documents(
        self,
        blobs_and_paths: List[Tuple[GitTreeResponseModel.GitTreeObject, str]],
        id: str = "",
    ) -> List[Document]:
        """从一组blob及其完整路径中生成文档。

:param `blobs_and_paths`: 元组列表，格式为
    (树对象, 相对于存储库根目录的文件完整路径)
:param `id`: 在加载存储库时使用的分支名称或提交sha
:return: 文档列表
"""
        buffered_iterator = BufferedGitBlobDataIterator(
            blobs_and_paths=blobs_and_paths,
            github_client=self._github_client,
            owner=self._owner,
            repo=self._repo,
            loop=self._loop,
            buffer_size=self._concurrent_requests,  # TODO: make this configurable
            verbose=self._verbose,
        )

        documents = []
        async for blob_data, full_path in buffered_iterator:
            print_if_verbose(self._verbose, f"generating document for {full_path}")
            assert (
                blob_data.encoding == "base64"
            ), f"blob encoding {blob_data.encoding} not supported"
            decoded_bytes = None
            try:
                decoded_bytes = base64.b64decode(blob_data.content)
                del blob_data.content
            except binascii.Error:
                print_if_verbose(
                    self._verbose, f"could not decode {full_path} as base64"
                )
                continue

            if self._use_parser:
                document = self._parse_supported_file(
                    file_path=full_path,
                    file_content=decoded_bytes,
                    tree_sha=blob_data.sha,
                    tree_path=full_path,
                )
                if document is not None:
                    documents.append(document)
                    continue
                print_if_verbose(
                    self._verbose,
                    f"could not parse {full_path} as a supported file type"
                    + " - falling back to decoding as utf-8 raw text",
                )

            try:
                if decoded_bytes is None:
                    raise ValueError("decoded_bytes is None")
                decoded_text = decoded_bytes.decode("utf-8")
            except UnicodeDecodeError:
                print_if_verbose(
                    self._verbose, f"could not decode {full_path} as utf-8"
                )
                continue
            print_if_verbose(
                self._verbose,
                f"got {len(decoded_text)} characters"
                + f"- adding to documents - {full_path}",
            )
            url = os.path.join(
                "https://github.com/", self._owner, self._repo, "blob/", id, full_path
            )
            document = Document(
                text=decoded_text,
                doc_id=blob_data.sha,
                extra_info={
                    "file_path": full_path,
                    "file_name": full_path.split("/")[-1],
                    "url": url,
                },
            )
            documents.append(document)
        return documents

    def _parse_supported_file(
        self,
        file_path: str,
        file_content: bytes,
        tree_sha: str,
        tree_path: str,
    ) -> Optional[Document]:
        """解析文件，如果被解析器支持。

:param `file_path`: 仓库中文件的路径
:param `file_content`: 文件的内容
:return: 如果文件被解析器支持，则返回文档，否则返回None
"""
        file_extension = get_file_extension(file_path)
        if file_extension not in self._supported_suffix:
            # skip
            return None

        if file_extension not in self._file_readers:
            # initialize reader
            cls_ = DEFAULT_FILE_READER_CLS[file_extension]
            self._file_readers[file_extension] = cls_()

        reader = self._file_readers[file_extension]

        print_if_verbose(
            self._verbose,
            f"parsing {file_path}"
            + f"as {file_extension} with "
            + f"{reader.__class__.__name__}",
        )
        with tempfile.TemporaryDirectory() as tmpdirname:
            with tempfile.NamedTemporaryFile(
                dir=tmpdirname,
                suffix=f".{file_extension}",
                mode="w+b",
                delete=False,
            ) as tmpfile:
                print_if_verbose(
                    self._verbose,
                    "created a temporary file"
                    + f"{tmpfile.name} for parsing {file_path}",
                )
                tmpfile.write(file_content)
                tmpfile.flush()
                tmpfile.close()
                try:
                    docs = reader.load_data(pathlib.Path(tmpfile.name))
                    parsed_file = "\n\n".join([doc.get_text() for doc in docs])
                except Exception as e:
                    print_if_verbose(self._verbose, f"error while parsing {file_path}")
                    logger.error(
                        "Error while parsing "
                        + f"{file_path} with "
                        + f"{reader.__class__.__name__}:\n{e}"
                    )
                    parsed_file = None
                finally:
                    os.remove(tmpfile.name)
                if parsed_file is None:
                    return None
                return Document(
                    text=parsed_file,
                    doc_id=tree_sha,
                    extra_info={
                        "file_path": file_path,
                        "file_name": tree_path,
                    },
                )

FilterType

Bases: Enum

过滤器类型。

用于确定过滤器是包含还是排除的。

属性： - EXCLUDE: 排除目录中的文件或具有特定扩展名的文件。 - INCLUDE: 仅包含目录中的文件或具有特定扩展名的文件。

Source code in llama_index/readers/github/repository/base.py

class FilterType(enum.Enum):
    """        过滤器类型。

    用于确定过滤器是包含还是排除的。

    属性：
        - EXCLUDE: 排除目录中的文件或具有特定扩展名的文件。
        - INCLUDE: 仅包含目录中的文件或具有特定扩展名的文件。"""

    EXCLUDE = enum.auto()
    INCLUDE = enum.auto()

load_data

load_data(
    commit_sha: Optional[str] = None,
    branch: Optional[str] = None,
) -> List[Document]

Load data from a commit or a branch.

Loads github repository data from a specific commit sha or a branch.

:param commit: commit sha :param branch: branch name

:return: list of documents

Source code in llama_index/readers/github/repository/base.py

def load_data(
    self,
    commit_sha: Optional[str] = None,
    branch: Optional[str] = None,
) -> List[Document]:
    """
    Load data from a commit or a branch.

    Loads github repository data from a specific commit sha or a branch.

    :param `commit`: commit sha
    :param `branch`: branch name

    :return: list of documents
    """
    if commit_sha is not None and branch is not None:
        raise ValueError("You can only specify one of commit or branch.")

    if commit_sha is None and branch is None:
        raise ValueError("You must specify one of commit or branch.")

    if commit_sha is not None:
        return self._load_data_from_commit(commit_sha)

    if branch is not None:
        return self._load_data_from_branch(branch)

    raise ValueError("You must specify one of commit or branch.")