如何为 RAGFlow 添加新的数据源

class SyncBase:
    SOURCE_NAME: str = None

    async def __call__(self, task: dict):
        ...
        
    async def _generate(self, task: dict):
        raise NotImplementedError

class Document(BaseModel):
    id: str
    source: str
    semantic_identifier: str
    extension: str
    blob: bytes
    doc_updated_at: datetime
    size_bytes: int

class LoadConnector(ABC):
    @abstractmethod
    def load_credentials(self, credentials: Dict[str, Any]) -> Dict[str, Any] | None: ...

    @abstractmethod
    def load_from_state(self) -> Generator[list[Document], None, None]: 
        """load all documents up to now"""
        ...

    @abstractmethod
    def validate_connector_settings(self) -> None: ...

class PollConnector(ABC):
    @abstractmethod
    def poll_source(self, start: SecondsSinceUnixEpoch, end: SecondsSinceUnixEpoch) -> Generator[list[Document], None, None]: 
        """load documents from start to end"""
        ...

class S3(SyncBase):
    SOURCE_NAME: str = FileSource.S3

    asyncdef_generate(self, task: dict):
        self.connector = BlobStorageConnector(
            bucket_type=self.conf.get("bucket_type", "s3"),
            bucket_name=self.conf["bucket_name"],
            prefix=self.conf.get("prefix", "")
        )
        self.connector.load_credentials(self.conf["credentials"])
        document_batch_generator = (
            self.connector.load_from_state()
            if task["reindex"] == "1"ornot task["poll_range_start"]
            elseself.connector.poll_source(
                task["poll_range_start"].timestamp(),
                datetime.now(timezone.utc).timestamp()
            )
        )
        return document_batch_generator

class BlobStorageConnector(LoadConnector, PollConnector):
    defload_from_state(self) -> GenerateDocumentsOutput:
        returnself._yield_blob_objects(
            start=datetime(1970, 1, 1, tzinfo=timezone.utc),
            end=datetime.now(timezone.utc),
        )

    defpoll_source(self, start: SecondsSinceUnixEpoch, end: SecondsSinceUnixEpoch) -> GenerateDocumentsOutput:
        start_datetime = datetime.fromtimestamp(start, tz=timezone.utc)
        end_datetime = datetime.fromtimestamp(end, tz=timezone.utc)
        for batch inself._yield_blob_objects(start_datetime, end_datetime):
            yield batch

    defvalidate_connector_settings(self) -> None:
        ...

class CheckpointedConnector(BaseConnector[CT]):
    @abc.abstractmethod
    defload_from_checkpoint(
        self,
        start: SecondsSinceUnixEpoch,
        end: SecondsSinceUnixEpoch,
        checkpoint: CT,
    ) -> CheckpointOutput[CT]:
        ...

    @abc.abstractmethod
    defbuild_dummy_checkpoint(self) -> CT:
        ...

    @abc.abstractmethod
    defvalidate_checkpoint_json(self, checkpoint_json: str) -> CT:
        ...

class ConfluenceCheckpoint(ConnectorCheckpoint):
    next_page_url: str | None

classConfluenceConnector(
    CheckpointedConnector[ConfluenceCheckpoint],
    SlimConnector,
    SlimConnectorWithPermSync,
    CredentialsConnector,
):
    defload_from_checkpoint(
        self,
        start: SecondsSinceUnixEpoch,
        end: SecondsSinceUnixEpoch,
        checkpoint: ConfluenceCheckpoint,
    ) -> CheckpointOutput[ConfluenceCheckpoint]:
        end += ONE_DAY  # handle time zone weirdness
        try:
            returnself._fetch_document_batches(checkpoint, start, end)
        except Exception as e:
            ...

    defbuild_dummy_checkpoint(self) -> ConfluenceCheckpoint:
        return ConfluenceCheckpoint(has_more=True, next_page_url=None)

    defvalidate_checkpoint_json(self, checkpoint_json: str) -> ConfluenceCheckpoint:
        return ConfluenceCheckpoint.model_validate_json(checkpoint_json)

for doc in load_all_docs_from_checkpoint_connector(
    connector=confluence_connector,
    start=start,
    end=end,
):
    print(doc)

class CheckpointedConnectorWithPermSync(ABC):
    @abstractmethod
    defload_from_checkpoint(
        self,
        start: SecondsSinceUnixEpoch,
        end: SecondsSinceUnixEpoch,
        checkpoint: ConnectorCheckpoint,
    ) -> Generator[Document | ConnectorFailure, None, ConnectorCheckpoint]:
        ...

    @abstractmethod
    defload_from_checkpoint_with_perm_sync(
        self,
        start: SecondsSinceUnixEpoch,
        end: SecondsSinceUnixEpoch,
        checkpoint: ConnectorCheckpoint,
    ) -> Generator[Document | ConnectorFailure, None, ConnectorCheckpoint]:
        ...

    @abstractmethod
    defbuild_dummy_checkpoint(self) -> ConnectorCheckpoint:
        ...

    @abstractmethod
    defvalidate_checkpoint_json(self, checkpoint_json: str) -> ConnectorCheckpoint:
        ...

