gmakstutis/ragflow
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

removed unnecessary code format

Browse source
This commit is contained in:
David L. Carrascal 2025-12-10 13:05:38 +01:00
parent d923ea021e
commit 699bfd41f1
1 changed files with 48 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

71
rag/svr/sync_data_source.py
View file

@ -70,13 +70,15 @@ class SyncBase:
return return
except Exception as ex: except Exception as ex:
msg = "\n".join( msg = "\n".join([
[ "".join(traceback.format_exception_only(None, ex)).strip(),
"".join(traceback.format_exception_only(None, ex)).strip(), "".join(traceback.format_exception(None, ex, ex.__traceback__)).strip(),
"".join(traceback.format_exception(None, ex, ex.__traceback__)).strip(), ])
] SyncLogsService.update_by_id(task["id"], {
) "status": TaskStatus.FAIL,
SyncLogsService.update_by_id(task["id"], {"status": TaskStatus.FAIL, "full_exception_trace": msg, "error_msg": str(ex)}) "full_exception_trace": msg,
"error_msg": str(ex)
})
return return
SyncLogsService.schedule(task["connector_id"], task["kb_id"], task["poll_range_start"]) SyncLogsService.schedule(task["connector_id"], task["kb_id"], task["poll_range_start"])
@ -91,7 +93,7 @@ class SyncBase:
if task["poll_range_start"]: if task["poll_range_start"]:
next_update = task["poll_range_start"] next_update = task["poll_range_start"]
async for document_batch in document_batch_generator: # 如果是 async generator async for document_batch in document_batch_generator: # 如果是 async generator
if not document_batch: if not document_batch:
continue continue
@ -117,8 +119,15 @@ class SyncBase:
try: try:
e, kb = KnowledgebaseService.get_by_id(task["kb_id"]) e, kb = KnowledgebaseService.get_by_id(task["kb_id"])
err, dids = SyncLogsService.duplicate_and_parse(kb, docs, task["tenant_id"], f"{self.SOURCE_NAME}/{task['connector_id']}", task["auto_parse"]) err, dids = SyncLogsService.duplicate_and_parse(
SyncLogsService.increase_docs(task["id"], min_update, max_update, len(docs), "\n".join(err), len(err)) kb, docs, task["tenant_id"],
f"{self.SOURCE_NAME}/{task['connector_id']}",
task["auto_parse"]
)
SyncLogsService.increase_docs(
task["id"], min_update, max_update,
len(docs), "\n".join(err), len(err)
)
doc_num += len(docs) doc_num += len(docs)
@ -342,7 +351,9 @@ class Dropbox(SyncBase):
begin_info = "totally" begin_info = "totally"
else: else:
poll_start = task["poll_range_start"] poll_start = task["poll_range_start"]
document_generator = self.connector.poll_source(poll_start.timestamp(), datetime.now(timezone.utc).timestamp()) document_generator = self.connector.poll_source(
poll_start.timestamp(), datetime.now(timezone.utc).timestamp()
)
begin_info = f"from {poll_start}" begin_info = f"from {poll_start}"
logging.info(f"[Dropbox] Connect to Dropbox {begin_info}") logging.info(f"[Dropbox] Connect to Dropbox {begin_info}")
@ -500,7 +511,9 @@ class Jira(SyncBase):
) )
for document, failure, next_checkpoint in generator: for document, failure, next_checkpoint in generator:
if failure is not None: if failure is not None:
logging.warning(f"[Jira] Jira connector failure: {getattr(failure, 'failure_message', failure)}") logging.warning(
f"[Jira] Jira connector failure: {getattr(failure, 'failure_message', failure)}"
)
continue continue
if document is not None: if document is not None:
pending_docs.append(document) pending_docs.append(document)
@ -555,12 +568,15 @@ class WebDAV(SyncBase):
SOURCE_NAME: str = FileSource.WEBDAV SOURCE_NAME: str = FileSource.WEBDAV
async def _generate(self, task: dict): async def _generate(self, task: dict):
self.connector = WebDAVConnector(base_url=self.conf["base_url"], remote_path=self.conf.get("remote_path", "/")) self.connector = WebDAVConnector(
base_url=self.conf["base_url"],
remote_path=self.conf.get("remote_path", "/")
)
self.connector.load_credentials(self.conf["credentials"]) self.connector.load_credentials(self.conf["credentials"])
logging.info(f"Task info: reindex={task['reindex']}, poll_range_start={task['poll_range_start']}") logging.info(f"Task info: reindex={task['reindex']}, poll_range_start={task['poll_range_start']}")
if task["reindex"] == "1" or not task["poll_range_start"]: if task["reindex"]=="1" or not task["poll_range_start"]:
logging.info("Using load_from_state (full sync)") logging.info("Using load_from_state (full sync)")
document_batch_generator = self.connector.load_from_state() document_batch_generator = self.connector.load_from_state()
begin_info = "totally" begin_info = "totally"
@ -570,17 +586,23 @@ class WebDAV(SyncBase):
logging.info(f"Polling WebDAV from {task['poll_range_start']} (ts: {start_ts}) to now (ts: {end_ts})") logging.info(f"Polling WebDAV from {task['poll_range_start']} (ts: {start_ts}) to now (ts: {end_ts})")
document_batch_generator = self.connector.poll_source(start_ts, end_ts) document_batch_generator = self.connector.poll_source(start_ts, end_ts)
begin_info = "from {}".format(task["poll_range_start"]) begin_info = "from {}".format(task["poll_range_start"])
logging.info("Connect to WebDAV: {}(path: {}) {}".format(self.conf["base_url"], self.conf.get("remote_path", "/"), begin_info)) logging.info("Connect to WebDAV: {}(path: {}) {}".format(
self.conf["base_url"],
self.conf.get("remote_path", "/"),
begin_info
))
return document_batch_generator return document_batch_generator
class Moodle(SyncBase): class Moodle(SyncBase):
SOURCE_NAME: str = FileSource.MOODLE SOURCE_NAME: str = FileSource.MOODLE
async def _generate(self, task: dict): async def _generate(self, task: dict):
self.connector = MoodleConnector(moodle_url=self.conf["moodle_url"], batch_size=self.conf.get("batch_size", INDEX_BATCH_SIZE)) self.connector = MoodleConnector(
moodle_url=self.conf["moodle_url"],
batch_size=self.conf.get("batch_size", INDEX_BATCH_SIZE)
)
self.connector.load_credentials(self.conf["credentials"]) self.connector.load_credentials(self.conf["credentials"])
# Determine the time range for synchronization based on reindex or poll_range_start # Determine the time range for synchronization based on reindex or poll_range_start
@ -593,7 +615,10 @@ class Moodle(SyncBase):
document_generator = self.connector.load_from_state() document_generator = self.connector.load_from_state()
begin_info = "totally" begin_info = "totally"
else: else:
document_generator = self.connector.poll_source(poll_start.timestamp(), datetime.now(timezone.utc).timestamp()) document_generator = self.connector.poll_source(
poll_start.timestamp(),
datetime.now(timezone.utc).timestamp()
)
begin_info = "from {}".format(poll_start) begin_info = "from {}".format(poll_start)
logging.info("Connect to Moodle: {} {}".format(self.conf["moodle_url"], begin_info)) logging.info("Connect to Moodle: {} {}".format(self.conf["moodle_url"], begin_info))