dsProject/dsLightRag/Routes/Ggb.py

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())