♻️ Refactor the conversation_management_app&conversation_management_service&data_process_app&es

Author: Phinease

backend/apps/conversation_management_app.py

@@ -1,4 +1,5 @@
 import logging
+from http import HTTPStatus
 from typing import Any, Dict, Optional
 
 from fastapi import APIRouter, Header, HTTPException, Request
@@ -11,7 +12,6 @@
     OpinionRequest,
     RenameRequest,
 )
-from database.conversation_db import get_message_id_by_index
 from services.conversation_management_service import (
     create_new_conversation,
     delete_conversation_service,
@@ -20,7 +20,7 @@
     get_conversation_list_service,
     get_sources_service,
     rename_conversation_service,
-    update_message_opinion_service,
+    update_message_opinion_service, get_message_id_by_index_impl,
 )
 from utils.auth_utils import get_current_user_id, get_current_user_info
 
@@ -50,7 +50,7 @@ async def create_new_conversation_endpoint(request: ConversationRequest, authori
         return ConversationResponse(code=0, message="success", data=conversation_data)
     except Exception as e:
         logging.error(f"Failed to create conversation: {str(e)}")
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
 
 
 @router.get("/list", response_model=ConversationResponse)
@@ -67,16 +67,12 @@ async def list_conversations_endpoint(authorization: Optional[str] = Header(None
     try:
         user_id, tenant_id = get_current_user_id(authorization)
         if not user_id:
-            raise HTTPException(status_code=401, detail="未授权访问，请先登录")
-
+            raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail="Unauthorized access, Please login first")
         conversations = get_conversation_list_service(user_id)
         return ConversationResponse(code=0, message="success", data=conversations)
-    except HTTPException as he:
-        # Throw HTTP Exception Directly
-        raise he
     except Exception as e:
         logging.error(f"Failed to get conversation list: {str(e)}")
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
 
 
 @router.post("/rename", response_model=ConversationResponse)
@@ -100,9 +96,7 @@ async def rename_conversation_endpoint(request: RenameRequest, authorization: Op
         return ConversationResponse(code=0, message="success", data=True)
     except Exception as e:
         logging.error(f"Failed to rename conversation: {str(e)}")
-        if isinstance(e, HTTPException):
-            raise e
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
 
 
 @router.delete("/{conversation_id}", response_model=ConversationResponse)
@@ -123,9 +117,7 @@ async def delete_conversation_endpoint(conversation_id: int, authorization: Opti
         return ConversationResponse(code=0, message="success", data=True)
     except Exception as e:
         logging.error(f"Failed to delete conversation: {str(e)}")
-        if isinstance(e, HTTPException):
-            raise e
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
 
 
 @router.get("/{conversation_id}", response_model=ConversationResponse)
@@ -147,9 +139,7 @@ async def get_conversation_history_endpoint(conversation_id: int, authorization:
         return ConversationResponse(code=0, message="success", data=history_data)
     except Exception as e:
         logging.error(f"Failed to get conversation history: {str(e)}")
-        if isinstance(e, HTTPException):
-            raise e
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
 
 
 @router.post("/sources", response_model=Dict[str, Any])
@@ -175,11 +165,7 @@ async def get_sources_endpoint(request: Dict[str, Any], authorization: Optional[
         return get_sources_service(conversation_id, message_id, source_type, user_id)
     except Exception as e:
         logging.error(f"Failed to get message sources: {str(e)}")
-        return {
-            "code": 500,
-            "message": str(e),
-            "data": None
-        }
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
 
 
 @router.post("/generate_title", response_model=ConversationResponse)
@@ -208,9 +194,7 @@ async def generate_conversation_title_endpoint(
         return ConversationResponse(code=0, message="success", data=title)
     except Exception as e:
         logging.error(f"Failed to generate conversation title: {str(e)}")
-        if isinstance(e, HTTPException):
-            raise e
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
 
 
 @router.post("/message/update_opinion", response_model=ConversationResponse)
@@ -230,9 +214,7 @@ async def update_opinion_endpoint(request: OpinionRequest, authorization: Option
         return ConversationResponse(code=0, message="success", data=True)
     except Exception as e:
         logging.error(f"Failed to update message like/dislike: {str(e)}")
-        if isinstance(e, HTTPException):
-            raise e
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
 
 
 @router.post("/message/id", response_model=ConversationResponse)
@@ -249,14 +231,8 @@ async def get_message_id_endpoint(request: MessageIdRequest):
         ConversationResponse object containing message_id
     """
     try:
-        message_id = get_message_id_by_index(
-            request.conversation_id, request.message_index)
-        if message_id is None:
-            raise HTTPException(status_code=404, detail="Message not found")
-
+        message_id = await get_message_id_by_index_impl(request.conversation_id, request.message_index)
         return ConversationResponse(code=0, message="success", data=message_id)
     except Exception as e:
         logging.error(f"Failed to get message ID: {str(e)}")
-        if isinstance(e, HTTPException):
-            raise e
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))

backend/apps/data_process_app.py

@@ -3,6 +3,7 @@
 import logging
 import time
 from contextlib import asynccontextmanager
+from http import HTTPStatus
 from typing import Optional
 
 from fastapi import APIRouter, File, Form, Header, HTTPException, UploadFile
@@ -119,7 +120,7 @@ async def process_sync_endpoint(
     except Exception as e:
         logger.error(f"Error in synchronous processing: {str(e)}")
         raise HTTPException(
-            status_code=500,
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
             detail=f"Error processing file: {str(e)}"
         )
 
@@ -133,48 +134,13 @@ async def create_batch_tasks(request: BatchTaskRequest, authorization: Optional[
     Processing happens in the background for each file independently.
     """
     try:
-        task_ids = []
-
-        # Create individual tasks for each source
-        for source_config in request.sources:
-            # Extract parameters
-            source = source_config.get('source')
-            source_type = source_config.get('source_type')
-            chunking_strategy = source_config.get('chunking_strategy')
-            index_name = source_config.get('index_name')
-            original_filename = source_config.get('original_filename')
-
-            # Validate required fields
-            if not source:
-                logger.error(
-                    f"Missing required field 'source' in source config: {source_config}")
-                continue
-            if not index_name:
-                logger.error(
-                    f"Missing required field 'index_name' in source config: {source_config}")
-                continue
-
-            # Create individual task for this source
-            task_result = process_and_forward.delay(
-                source=source,
-                source_type=source_type,
-                chunking_strategy=chunking_strategy,
-                index_name=index_name,
-                original_filename=original_filename,
-                authorization=authorization
-            )
-
-            task_ids.append(task_result.id)
-            logger.debug(f"Created task {task_result.id} for source: {source}")
-
-        logger.info(
-            f"Created {len(task_ids)} individual tasks for batch processing")
+        task_ids = await service.create_batch_tasks_impl(authorization=authorization, request=request)
         return BatchTaskResponse(task_ids=task_ids)
 
     except Exception as e:
         logger.error(f"Error creating batch tasks: {str(e)}")
         raise HTTPException(
-            status_code=500, detail=f"Failed to create batch tasks: {str(e)}")
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=f"Failed to create batch tasks: {str(e)}")
 
 
 @router.get("/load_image")
@@ -194,7 +160,7 @@ async def load_image(url: str):
 
         if image is None:
             raise HTTPException(
-                status_code=404, detail="Failed to load image or image format not supported")
+                status_code=HTTPStatus.NOT_FOUND, detail="Failed to load image or image format not supported")
 
         # Convert PIL image to base64
         img_byte_arr = io.BytesIO()
@@ -214,7 +180,7 @@ async def load_image(url: str):
     except Exception as e:
         logger.error(f"Error loading image: {str(e)}")
         raise HTTPException(
-            status_code=500, detail=f"Error loading image: {str(e)}")
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=f"Error loading image: {str(e)}")
 
 
 @router.get("/{task_id}", response_model=SimpleTaskStatusResponse)
@@ -224,7 +190,7 @@ async def get_task(task_id: str):
 
     if not task:
         raise HTTPException(
-            status_code=404, detail=f"Task with ID {task_id} not found")
+            status_code=HTTPStatus.NOT_FOUND, detail=f"Task with ID {task_id} not found")
     return SimpleTaskStatusResponse(
         id=task["id"],
         task_name=task["task_name"],
@@ -272,7 +238,7 @@ async def get_index_tasks(index_name: str):
     try:
         return await service.get_index_tasks(index_name)
     except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
 
 
 @router.get("/{task_id}/details")
@@ -281,7 +247,7 @@ async def get_task_details(task_id: str):
     from data_process.utils import get_task_details as utils_get_task_details
     task = await utils_get_task_details(task_id)
     if not task:
-        raise HTTPException(status_code=404, detail="Task not found")
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Task not found")
     return task
 
 
@@ -307,7 +273,7 @@ async def filter_important_image(
     except Exception as e:
         logger.error(f"Error processing image: {str(e)}")
         raise HTTPException(
-            status_code=500, detail=f"Error processing image: {str(e)}")
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=f"Error processing image: {str(e)}")
 
 
 @router.post("/process_text_file", response_model=dict, status_code=200)
@@ -380,7 +346,7 @@ async def process_text_file(
         logger.exception(
             f"Error processing uploaded file {file.filename}: {str(e)}")
         raise HTTPException(
-            status_code=500,
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
             detail=f"An error occurred while processing the file: {str(e)}"
         )