zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-08-04 14:50:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor: Doc change parser (#14327)

Browse Source 
### What problem does this PR solve?

Before migration
Web API: POST /v1/document/change_parser
HTTP API: PATCH /api/v1/datasets/<dataset_id>/documents

After consolidation, Restful API
PATCH /api/v1/datasets/<dataset_id>/documents

### Type of change

- [x] Refactoring
This commit is contained in:
Jack
2026-04-27 23:42:57 +08:00
committed by GitHub
parent 872ff08304
commit c81081f8ef
14 changed files with 272 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
api/apps/document_app.py
View File

@@ -23,16 +23,11 @@ from api.db.services.document_service import DocumentService
from api.db.services.file2document_service import File2DocumentService
from api.utils.api_utils import (
get_data_error_result,
get_json_result,
get_request_json,
server_error_response,
validate_request,
)
from api.utils.web_utils import CONTENT_TYPE_MAP, apply_safe_file_response_headers
from common import settings
from common.constants import RetCode, TaskStatus
from common.misc_utils import thread_pool_exec
from rag.nlp import search
@manager.route("/get/<doc_id>", methods=["GET"]) # noqa: F821
@@ -74,56 +69,3 @@ async def download_attachment(attachment_id):
except Exception as e:
return server_error_response(e)
@manager.route("/change_parser", methods=["POST"]) # noqa: F821
@login_required
@validate_request("doc_id")
async def change_parser():
req = await get_request_json()
if not DocumentService.accessible(req["doc_id"], current_user.id):
return get_json_result(data=False, message="No authorization.", code=RetCode.AUTHENTICATION_ERROR)
e, doc = DocumentService.get_by_id(req["doc_id"])
if not e:
return get_data_error_result(message="Document not found!")
def reset_doc():
nonlocal doc
e = DocumentService.update_by_id(doc.id, {"pipeline_id": req["pipeline_id"], "parser_id": req["parser_id"], "progress": 0, "progress_msg": "", "run": TaskStatus.UNSTART.value})
if not e:
return get_data_error_result(message="Document not found!")
if doc.token_num > 0:
e = DocumentService.increment_chunk_num(doc.id, doc.kb_id, doc.token_num * -1, doc.chunk_num * -1, doc.process_duration * -1)
if not e:
return get_data_error_result(message="Document not found!")
tenant_id = DocumentService.get_tenant_id(req["doc_id"])
if not tenant_id:
return get_data_error_result(message="Tenant not found!")
DocumentService.delete_chunk_images(doc, tenant_id)
if settings.docStoreConn.index_exist(search.index_name(tenant_id), doc.kb_id):
settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), doc.kb_id)
return None
try:
if "pipeline_id" in req and req["pipeline_id"] != "":
if doc.pipeline_id == req["pipeline_id"]:
return get_json_result(data=True)
DocumentService.update_by_id(doc.id, {"pipeline_id": req["pipeline_id"]})
reset_doc()
return get_json_result(data=True)
if doc.parser_id.lower() == req["parser_id"].lower():
if "parser_config" in req:
if req["parser_config"] == doc.parser_config:
return get_json_result(data=True)
else:
return get_json_result(data=True)
if (doc.type == FileType.VISUAL and req["parser_id"] != "picture") or (re.search(r"\.(ppt|pptx|pages)$", doc.name) and req["parser_id"] != "presentation"):
return get_data_error_result(message="Not supported yet!")
if "parser_config" in req:
DocumentService.update_parser_config(doc.id, req["parser_config"])
reset_doc()
return get_json_result(data=True)
except Exception as e:
return server_error_response(e)

23
api/apps/restful_apis/document_api.py
View File

@@ -24,10 +24,11 @@ from peewee import OperationalError
from pydantic import ValidationError
from api.apps import login_required
from api.apps.services.document_api_service import validate_document_update_fields, map_doc_keys, \
map_doc_keys_with_run_status, update_document_name_only, update_chunk_method_only, update_document_status_only
from api.constants import FILE_NAME_LEN_LIMIT, IMG_BASE64_PREFIX
from api.db import FileType, VALID_FILE_TYPES
from api.apps.services.document_api_service import validate_document_update_fields, map_doc_keys, \
map_doc_keys_with_run_status, update_document_name_only, update_chunk_method, update_document_status_only, \
reset_document_for_reparse
from api.db import VALID_FILE_TYPES, FileType
from api.db.services import duplicate_name
from api.db.services.doc_metadata_service import DocMetadataService
from api.db.db_models import Task
@@ -204,16 +205,26 @@ async def update_document(tenant_id, dataset_id, document_id):
if error := update_document_name_only(document_id, req["name"]):
return error
# "parser_id" provided but does not match with existing doc's file type
if "parser_id" in req and ((doc.type == FileType.VISUAL and req["parser_id"] != "picture")
or (re.search(r"\.(ppt|pptx|pages)$", doc.name) and req["parser_id"] != "presentation")):
return get_data_error_result(message="Not supported yet!")
# parser config provided (already validated in UpdateDocumentReq), update it
if update_doc_req.parser_config:
req["parser_config"].update(update_doc_req.parser_config.ext)
DocumentService.update_parser_config(doc.id, req["parser_config"])
# pipeline_id provided - reset document for reparse
if update_doc_req.pipeline_id:
if error := reset_document_for_reparse(doc, tenant_id, pipeline_id=update_doc_req.pipeline_id):
return error
# chunk method provided - the update method will check if it's different with existing one
if update_doc_req.chunk_method:
if error := update_chunk_method_only(req, doc, dataset_id, tenant_id):
elif update_doc_req.chunk_method:
if error := update_chunk_method(req, doc, tenant_id):
return error
if "enabled" in req: # already checked in UpdateDocumentReq - it's int if it's present
if "enabled" in req: # already checked in UpdateDocumentReq - it's int if present
# "enabled" flag provided, the update method will check if it's changed and then update if so
if error := update_document_status_only(int(req["enabled"]), doc, kb):
return error

