119 lines
4.8 KiB
Python
119 lines
4.8 KiB
Python
import json
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
|
||
from fastapi import APIRouter, HTTPException, Request
|
||
from sse_starlette import EventSourceResponse
|
||
|
||
from Dsideal.Math.ShiTuGetGgb import process_geometry_image
|
||
from Util.GGBUtil import getGgbCommand
|
||
from Util.GGBUtil import translate_to_ggb as translate_to_ggb_util
|
||
|
||
# 创建路由对象
|
||
router = APIRouter(prefix="/api", tags=["几何图形处理"])
|
||
|
||
# 设置日志
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# 几何图形接口实现
|
||
@router.post("/get_ggb")
|
||
async def get_ggb(request: Request):
|
||
try:
|
||
data = await request.json()
|
||
image_url = data.get("image_url")
|
||
except Exception:
|
||
# 如果JSON解析失败，尝试从表单数据中获取
|
||
form = await request.form()
|
||
image_url = form.get("image_url")
|
||
|
||
# 如果两种方式都没有获取到image_url，则返回错误
|
||
if not image_url:
|
||
raise HTTPException(status_code=400, detail="缺少image_url参数")
|
||
|
||
async def generate_ggb_stream():
|
||
# 第一步：处理图片并生成几何描述
|
||
try:
|
||
# 步骤1: 识别生成描述语言
|
||
logger.info("开始处理图片...")
|
||
content='开始处理图片...\n'
|
||
yield f"{json.dumps({'step': 'image_processing', 'content': content}, ensure_ascii=False)}"
|
||
|
||
# 获取完整的几何描述
|
||
QvqResult = ""
|
||
async for chunk in process_geometry_image(image_url):
|
||
if chunk:
|
||
QvqResult += chunk
|
||
# 发送实际内容给前端
|
||
yield f"{json.dumps({'step': 'image_processing', 'content': chunk}, ensure_ascii=False)}"
|
||
# 添加控制台输出，显示发送给前端的内容
|
||
print(chunk, end="", flush=True)
|
||
# 添加小延迟，控制输出速度
|
||
await asyncio.sleep(0.1)
|
||
|
||
if not QvqResult.strip():
|
||
yield f"{json.dumps({'error': '图片处理失败'}, ensure_ascii=False)}"
|
||
return
|
||
|
||
|
||
# 第二步：根据几何描述生成GGB指令
|
||
try:
|
||
async for chunk in getGgbCommand(QvqResult):
|
||
if chunk:
|
||
yield f"{json.dumps({'step': 'ggb_generation', 'content': chunk}, ensure_ascii=False)}"
|
||
# 添加控制台输出，显示发送给前端的内容
|
||
print(chunk, end="", flush=True)
|
||
await asyncio.sleep(0.1)
|
||
|
||
except Exception as e:
|
||
logger.error(f"GGB指令生成异常: {str(e)}")
|
||
yield f"{json.dumps({'error': f'GGB指令生成异常: {str(e)}'}, ensure_ascii=False)}"
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理异常: {str(e)}")
|
||
yield f"{json.dumps({'error': f'处理异常: {str(e)}'}, ensure_ascii=False)}"
|
||
finally:
|
||
yield f"{json.dumps({'DONE': True}, ensure_ascii=False)}"
|
||
|
||
# 返回SSE响应
|
||
return EventSourceResponse(generate_ggb_stream())
|
||
|
||
@router.post("/translate_to_ggb")
|
||
async def translate_to_ggb(request: Request):
|
||
try:
|
||
# 获取请求数据
|
||
data = await request.json()
|
||
natural_language_cmd = data.get("natural_language_cmd")
|
||
except Exception:
|
||
# 如果JSON解析失败，尝试从表单数据中获取
|
||
form = await request.form()
|
||
natural_language_cmd = form.get("natural_language_cmd")
|
||
|
||
# 验证参数是否存在
|
||
if not natural_language_cmd:
|
||
raise HTTPException(status_code=400, detail="缺少natural_language_cmd参数")
|
||
|
||
async def generate_translation_stream():
|
||
try:
|
||
logger.info(f"开始翻译自然语言指令: {natural_language_cmd}")
|
||
# 发送开始翻译的通知
|
||
yield f"{json.dumps({'status': 'translating', 'content': '开始翻译指令...'}, ensure_ascii=False)}"
|
||
await asyncio.sleep(0.1)
|
||
|
||
# 调用翻译函数并流式返回结果（使用别名）
|
||
async for chunk in translate_to_ggb_util(natural_language_cmd):
|
||
if chunk:
|
||
yield f"{json.dumps({'status': 'translating', 'content': chunk}, ensure_ascii=False)}"
|
||
# 控制台输出
|
||
print(chunk, end="", flush=True)
|
||
await asyncio.sleep(0.1)
|
||
|
||
except Exception as e:
|
||
logger.error(f"翻译指令异常: {str(e)}")
|
||
yield f"{json.dumps({'error': f'翻译指令异常: {str(e)}'}, ensure_ascii=False)}"
|
||
finally:
|
||
yield f"{json.dumps({'DONE': True}, ensure_ascii=False)}"
|
||
|
||
# 返回SSE响应
|
||
return EventSourceResponse(generate_translation_stream()) |