Storage: Rename all the variables about get file to storage from minio. (#2497)
Browse fileshttps://github.com/infiniflow/ragflow/issues/2496
### What problem does this PR solve?
_Briefly describe what this PR aims to solve. Include background context
that will help reviewers understand the purpose of the PR._
### Type of change
- [ ] Bug Fix (non-breaking change which fixes an issue)
- [ ] New Feature (non-breaking change which adds functionality)
- [ ] Documentation Update
- [x] Refactoring
- [ ] Performance Improvement
- [ ] Other (please describe):
- api/apps/api_app.py +2 -2
- api/apps/dataset_api.py +3 -3
- api/apps/document_app.py +3 -3
- api/apps/file_app.py +1 -1
- api/apps/sdk/doc.py +3 -3
- api/db/services/file2document_service.py +1 -1
- rag/svr/task_executor.py +3 -3
api/apps/api_app.py
CHANGED
|
@@ -478,7 +478,7 @@ def upload():
|
|
| 478 |
e, doc = DocumentService.get_by_id(doc["id"])
|
| 479 |
doc = doc.to_dict()
|
| 480 |
doc["tenant_id"] = tenant_id
|
| 481 |
-
bucket, name = File2DocumentService.
|
| 482 |
queue_tasks(doc, bucket, name)
|
| 483 |
except Exception as e:
|
| 484 |
return server_error_response(e)
|
|
@@ -640,7 +640,7 @@ def document_rm():
|
|
| 640 |
if not tenant_id:
|
| 641 |
return get_data_error_result(retmsg="Tenant not found!")
|
| 642 |
|
| 643 |
-
b, n = File2DocumentService.
|
| 644 |
|
| 645 |
if not DocumentService.remove_document(doc, tenant_id):
|
| 646 |
return get_data_error_result(
|
|
|
|
| 478 |
e, doc = DocumentService.get_by_id(doc["id"])
|
| 479 |
doc = doc.to_dict()
|
| 480 |
doc["tenant_id"] = tenant_id
|
| 481 |
+
bucket, name = File2DocumentService.get_storage_address(doc_id=doc["id"])
|
| 482 |
queue_tasks(doc, bucket, name)
|
| 483 |
except Exception as e:
|
| 484 |
return server_error_response(e)
|
|
|
|
| 640 |
if not tenant_id:
|
| 641 |
return get_data_error_result(retmsg="Tenant not found!")
|
| 642 |
|
| 643 |
+
b, n = File2DocumentService.get_storage_address(doc_id=doc_id)
|
| 644 |
|
| 645 |
if not DocumentService.remove_document(doc, tenant_id):
|
| 646 |
return get_data_error_result(
|
api/apps/dataset_api.py
CHANGED
|
@@ -420,7 +420,7 @@ def delete_document(document_id, dataset_id): # string
|
|
| 420 |
f" reason!", code=RetCode.AUTHENTICATION_ERROR)
|
| 421 |
|
| 422 |
# get the doc's id and location
|
| 423 |
-
real_dataset_id, location = File2DocumentService.
|
| 424 |
|
| 425 |
if real_dataset_id != dataset_id:
|
| 426 |
return construct_json_result(message=f"The document {document_id} is not in the dataset: {dataset_id}, "
|
|
@@ -595,7 +595,7 @@ def download_document(dataset_id, document_id):
|
|
| 595 |
code=RetCode.ARGUMENT_ERROR)
|
| 596 |
|
| 597 |
# The process of downloading
|
| 598 |
-
doc_id, doc_location = File2DocumentService.
|
| 599 |
file_stream = STORAGE_IMPL.get(doc_id, doc_location)
|
| 600 |
if not file_stream:
|
| 601 |
return construct_json_result(message="This file is empty.", code=RetCode.DATA_ERROR)
|
|
@@ -736,7 +736,7 @@ def parsing_document_internal(id):
|
|
| 736 |
doc_attributes = doc_attributes.to_dict()
|
| 737 |
doc_id = doc_attributes["id"]
|
| 738 |
|
| 739 |
-
bucket, doc_name = File2DocumentService.
|
| 740 |
binary = STORAGE_IMPL.get(bucket, doc_name)
|
| 741 |
parser_name = doc_attributes["parser_id"]
|
| 742 |
if binary:
|
|
|
|
| 420 |
f" reason!", code=RetCode.AUTHENTICATION_ERROR)
|
| 421 |
|
| 422 |
# get the doc's id and location
|
| 423 |
+
real_dataset_id, location = File2DocumentService.get_storage_address(doc_id=document_id)
|
| 424 |
|
| 425 |
if real_dataset_id != dataset_id:
|
| 426 |
return construct_json_result(message=f"The document {document_id} is not in the dataset: {dataset_id}, "
|
|
|
|
| 595 |
code=RetCode.ARGUMENT_ERROR)
|
| 596 |
|
| 597 |
# The process of downloading
|
| 598 |
+
doc_id, doc_location = File2DocumentService.get_storage_address(doc_id=document_id) # minio address
|
| 599 |
file_stream = STORAGE_IMPL.get(doc_id, doc_location)
|
| 600 |
if not file_stream:
|
| 601 |
return construct_json_result(message="This file is empty.", code=RetCode.DATA_ERROR)
|
|
|
|
| 736 |
doc_attributes = doc_attributes.to_dict()
|
| 737 |
doc_id = doc_attributes["id"]
|
| 738 |
|
| 739 |
+
bucket, doc_name = File2DocumentService.get_storage_address(doc_id=doc_id)
|
| 740 |
binary = STORAGE_IMPL.get(bucket, doc_name)
|
| 741 |
parser_name = doc_attributes["parser_id"]
|
| 742 |
if binary:
|
api/apps/document_app.py
CHANGED
|
@@ -297,7 +297,7 @@ def rm():
|
|
| 297 |
if not tenant_id:
|
| 298 |
return get_data_error_result(retmsg="Tenant not found!")
|
| 299 |
|
| 300 |
-
b, n = File2DocumentService.
|
| 301 |
|
| 302 |
if not DocumentService.remove_document(doc, tenant_id):
|
| 303 |
return get_data_error_result(
|
|
@@ -342,7 +342,7 @@ def run():
|
|
| 342 |
e, doc = DocumentService.get_by_id(id)
|
| 343 |
doc = doc.to_dict()
|
| 344 |
doc["tenant_id"] = tenant_id
|
| 345 |
-
bucket, name = File2DocumentService.
|
| 346 |
queue_tasks(doc, bucket, name)
|
| 347 |
|
| 348 |
return get_json_result(data=True)
|
|
@@ -393,7 +393,7 @@ def get(doc_id):
|
|
| 393 |
if not e:
|
| 394 |
return get_data_error_result(retmsg="Document not found!")
|
| 395 |
|
| 396 |
-
b, n = File2DocumentService.
|
| 397 |
response = flask.make_response(STORAGE_IMPL.get(b, n))
|
| 398 |
|
| 399 |
ext = re.search(r"\.([^.]+)$", doc.name)
|
|
|
|
| 297 |
if not tenant_id:
|
| 298 |
return get_data_error_result(retmsg="Tenant not found!")
|
| 299 |
|
| 300 |
+
b, n = File2DocumentService.get_storage_address(doc_id=doc_id)
|
| 301 |
|
| 302 |
if not DocumentService.remove_document(doc, tenant_id):
|
| 303 |
return get_data_error_result(
|
|
|
|
| 342 |
e, doc = DocumentService.get_by_id(id)
|
| 343 |
doc = doc.to_dict()
|
| 344 |
doc["tenant_id"] = tenant_id
|
| 345 |
+
bucket, name = File2DocumentService.get_storage_address(doc_id=doc["id"])
|
| 346 |
queue_tasks(doc, bucket, name)
|
| 347 |
|
| 348 |
return get_json_result(data=True)
|
|
|
|
| 393 |
if not e:
|
| 394 |
return get_data_error_result(retmsg="Document not found!")
|
| 395 |
|
| 396 |
+
b, n = File2DocumentService.get_storage_address(doc_id=doc_id)
|
| 397 |
response = flask.make_response(STORAGE_IMPL.get(b, n))
|
| 398 |
|
| 399 |
ext = re.search(r"\.([^.]+)$", doc.name)
|
api/apps/file_app.py
CHANGED
|
@@ -332,7 +332,7 @@ def get(file_id):
|
|
| 332 |
e, file = FileService.get_by_id(file_id)
|
| 333 |
if not e:
|
| 334 |
return get_data_error_result(retmsg="Document not found!")
|
| 335 |
-
b, n = File2DocumentService.
|
| 336 |
response = flask.make_response(STORAGE_IMPL.get(b, n))
|
| 337 |
ext = re.search(r"\.([^.]+)$", file.name)
|
| 338 |
if ext:
|
|
|
|
| 332 |
e, file = FileService.get_by_id(file_id)
|
| 333 |
if not e:
|
| 334 |
return get_data_error_result(retmsg="Document not found!")
|
| 335 |
+
b, n = File2DocumentService.get_storage_address(file_id=file_id)
|
| 336 |
response = flask.make_response(STORAGE_IMPL.get(b, n))
|
| 337 |
ext = re.search(r"\.([^.]+)$", file.name)
|
| 338 |
if ext:
|
api/apps/sdk/doc.py
CHANGED
|
@@ -286,7 +286,7 @@ def download_document(document_id,tenant_id):
|
|
| 286 |
code=RetCode.ARGUMENT_ERROR)
|
| 287 |
|
| 288 |
# The process of downloading
|
| 289 |
-
doc_id, doc_location = File2DocumentService.
|
| 290 |
file_stream = STORAGE_IMPL.get(doc_id, doc_location)
|
| 291 |
if not file_stream:
|
| 292 |
return construct_json_result(message="This file is empty.", code=RetCode.DATA_ERROR)
|
|
@@ -373,7 +373,7 @@ def rm(tenant_id):
|
|
| 373 |
if not tenant_id:
|
| 374 |
return get_data_error_result(retmsg="Tenant not found!")
|
| 375 |
|
| 376 |
-
b, n = File2DocumentService.
|
| 377 |
|
| 378 |
if not DocumentService.remove_document(doc, tenant_id):
|
| 379 |
return get_data_error_result(
|
|
@@ -438,7 +438,7 @@ def run(tenant_id):
|
|
| 438 |
e, doc = DocumentService.get_by_id(id)
|
| 439 |
doc = doc.to_dict()
|
| 440 |
doc["tenant_id"] = tenant_id
|
| 441 |
-
bucket, name = File2DocumentService.
|
| 442 |
queue_tasks(doc, bucket, name)
|
| 443 |
|
| 444 |
return get_json_result(data=True)
|
|
|
|
| 286 |
code=RetCode.ARGUMENT_ERROR)
|
| 287 |
|
| 288 |
# The process of downloading
|
| 289 |
+
doc_id, doc_location = File2DocumentService.get_storage_address(doc_id=document_id) # minio address
|
| 290 |
file_stream = STORAGE_IMPL.get(doc_id, doc_location)
|
| 291 |
if not file_stream:
|
| 292 |
return construct_json_result(message="This file is empty.", code=RetCode.DATA_ERROR)
|
|
|
|
| 373 |
if not tenant_id:
|
| 374 |
return get_data_error_result(retmsg="Tenant not found!")
|
| 375 |
|
| 376 |
+
b, n = File2DocumentService.get_storage_address(doc_id=doc_id)
|
| 377 |
|
| 378 |
if not DocumentService.remove_document(doc, tenant_id):
|
| 379 |
return get_data_error_result(
|
|
|
|
| 438 |
e, doc = DocumentService.get_by_id(id)
|
| 439 |
doc = doc.to_dict()
|
| 440 |
doc["tenant_id"] = tenant_id
|
| 441 |
+
bucket, name = File2DocumentService.get_storage_address(doc_id=doc["id"])
|
| 442 |
queue_tasks(doc, bucket, name)
|
| 443 |
|
| 444 |
return get_json_result(data=True)
|
api/db/services/file2document_service.py
CHANGED
|
@@ -69,7 +69,7 @@ class File2DocumentService(CommonService):
|
|
| 69 |
|
| 70 |
@classmethod
|
| 71 |
@DB.connection_context()
|
| 72 |
-
def
|
| 73 |
if doc_id:
|
| 74 |
f2d = cls.get_by_document_id(doc_id)
|
| 75 |
else:
|
|
|
|
| 69 |
|
| 70 |
@classmethod
|
| 71 |
@DB.connection_context()
|
| 72 |
+
def get_storage_address(cls, doc_id=None, file_id=None):
|
| 73 |
if doc_id:
|
| 74 |
f2d = cls.get_by_document_id(doc_id)
|
| 75 |
else:
|
rag/svr/task_executor.py
CHANGED
|
@@ -137,7 +137,7 @@ def collect():
|
|
| 137 |
return tasks
|
| 138 |
|
| 139 |
|
| 140 |
-
def
|
| 141 |
return STORAGE_IMPL.get(bucket, name)
|
| 142 |
|
| 143 |
|
|
@@ -155,8 +155,8 @@ def build(row):
|
|
| 155 |
chunker = FACTORY[row["parser_id"].lower()]
|
| 156 |
try:
|
| 157 |
st = timer()
|
| 158 |
-
bucket, name = File2DocumentService.
|
| 159 |
-
binary =
|
| 160 |
cron_logger.info(
|
| 161 |
"From minio({}) {}/{}".format(timer() - st, row["location"], row["name"]))
|
| 162 |
except TimeoutError as e:
|
|
|
|
| 137 |
return tasks
|
| 138 |
|
| 139 |
|
| 140 |
+
def get_storage_binary(bucket, name):
|
| 141 |
return STORAGE_IMPL.get(bucket, name)
|
| 142 |
|
| 143 |
|
|
|
|
| 155 |
chunker = FACTORY[row["parser_id"].lower()]
|
| 156 |
try:
|
| 157 |
st = timer()
|
| 158 |
+
bucket, name = File2DocumentService.get_storage_address(doc_id=row["doc_id"])
|
| 159 |
+
binary = get_storage_binary(bucket, name)
|
| 160 |
cron_logger.info(
|
| 161 |
"From minio({}) {}/{}".format(timer() - st, row["location"], row["name"]))
|
| 162 |
except TimeoutError as e:
|