""" 队列管理API路由 提供任务队列状态查询、任务取消等接口 """ import re from fastapi import APIRouter, HTTPException from datetime import datetime from app.services.task_queue import task_queue from app.services.storage_service import StorageService from app.core.logging import get_logger logger = get_logger(__name__) router = APIRouter() @router.get("/queue/status") async def get_queue_status(): """获取队列状态""" try: status = task_queue.get_queue_status() return status except Exception as e: logger.error(f"获取队列状态失败: {str(e)}") raise HTTPException( status_code=500, detail={ "error": { "code": "INTERNAL_SERVER_ERROR", "message": "服务器内部错误", "timestamp": datetime.now().isoformat(), } }, ) @router.get("/queue/status/{task_id}") async def get_task_status(task_id: str): """获取指定任务状态""" try: # 验证任务ID格式 if not re.match(r"^audio_[0-9a-f-]+$", task_id): raise HTTPException( status_code=400, detail={ "error": { "code": "INVALID_TASK_ID", "message": "任务ID格式无效", "timestamp": datetime.now().isoformat(), } }, ) # 获取任务信息 task = task_queue.get_task(task_id) if not task: raise HTTPException( status_code=404, detail={ "error": { "code": "TASK_NOT_FOUND", "message": "任务不存在", "timestamp": datetime.now().isoformat(), } }, ) return task.to_dict() except HTTPException: raise except Exception as e: logger.error(f"获取任务状态失败: {str(e)}") raise HTTPException( status_code=500, detail={ "error": { "code": "INTERNAL_SERVER_ERROR", "message": "服务器内部错误", "timestamp": datetime.now().isoformat(), } }, ) @router.get("/queue/result/{task_id}") async def get_task_result(task_id: str): """获取任务转换结果""" try: # 验证任务ID格式 if not re.match(r"^audio_[0-9a-f-]+$", task_id): raise HTTPException( status_code=400, detail={ "error": { "code": "INVALID_TASK_ID", "message": "任务ID格式无效", "timestamp": datetime.now().isoformat(), } }, ) # 获取任务信息 task = task_queue.get_task(task_id) if not task: raise HTTPException( status_code=404, detail={ "error": { "code": "TASK_NOT_FOUND", "message": "任务不存在", "timestamp": datetime.now().isoformat(), } }, ) # 检查任务状态 from app.models.task import TaskStatus if task.status != TaskStatus.COMPLETED: raise HTTPException( status_code=400, detail={ "error": { "code": "TASK_NOT_COMPLETED", "message": f"任务尚未完成,当前状态: {task.status.value}", "timestamp": datetime.now().isoformat(), } }, ) # 返回完整的任务结果 result = task.to_dict() result.update( { "original_filename": task.original_filename, "audio_files": task.audio_files, "metadata": task.metadata, } ) return result except HTTPException: raise except Exception as e: logger.error(f"获取任务结果失败: {str(e)}") raise HTTPException( status_code=500, detail={ "error": { "code": "INTERNAL_SERVER_ERROR", "message": "服务器内部错误", "timestamp": datetime.now().isoformat(), } }, ) @router.delete("/queue/cancel/{task_id}") async def cancel_task(task_id: str): """取消任务""" try: # 验证任务ID格式 if not re.match(r"^audio_[0-9a-f-]+$", task_id): raise HTTPException( status_code=400, detail={ "error": { "code": "INVALID_TASK_ID", "message": "任务ID格式无效", "timestamp": datetime.now().isoformat(), } }, ) # 获取任务信息以便获取用户ID task = task_queue.get_task(task_id) if not task: raise HTTPException( status_code=404, detail={ "error": { "code": "TASK_NOT_FOUND", "message": "任务不存在", "timestamp": datetime.now().isoformat(), } }, ) # 取消任务 success = task_queue.cancel_task(task_id) if not success: raise HTTPException( status_code=400, detail={ "error": { "code": "TASK_CANNOT_BE_CANCELLED", "message": "任务无法取消,可能已在处理中或已完成", "timestamp": datetime.now().isoformat(), } }, ) # 如果任务已经有部分文件生成,尝试清理 if hasattr(task, 'user_id') and task.user_id: try: storage_service = StorageService() await storage_service.delete_task_files(task_id, task.user_id) logger.info(f"已清理取消任务的文件: {task_id}") except Exception as e: logger.warning(f"清理取消任务文件失败: {str(e)}") return { "message": "任务已取消", "task_id": task_id, "timestamp": datetime.now().isoformat(), } except HTTPException: raise except Exception as e: logger.error(f"取消任务失败: {str(e)}") raise HTTPException( status_code=500, detail={ "error": { "code": "INTERNAL_SERVER_ERROR", "message": "服务器内部错误", "timestamp": datetime.now().isoformat(), } }, )