Spaces:
Runtime error
Runtime error
File size: 7,262 Bytes
ac251c6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 | """
队列管理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(),
}
},
)
|