zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-13 08:28:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix: #6098 - Add validation logic for parser_config when update document (#13911)

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

Add validation logic for parser_config.
Refactor the processing flow. Before change, validation logics and
update logics are mixed up - some validation logis executes followed by
some update logic executes and then another such
"validation-and-then-update" which is not good. After change, all
validation logic executes firstly. Update logic will be executed after
ALL validation logic executed.
Validation logic for parameters (that come from front end) will be
checked using Pydantic. For validation logic that depends on data from
DB, they will be in separate methods.

### Type of change

- [x] Bug Fix (non-breaking change which fixes an issue)
- [x] Refactoring
This commit is contained in:
Jack
2026-04-07 11:33:05 +08:00
committed by GitHub
parent 5673245134
commit c4b0aaa874
11 changed files with 1079 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

199
api/apps/sdk/doc.py
View File

@@ -16,17 +16,15 @@
import datetime
import json
import logging
import pathlib
import re
from io import BytesIO
import xxhash
from peewee import OperationalError
from pydantic import BaseModel, Field, validator
from pydantic import BaseModel, Field, validator, ValidationError
from quart import request, send_file
from api.constants import FILE_NAME_LEN_LIMIT
from api.db import FileType
from api.db.db_models import APIToken, Document, File, Task
from api.db.joint_services.tenant_model_service import get_model_config_by_id, get_model_config_by_type_and_name, get_tenant_default_model_by_type
from api.db.services.doc_metadata_service import DocMetadataService
@@ -37,8 +35,10 @@ from api.db.services.knowledgebase_service import KnowledgebaseService
from api.db.services.llm_service import LLMBundle
from api.db.services.task_service import TaskService, cancel_all_task_of, queue_tasks
from api.db.services.tenant_llm_service import TenantLLMService
from api.utils import validation_utils
from api.utils.api_utils import check_duplicate_ids, construct_json_result, get_error_data_result, get_parser_config, get_request_json, get_result, server_error_response, token_required
from api.utils.image_utils import store_chunk_image
from api.utils.validation_utils import format_validation_error_message, UpdateDocumentReq
from common import settings
from common.constants import FileSource, LLMType, ParserType, RetCode, TaskStatus
from common.metadata_utils import convert_conditions, meta_filter
@@ -185,6 +185,60 @@ async def upload(dataset_id, tenant_id):
return get_result(data=renamed_doc_list)
def _update_document_name_only(document_id, req_doc_name):
"""Update document name only (without validation)."""
if not DocumentService.update_by_id(document_id, {"name": req_doc_name}):
return get_error_data_result(message="Database error (Document rename)!")
informs = File2DocumentService.get_by_document_id(document_id)
if informs:
e, file = FileService.get_by_id(informs[0].file_id)
FileService.update_by_id(file.id, {"name": req_doc_name})
return None
def _update_chunk_method_only(req, doc, dataset_id, tenant_id):
"""Update chunk method only (without validation)."""
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 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"])
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_error_data_result(message="Document not found!")
settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), dataset_id)
return None
def _update_document_status_only(status:int, doc, kb):
"""Update document status only (without validation)."""
if doc.status is None or (int(doc.status) != status):
try:
if not DocumentService.update_by_id(doc.id, {"status": str(status)}):
return get_error_data_result(message="Database error (Document update)!")
settings.docStoreConn.update({"doc_id": doc.id}, {"available_int": status}, search.index_name(kb.tenant_id), doc.kb_id)
except Exception as e:
return server_error_response(e)
return None
@manager.route("/datasets/<dataset_id>/documents/<document_id>", methods=["PUT"]) # noqa: F821
@token_required
async def update_doc(tenant_id, dataset_id, document_id):
@@ -237,101 +291,55 @@ async def update_doc(tenant_id, dataset_id, document_id):
type: object
"""
req = await get_request_json()
# Verify ownership and existence of dataset and document
if not KnowledgebaseService.query(id=dataset_id, tenant_id=tenant_id):
return get_error_data_result(message="You don't own the dataset.")
e, kb = KnowledgebaseService.get_by_id(dataset_id)
if not e:
return get_error_data_result(message="Can't find this dataset!")
doc = DocumentService.query(kb_id=dataset_id, id=document_id)
if not doc:
# Prepare data for validation
docs = DocumentService.query(kb_id=dataset_id, id=document_id)
if not docs:
return get_error_data_result(message="The dataset doesn't own the document.")
doc = doc[0]
if "chunk_count" in req:
if req["chunk_count"] != doc.chunk_num:
return get_error_data_result(message="Can't change `chunk_count`.")
if "token_count" in req:
if req["token_count"] != doc.token_num:
return get_error_data_result(message="Can't change `token_count`.")
if "progress" in req:
if req["progress"] != doc.progress:
return get_error_data_result(message="Can't change `progress`.")
if "meta_fields" in req:
if not isinstance(req["meta_fields"], dict):
return get_error_data_result(message="meta_fields must be a dictionary")
if not DocMetadataService.update_document_metadata(document_id, req["meta_fields"]):
# Validate document update request parameters
try:
update_doc_req = UpdateDocumentReq(**req)
except ValidationError as e:
return get_error_data_result(message=format_validation_error_message(e), code=RetCode.DATA_ERROR)
doc = docs[0]
# further check with inner status (from DB)
error_msg, error_code = _validate_document_update_fields(update_doc_req, doc, req)
if error_msg:
return get_error_data_result(message=error_msg, code=error_code)
# All validations passed, now perform all updates
# meta_fields provided, then update it
if update_doc_req.meta_fields:
if not DocMetadataService.update_document_metadata(document_id, update_doc_req.meta_fields):
return get_error_data_result(message="Failed to update metadata")
# doc name provided from request and diff with existing value, update
if "name" in req and req["name"] != doc.name:
if not isinstance(req["name"], str):
return server_error_response(AttributeError(f"'{type(req['name']).__name__}' object has no attribute 'encode'"))
if len(req["name"].encode("utf-8")) > FILE_NAME_LEN_LIMIT:
return get_result(
message=f"File name must be {FILE_NAME_LEN_LIMIT} bytes or less.",
code=RetCode.ARGUMENT_ERROR,
)
if pathlib.Path(req["name"].lower()).suffix != pathlib.Path(doc.name.lower()).suffix:
return get_result(
message="The extension of file can't be changed",
code=RetCode.ARGUMENT_ERROR,
)
for d in DocumentService.query(name=req["name"], kb_id=doc.kb_id):
if d.name == req["name"]:
return get_error_data_result(message="Duplicated document name in the same dataset.")
if not DocumentService.update_by_id(document_id, {"name": req["name"]}):
return get_error_data_result(message="Database error (Document rename)!")
if error := _update_document_name_only(document_id, req["name"]):
return error
informs = File2DocumentService.get_by_document_id(document_id)
if informs:
e, file = FileService.get_by_id(informs[0].file_id)
FileService.update_by_id(file.id, {"name": req["name"]})
if "parser_config" in req:
# parser config provided (already validated in UpdateDocumentReq), update it
if update_doc_req.parser_config:
DocumentService.update_parser_config(doc.id, req["parser_config"])
if "chunk_method" in req:
valid_chunk_method = {"naive", "manual", "qa", "table", "paper", "book", "laws", "presentation", "picture", "one", "knowledge_graph", "email", "tag"}
if req.get("chunk_method") not in valid_chunk_method:
return get_error_data_result(f"`chunk_method` {req['chunk_method']} doesn't exist")
if doc.type == FileType.VISUAL or re.search(r"\.(ppt|pptx|pages)$", doc.name):
return get_error_data_result(message="Not supported yet!")
# 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):
return error
if doc.parser_id.lower() != req["chunk_method"].lower():
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 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"])
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_error_data_result(message="Document not found!")
settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), dataset_id)
if "enabled" in req:
status = int(req["enabled"])
if doc.status != req["enabled"]:
try:
if not DocumentService.update_by_id(doc.id, {"status": str(status)}):
return get_error_data_result(message="Database error (Document update)!")
settings.docStoreConn.update({"doc_id": doc.id}, {"available_int": status}, search.index_name(kb.tenant_id), doc.kb_id)
except Exception as e:
return server_error_response(e)
if "enabled" in req: # already checked in UpdateDocumentReq - it's int if it's 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
try:
ok, doc = DocumentService.get_by_id(doc.id)
@@ -363,6 +371,27 @@ async def update_doc(tenant_id, dataset_id, document_id):
return get_result(data=renamed_doc)
def _validate_document_update_fields(update_doc_req:UpdateDocumentReq, doc, req):
"""Validate document update fields in a single method."""
# Validate immutable fields
error_msg, error_code = validation_utils.validate_immutable_fields(update_doc_req, doc)
if error_msg:
return error_msg, error_code
# Validate document name if present
if "name" in req and req["name"] != doc.name:
docs_from_name = DocumentService.query(name=req["name"], kb_id=doc.kb_id)
error_msg, error_code = validation_utils.validate_document_name(req["name"], doc, docs_from_name)
if error_msg:
return error_msg, error_code
# Validate chunk method if present
if "chunk_method" in req:
error_msg, error_code = validation_utils.validate_chunk_method(doc, req["chunk_method"])
if error_msg:
return error_msg, error_code
return None, None
@manager.route("/datasets/<dataset_id>/documents/<document_id>", methods=["GET"]) # noqa: F821
@token_required

78
api/utils/validation_utils.py
View File

@@ -13,6 +13,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
import math
import pathlib
import re
from collections import Counter
import string
from typing import Annotated, Any, Literal
@@ -32,7 +35,9 @@ from pydantic import (
from pydantic_core import PydanticCustomError
from werkzeug.exceptions import BadRequest, UnsupportedMediaType
from api.constants import DATASET_NAME_LIMIT
from api.constants import DATASET_NAME_LIMIT, FILE_NAME_LEN_LIMIT
from api.db import FileType
from common.constants import RetCode
async def validate_and_parse_json_request(
@@ -390,6 +395,36 @@ class ParserConfig(Base):
pages: Annotated[list[list[int]] | None, Field(default=None)]
ext: Annotated[dict, Field(default={})]
class UpdateDocumentReq(Base):
model_config = ConfigDict(extra='ignore')
chunk_method: 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)]
progress: Annotated[float | None, Field(default=None, ge=0.0, le=1.0)]
parser_config: Annotated[ParserConfig | None, Field(default=None)]
meta_fields: Annotated[dict | None, Field(default={})]
@field_validator("chunk_method", mode="after")
@classmethod
def validate_document_chunk_method(cls, chunk_method: str | None):
if chunk_method:
# Validate chunk method if present
valid_chunk_method = {"naive", "manual", "qa", "table", "paper", "book", "laws", "presentation", "picture", "one", "knowledge_graph", "email", "tag"}
if chunk_method not in valid_chunk_method:
raise PydanticCustomError("format_invalid", "`chunk_method` {chunk_method} doesn't exist", {"chunk_method":chunk_method})
return chunk_method
@field_validator("enabled", mode="after")
@classmethod
def validate_document_enabled(cls, enabled: str | None):
if enabled:
converted = int(enabled)
if converted < 0 or converted > 1:
raise PydanticCustomError("format_invalid", "`enabled` value invalid, only accept 0 or 1 but is {enabled}", {"enabled":enabled})
return enabled
class CreateDatasetReq(Base):
name: Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=DATASET_NAME_LIMIT), Field(...)]
@@ -810,3 +845,44 @@ class ListFileReq(BaseModel):
page_size: Annotated[int, Field(default=15, ge=1, le=100)]
orderby: Annotated[str, Field(default="create_time")]
desc: Annotated[bool, Field(default=True)]
def validate_immutable_fields(update_doc_req:UpdateDocumentReq, doc):
"""Validate that immutable fields have not been changed."""
if update_doc_req.chunk_count and update_doc_req.chunk_count != int(getattr(doc, "chunk_num", -1)):
return "Can't change `chunk_count`.", RetCode.DATA_ERROR
if update_doc_req.token_count and update_doc_req.token_count != int(getattr(doc, "token_num", -1)):
return "Can't change `token_count`.", RetCode.DATA_ERROR
if update_doc_req.progress:
progress_from_db = float(getattr(doc, "progress", -1.0))
# should not use "==" to compare two float values
if not math.isclose(update_doc_req.progress, progress_from_db):
return "Can't change `progress`.", RetCode.DATA_ERROR
return None, None
def validate_document_name(req_doc_name:str, doc, docs_from_name):
"""Validate document name update."""
if not isinstance(req_doc_name, str):
return f"AttributeError('{type(req_doc_name).__name__}' object has no attribute 'encode')", RetCode.EXCEPTION_ERROR
if len(req_doc_name.encode("utf-8")) > FILE_NAME_LEN_LIMIT:
return f"File name must be {FILE_NAME_LEN_LIMIT} bytes or less.", RetCode.ARGUMENT_ERROR
if pathlib.Path(req_doc_name.lower()).suffix != pathlib.Path(doc.name.lower()).suffix:
return "The extension of file can't be changed", RetCode.ARGUMENT_ERROR
for d in docs_from_name:
if d.name == req_doc_name:
return "Duplicated document name in the same dataset.", RetCode.DATA_ERROR
return None, None
def validate_chunk_method(doc, chunk_method=None):
"""Validate chunk method update."""
if chunk_method is not None and len(chunk_method) == 0: # will not be detected in UpdateDocumentReq
return "`chunk_method` (empty string) is not valid", RetCode.DATA_ERROR
if doc.type == FileType.VISUAL or re.search(r"\.(ppt|pptx|pages)$", doc.name):
return "Not supported yet!", RetCode.DATA_ERROR
return None, None