From 2932b65da6b69f834e52888a0c13157d646d7d34 Mon Sep 17 00:00:00 2001 From: NeedmeFordev <124189514+spider-yamet@users.noreply.github.com> Date: Thu, 30 Apr 2026 06:05:12 +0200 Subject: [PATCH] feat(seafile): support deleted-file sync via slim snapshot (#14499) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### What problem does this PR solve? Incremental Seafile sync only ingests files whose modification time falls in the poll window; documents removed in Seafile were never removed from the knowledge base. This contributes to [#14362](https://github.com/infiniflow/ragflow/issues/14362) (datasource “sync deleted files” coordination). This PR adds a **slim snapshot** (`retrieve_all_slim_docs_perm_sync`) that enumerates current remote file IDs **without downloading content**, using the same logical IDs as full ingest (`seafile:{repo_id}:{file_id}`). When **`sync_deleted_files`** is enabled on incremental runs, **`SeaFile._generate`** returns **`(document_generator, file_list)`** so **`SyncBase`** can run **`cleanup_stale_documents_for_task`** and remove stale KB documents. ### Type of change - [x] New Feature (non-breaking change which adds functionality) ### What changed - **`common/data_source/seafile_connector.py`**: `SeaFileConnector` implements **`SlimConnectorWithPermSync`**; **`_list_files_recursive(..., filter_by_mtime=...)`** supports full-tree listing for snapshots; **`retrieve_all_slim_docs_perm_sync()`** reuses the same library/root scan as ingest and applies the same **size** ceiling; logging for snapshot start/end and counts. - **`rag/svr/sync_data_source.py`**: **`SeaFile._generate`** validates **`batch_size`**, captures **`end_ts`** before snapshot + **`poll_source`**, wraps slim retrieval in **`try`/`except`** ( **`file_list = None`** on failure so ingest continues), returns **`(generator, file_list)`**. - **`web/src/pages/user-setting/data-source/constant/index.tsx`**: **`syncDeletedFiles`** for Seafile in **`DataSourceFeatureVisibilityMap`**. --- common/data_source/seafile_connector.py | 114 ++++++++++++++++-- rag/svr/sync_data_source.py | 37 +++++- .../data-source/constant/index.tsx | 3 + 3 files changed, 141 insertions(+), 13 deletions(-) diff --git a/common/data_source/seafile_connector.py b/common/data_source/seafile_connector.py index ef7afeecf4..66bcf954fd 100644 --- a/common/data_source/seafile_connector.py +++ b/common/data_source/seafile_connector.py @@ -20,17 +20,19 @@ from common.data_source.exceptions import ( CredentialExpiredError, InsufficientPermissionsError, ) -from common.data_source.interfaces import LoadConnector, PollConnector +from common.data_source.interfaces import LoadConnector, PollConnector, SlimConnectorWithPermSync from common.data_source.models import ( Document, SecondsSinceUnixEpoch, GenerateDocumentsOutput, + GenerateSlimDocumentOutput, SeafileSyncScope, + SlimDocument, ) logger = logging.getLogger(__name__) -class SeaFileConnector(LoadConnector, PollConnector): +class SeaFileConnector(LoadConnector, PollConnector, SlimConnectorWithPermSync): """SeaFile connector supporting account-, library- and directory-level sync. API endpoints used: @@ -357,8 +359,18 @@ class SeaFileConnector(LoadConnector, PollConnector): return self._get_repo_info_via_account(self.repo_id) @retry(tries=3, delay=1, backoff=2) - def _get_directory_entries(self, repo_id: str, path: str = "/") -> list[dict]: - """List directory contents using the appropriate endpoint.""" + def _get_directory_entries( + self, + repo_id: str, + path: str = "/", + *, + raise_on_failure: bool = False, + ) -> list[dict]: + """List directory contents using the appropriate endpoint. + + When ``raise_on_failure`` is True (used for slim snapshots), HTTP/API errors + propagate so callers do not treat a failed listing as an empty directory. + """ try: if self._use_repo_token: # GET /api/v2.1/via-repo-token/dir/?path=/foo @@ -380,6 +392,8 @@ class SeaFileConnector(LoadConnector, PollConnector): logger.warning( "Error fetching directory %s in repo %s: %s", path, repo_id, e, ) + if raise_on_failure: + raise return [] @retry(tries=3, delay=1, backoff=2) @@ -412,9 +426,14 @@ class SeaFileConnector(LoadConnector, PollConnector): path: str, start: datetime, end: datetime, + *, + filter_by_mtime: bool = True, + strict_listing: bool = False, ) -> list[tuple[str, dict, dict]]: files = [] - entries = self._get_directory_entries(repo_id, path) + entries = self._get_directory_entries( + repo_id, path, raise_on_failure=strict_listing, + ) for entry in entries: entry_type = entry.get("type") @@ -424,15 +443,33 @@ class SeaFileConnector(LoadConnector, PollConnector): if entry_type == "dir": files.extend( self._list_files_recursive( - repo_id, repo_name, entry_path, start, end, + repo_id, + repo_name, + entry_path, + start, + end, + filter_by_mtime=filter_by_mtime, + strict_listing=strict_listing, ) ) elif entry_type == "file": modified = self._parse_mtime(entry.get("mtime")) - if start < modified <= end: + if filter_by_mtime: + if start < modified <= end: + files.append( + ( + entry_path, + entry, + {"id": repo_id, "name": repo_name}, + ) + ) + else: files.append( - (entry_path, entry, - {"id": repo_id, "name": repo_name}) + ( + entry_path, + entry, + {"id": repo_id, "name": repo_name}, + ) ) return files @@ -473,6 +510,8 @@ class SeaFileConnector(LoadConnector, PollConnector): try: files = self._list_files_recursive( lib["id"], lib["name"], root, start, end, + filter_by_mtime=True, + strict_listing=False, ) all_files.extend(files) except Exception as e: @@ -539,4 +578,59 @@ class SeaFileConnector(LoadConnector, PollConnector): for batch in self._yield_seafile_documents(start_dt, end_dt): yield batch - \ No newline at end of file + def retrieve_all_slim_docs_perm_sync( + self, + callback: Any = None, + ) -> GenerateSlimDocumentOutput: + """Full snapshot of file IDs eligible for indexing (no downloads). + + Uses ``seafile:{repo_id}:{file_id}`` matching :meth:`_yield_seafile_documents`. + Listing uses strict directory reads (errors propagate) so partial snapshots + are never treated as authoritative for stale-document cleanup. + """ + del callback + logger.info( + "Starting SeaFile slim snapshot: scope=%s url=%s", + self.sync_scope.value, + self.seafile_url, + ) + + libraries = self._resolve_libraries_to_scan() + all_files: list[tuple[str, dict, dict]] = [] + for lib in libraries: + root = self._root_path_for_repo(lib["id"]) + span_start = datetime(1970, 1, 1, tzinfo=timezone.utc) + span_end = datetime.now(timezone.utc) + listed = self._list_files_recursive( + lib["id"], + lib["name"], + root, + span_start, + span_end, + filter_by_mtime=False, + strict_listing=True, + ) + all_files.extend(listed) + + batch: list[SlimDocument] = [] + total = 0 + for file_path, file_entry, library in all_files: + file_size = file_entry.get("size", 0) + if file_size > self.size_threshold: + continue + file_id = file_entry.get("id", "") + repo_id = library["id"] + batch.append(SlimDocument(id=f"seafile:{repo_id}:{file_id}")) + total += 1 + if len(batch) >= self.batch_size: + yield batch + batch = [] + + if batch: + yield batch + + logger.info( + "Completed SeaFile slim snapshot: %d documents (listed_paths=%d)", + total, + len(all_files), + ) diff --git a/rag/svr/sync_data_source.py b/rag/svr/sync_data_source.py index a3afbba902..8bcd0d6d84 100644 --- a/rag/svr/sync_data_source.py +++ b/rag/svr/sync_data_source.py @@ -1412,9 +1412,17 @@ class SeaFile(SyncBase): async def _generate(self, task: dict): conf = self.conf + raw_batch_size = conf.get("batch_size", INDEX_BATCH_SIZE) + try: + batch_size = int(raw_batch_size) + except (TypeError, ValueError): + batch_size = INDEX_BATCH_SIZE + if batch_size <= 0: + batch_size = INDEX_BATCH_SIZE + self.connector = SeaFileConnector( seafile_url=conf["seafile_url"], - batch_size=conf.get("batch_size", INDEX_BATCH_SIZE), + batch_size=batch_size, include_shared=conf.get("include_shared", True), sync_scope=conf.get("sync_scope", SeafileSyncScope.ACCOUNT), repo_id=conf.get("repo_id") or None, @@ -1422,14 +1430,37 @@ class SeaFile(SyncBase): ) self.connector.load_credentials(conf["credentials"]) + file_list = None poll_start = task.get("poll_range_start") if task["reindex"] == "1" or poll_start is None: document_generator = self.connector.load_from_state() _begin_info = "totally" else: + end_ts = datetime.now(timezone.utc).timestamp() + if self.conf.get("sync_deleted_files"): + file_list = [] + logging.info( + "SeaFile: fetching slim snapshot for stale-document reconciliation " + "(connector_id=%s, kb_id=%s, scope=%s)", + task["connector_id"], + task["kb_id"], + conf.get("sync_scope") + or SeafileSyncScope.ACCOUNT.value, + ) + try: + for slim_batch in self.connector.retrieve_all_slim_docs_perm_sync(): + file_list.extend(slim_batch) + except Exception: + logging.exception( + "SeaFile slim snapshot failed; continuing without stale-document cleanup " + "(connector_id=%s, kb_id=%s)", + task["connector_id"], + task["kb_id"], + ) + file_list = None document_generator = self.connector.poll_source( poll_start.timestamp(), - datetime.now(timezone.utc).timestamp(), + end_ts, ) _begin_info = f"from {poll_start}" @@ -1441,7 +1472,7 @@ class SeaFile(SyncBase): extra += f" path={conf.get('sync_path')}" self.log_connection("SeaFile", f"{conf['seafile_url']} (scope={scope}{extra})", task) - return document_generator + return document_generator, file_list class DingTalkAITable(SyncBase): diff --git a/web/src/pages/user-setting/data-source/constant/index.tsx b/web/src/pages/user-setting/data-source/constant/index.tsx index 32619c05f0..86fcdeb8eb 100644 --- a/web/src/pages/user-setting/data-source/constant/index.tsx +++ b/web/src/pages/user-setting/data-source/constant/index.tsx @@ -105,6 +105,9 @@ export const DataSourceFeatureVisibilityMap: Partial< [DataSourceKey.AIRTABLE]: { syncDeletedFiles: true, }, + [DataSourceKey.SEAFILE]: { + syncDeletedFiles: true, + }, }; const isDataSourceFeatureVisible = (