langchain_community.document_loaders.larksuite

import json
import urllib.request
from typing import Any, Iterator

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader


[docs]class LarkSuiteDocLoader(BaseLoader):
    """从`LarkSuite`（`飞书`）加载。"""

[docs]    def __init__(self, domain: str, access_token: str, document_id: str):
        """初始化领域、访问令牌（租户/用户）和文档ID。

参数：
    domain：要加载LarkSuite的域。
    access_token：要使用的访问令牌。
    document_id：要加载的文档ID。
"""
        self.domain = domain
        self.access_token = access_token
        self.document_id = document_id

    def _get_larksuite_api_json_data(self, api_url: str) -> Any:
        """获取 LarkSuite (飞书) API 响应的 JSON 数据。"""
        headers = {"Authorization": f"Bearer {self.access_token}"}
        request = urllib.request.Request(api_url, headers=headers)
        with urllib.request.urlopen(request) as response:
            json_data = json.loads(response.read().decode())
            return json_data

[docs]    def lazy_load(self) -> Iterator[Document]:
        """延迟加载 LarkSuite（飞书）文档。"""
        api_url_prefix = f"{self.domain}/open-apis/docx/v1/documents"
        metadata_json = self._get_larksuite_api_json_data(
            f"{api_url_prefix}/{self.document_id}"
        )
        raw_content_json = self._get_larksuite_api_json_data(
            f"{api_url_prefix}/{self.document_id}/raw_content"
        )
        text = raw_content_json["data"]["content"]
        metadata = {
            "document_id": self.document_id,
            "revision_id": metadata_json["data"]["document"]["revision_id"],
            "title": metadata_json["data"]["document"]["title"],
        }
        yield Document(page_content=text, metadata=metadata)


[docs]class LarkSuiteWikiLoader(LarkSuiteDocLoader):
    """从`LarkSuite`（`飞书`）wiki加载。"""

[docs]    def __init__(self, domain: str, access_token: str, wiki_id: str):
        """初始化领域、访问令牌（租户/用户）和wiki_id。

参数：
    domain: 加载LarkSuite的域。
    access_token: 要使用的访问令牌。
    wiki_id: 要加载的wiki_id。
"""
        self.domain = domain
        self.access_token = access_token
        self.wiki_id = wiki_id
        self.document_id = ""

[docs]    def lazy_load(self) -> Iterator[Document]:
        """延迟加载 LarkSuite（飞书）的wiki文档。"""

        # convert Feishu wiki id to document id
        if not self.document_id:
            wiki_url_prefix = f"{self.domain}/open-apis/wiki/v2/spaces/get_node"
            wiki_node_info_json = self._get_larksuite_api_json_data(
                f"{wiki_url_prefix}?token={self.wiki_id}"
            )
            self.document_id = wiki_node_info_json["data"]["node"]["obj_token"]

        yield from super().lazy_load()
Source code for langchain_community.document_loaders.larksuite
