mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-07-16 12:47:19 +08:00
feat(seafile): support deleted-file sync via slim snapshot (#14499)
### 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`**.
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
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),
|
||||
)
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -105,6 +105,9 @@ export const DataSourceFeatureVisibilityMap: Partial<
|
||||
[DataSourceKey.AIRTABLE]: {
|
||||
syncDeletedFiles: true,
|
||||
},
|
||||
[DataSourceKey.SEAFILE]: {
|
||||
syncDeletedFiles: true,
|
||||
},
|
||||
};
|
||||
|
||||
const isDataSourceFeatureVisible = (
|
||||
|
||||
Reference in New Issue
Block a user