70
api/apps/services/document_api_service.py
View File

@@ -13,6 +13,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging
from api.db.services.document_service import DocumentService
from api.db.services.file2document_service import File2DocumentService
from api.db.services.file_service import FileService
@@ -58,7 +60,7 @@ def update_document_name_only(document_id, req_doc_name):
)
return None
def update_chunk_method_only(req, doc, dataset_id, tenant_id):
def update_chunk_method(req, doc, tenant_id):
"""
Update chunk method only (without validation).
@@ -69,28 +71,56 @@ def update_chunk_method_only(req, doc, dataset_id, tenant_id):
Args:
req: The request dictionary containing chunk_method and parser_config.
doc: The document model from the database.
dataset_id: The ID of the dataset containing the document.
tenant_id: The tenant ID for the document store.
Returns:
None if successful, or an error result dictionary if failed.
"""
if doc.parser_id.lower() != req["chunk_method"].lower():
# if chunk method changed
e = DocumentService.update_by_id(
doc.id,
{
"parser_id": req["chunk_method"],
"progress": 0,
"progress_msg": "",
"run": TaskStatus.UNSTART.value,
},
)
if not e:
return get_error_data_result(message="Document not found!")
# if chunk method changed, reset document for reparse
result = reset_document_for_reparse(doc, tenant_id, parser_id=req["chunk_method"])
if result:
return result
if not req.get("parser_config"):
req["parser_config"] = get_parser_config(req["chunk_method"], req.get("parser_config"))
DocumentService.update_parser_config(doc.id, req["parser_config"])
return None
def reset_document_for_reparse(doc, tenant_id, parser_id=None, pipeline_id=None):
"""
Reset document for reparsing.
Updates the parser_id and/or pipeline_id for a document, resets its progress,
clears existing chunks from the document store, and removes chunk images.
Args:
doc: The document model from the database.
tenant_id: The tenant ID for the document store.
parser_id: Optional new parser_id (chunk method). If None, keeps existing.
pipeline_id: Optional new pipeline_id. If None, keeps existing.
Returns:
None if successful, or an error result dictionary if failed.
"""
# Build update fields
update_fields = {
"progress": 0,
"progress_msg": "",
"run": TaskStatus.UNSTART.value,
}
if parser_id is not None:
update_fields["parser_id"] = parser_id
if pipeline_id is not None:
update_fields["pipeline_id"] = pipeline_id
# Update document
e = DocumentService.update_by_id(doc.id, update_fields)
if not e:
return get_error_data_result(message="Document not found!")
# Delete chunks from document store
if doc.token_num > 0:
e = DocumentService.increment_chunk_num(
doc.id,
@@ -98,12 +128,20 @@ def update_chunk_method_only(req, doc, dataset_id, tenant_id):
doc.token_num * -1,
doc.chunk_num * -1,
doc.process_duration * -1,
)
)
if not e:
return get_error_data_result(message="Document not found!")
settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), dataset_id)
settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), doc.kb_id)
# Delete chunk images
try:
DocumentService.delete_chunk_images(doc, tenant_id)
except Exception as e:
logging.error(f"error when delete chunk images:{e}")
return None
def update_document_status_only(status:int, doc, kb):
"""
Update document status only (without validation).

1
api/utils/validation_utils.py
View File

@@ -411,6 +411,7 @@ class UpdateDocumentReq(Base):
model_config = ConfigDict(extra='ignore')
name: Annotated[str | None, Field(default=None, max_length=65535)]
chunk_method: Annotated[str | None, Field(default=None, max_length=65535)]
pipeline_id: Annotated[str | None, Field(default=None, max_length=65535)]
enabled: Annotated[int | None, Field(default=None, ge=0, le=1)]
chunk_count: Annotated[int | None, Field(default=None, ge=0)]
token_count: Annotated[int | None, Field(default=None, ge=0)]