class JiraCheckpoint(ConnectorCheckpoint):
    """Checkpoint that tracks which slice of the current JQL result set was emitted."""

    start_at: int = 0
    cursor: str | None = None
    ids_done: bool = False
    all_issue_ids: list[list[str]]


class JiraConnector(CheckpointedConnectorWithPermSync):
    ...

def document_batches():
    checkpoint = self.connector.build_dummy_checkpoint()
    pending_docs = []
    ...
    while checkpoint.has_more:
        wrapper = CheckpointOutputWrapper()
        generator = wrapper(
            self.connector.load_from_checkpoint(
                start_time,
                end_time,
                checkpoint,
            )
        )
        for document, failure, next_checkpoint in generator:
            if failure isnotNone:
                continue
            if document isnotNone:
                pending_docs.append(document)
                iflen(pending_docs) >= batch_size:
                    yield pending_docs
                    pending_docs = []
            if next_checkpoint isnotNone:
                checkpoint = next_checkpoint
    ...
    if pending_docs:
        yield pending_docs

class S3(SyncBase):
    SOURCE_NAME: str = FileSource.S3

    asyncdef_generate(self, task: dict):
        self.connector = BlobStorageConnector(
            bucket_type=self.conf.get("bucket_type", "s3"),
            bucket_name=self.conf["bucket_name"],
            prefix=self.conf.get("prefix", "")
        )
        self.connector.load_credentials(self.conf["credentials"])
        document_batch_generator = (
            self.connector.load_from_state()
            if task["reindex"] == "1"ornot task["poll_range_start"]
            elseself.connector.poll_source(
                task["poll_range_start"].timestamp(),
                datetime.now(timezone.utc).timestamp()
            )
        )
        begin_info = "totally"if task["reindex"] == "1"ornot task["poll_range_start"] elsef"from {task['poll_range_start']}"
        logging.info(
            f"Connect to S3: {self.conf['bucket_name']}/"
            f"{self.conf.get('prefix', '')} {begin_info}"
        )
        return document_batch_generator

class BlobStorageConnector(LoadConnector, PollConnector):
    def__init__(self, bucket_type: str, bucket_name: str, prefix: str = "", batch_size: int = INDEX_BATCH_SIZE, european_residency: bool = False) -> None:
        self.bucket_type: BlobType = BlobType(bucket_type)
        self.bucket_name = bucket_name.strip()
        self.prefix = prefix ifnot prefix or prefix.endswith("/") else prefix + "/"
        self.batch_size = batch_size
        self.s3_client: Optional[Any] = None

    defload_credentials(self, credentials: dict[str, Any]) -> dict[str, Any] | None:
        ifself.bucket_type == BlobType.S3:
            authentication_method = credentials.get("authentication_method", "access_key")
            if authentication_method == "access_key":
                ifnotall(credentials.get(key) for key in ["aws_access_key_id", "aws_secret_access_key"]):
                    raise ConnectorMissingCredentialError("Amazon S3")
            elif authentication_method == "iam_role":
                ifnot credentials.get("aws_role_arn"):
                    raise ConnectorMissingCredentialError("Amazon S3 IAM role ARN is required")
        ...
        self.s3_client = create_s3_client(self.bucket_type, credentials, self.european_residency)
        returnNone

    def_yield_blob_objects(self, start: datetime, end: datetime) -> GenerateDocumentsOutput:
        paginator = self.s3_client.get_paginator("list_objects_v2")
        pages = paginator.paginate(Bucket=self.bucket_name, Prefix=self.prefix)
        batch: list[Document] = []
        for page in pages:
            for obj in page.get("Contents", []):
                ...
                file_name = os.path.basename(obj["Key"])
                blob = download_object(self.s3_client, self.bucket_name, obj["Key"], self.size_threshold)
                if blob isNone:
                    continue
                batch.append(Document(
                    id=f"{self.bucket_type}:{self.bucket_name}:{obj['Key']}",
                    blob=blob,
                    source=DocumentSource(self.bucket_type.value),
                    semantic_identifier=file_name,
                    extension=get_file_ext(file_name),
                    doc_updated_at=last_modified,
                    size_bytes=extract_size_bytes(obj) or0
                ))
                iflen(batch) == self.batch_size:
                    yield batch
        if batch:
            yield batch