Refactor: Consolidation WEB API & HTTP API for document delete api (#14254)

### What problem does this PR solve?

Before consolidation
Web API: POST /v1/document/rm
Http API - DELETE /api/v1/datasets/<dataset_id>/documents

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

### Type of change

- [x] Refactoring

api/apps/restful_apis/document_api.py

@@ -27,14 +27,17 @@ from api.constants import IMG_BASE64_PREFIX
 from api.db import VALID_FILE_TYPES
 from api.db.services.doc_metadata_service import DocMetadataService
 from api.db.services.document_service import DocumentService
+from api.db.services.file_service import FileService
 from api.db.services.knowledgebase_service import KnowledgebaseService
+from api.common.check_team_permission import check_kb_team_permission
 from api.utils.api_utils import get_data_error_result, get_error_data_result, get_result, get_json_result, \
-    server_error_response, add_tenant_id_to_kwargs, get_request_json
+    server_error_response, add_tenant_id_to_kwargs, get_request_json, get_error_argument_result, check_duplicate_ids
 from api.utils.validation_utils import (
-    UpdateDocumentReq, format_validation_error_message,
+    UpdateDocumentReq, format_validation_error_message, validate_and_parse_json_request, DeleteDocumentReq,
 )
 from common.constants import RetCode
 from common.metadata_utils import convert_conditions, meta_filter, turn2jsonschema
+from common.misc_utils import thread_pool_exec
 
 @manager.route("/datasets/<dataset_id>/documents/<document_id>", methods=["PATCH"]) # noqa: F821
 @login_required
@@ -260,9 +263,7 @@ async def upload_document(dataset_id, tenant_id):
                     description: Processing status.
     """
     from api.constants import FILE_NAME_LEN_LIMIT
-    from api.common.check_team_permission import check_kb_team_permission
     from api.db.services.file_service import FileService
-    from common.misc_utils import thread_pool_exec
     
     form = await request.form
     files = await request.files
@@ -674,6 +675,89 @@ def _parse_doc_id_filter_with_metadata(req, kb_id):
     return RetCode.SUCCESS, "", list(doc_ids_filter) if doc_ids_filter is not None else [], return_empty_metadata
 
 
+@manager.route("/datasets/<dataset_id>/documents", methods=["DELETE"])  # noqa: F821
+@login_required
+@add_tenant_id_to_kwargs
+async def delete_documents(tenant_id, dataset_id):
+    """
+    Delete documents from a dataset.
+    ---
+    tags:
+      - Documents
+    security:
+      - ApiKeyAuth: []
+    parameters:
+      - in: path
+        name: dataset_id
+        type: string
+        required: true
+        description: ID of the dataset containing the documents.
+      - in: header
+        name: Authorization
+        type: string
+        required: true
+        description: Bearer token for authentication.
+      - in: body
+        name: body
+        description: Document deletion parameters.
+        required: true
+        schema:
+          type: object
+          properties:
+            ids:
+              type: array or null
+              items:
+                type: string
+              description: |
+                Specifies the documents to delete:
+                - An array of IDs, only the specified documents will be deleted.
+            delete_all:
+              type: boolean
+              default: false
+              description: Whether to delete all documents in the dataset.
+    responses:
+      200:
+        description: Successful operation.
+        schema:
+          type: object
+    """
+    req, err = await validate_and_parse_json_request(request, DeleteDocumentReq)
+    if err is not None or req is None:
+        return get_error_argument_result(err)
+
+    try:
+        # Validate dataset exists and user has permission
+        if not KnowledgebaseService.accessible(kb_id=dataset_id, user_id=tenant_id):
+            return get_error_data_result(message=f"You don't own the dataset {dataset_id}. ")
+
+        # Get documents to delete
+        doc_ids = req.get("ids") or []
+        delete_all = req.get("delete_all", False)
+        if not delete_all and len(doc_ids) == 0:
+            return get_error_data_result(message=f"should either provide doc ids or set delete_all(true), dataset: {dataset_id}. ")
+
+        if len(doc_ids) > 0 and delete_all:
+            return get_error_data_result(message=f"should not provide both doc ids and delete_all(true), dataset: {dataset_id}. ")
+        if delete_all:
+            doc_ids = [doc.id for doc in DocumentService.query(kb_id=dataset_id)]
+
+        # make sure each id is unique
+        unique_doc_ids, duplicate_messages = check_duplicate_ids(doc_ids, "document")
+        if duplicate_messages:
+            logging.warning(f"duplicate_messages:{duplicate_messages}")
+        else:
+            doc_ids = unique_doc_ids
+
+        # Delete documents using existing FileService.delete_docs
+        errors = await thread_pool_exec(FileService.delete_docs, doc_ids, tenant_id)
+
+        if errors:
+            return get_error_data_result(message=str(errors))
+
+        return get_result(data={"deleted": len(doc_ids)})
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message="Internal server error")
 def _aggregate_filters(docs):
     """Aggregate filter options from a list of documents.