main
HuangHai 4 months ago
parent 97d8e70b54
commit b370f7a57f

@ -68,3 +68,12 @@ def update_question_by_id(db: PostgreSQLUtil, question_id: str, **kwargs):
except Exception as e:
print(f"更新失败: {e}")
return False
# 根据 问题id 查询 sql
def get_question_sql_by_id(db, question_id: str):
select_sql = """
select * from t_bi_question where id=%s
"""
_data = db.execute_query(select_sql, (question_id,))
sql = _data[0]['sql']
return sql

@ -2,8 +2,9 @@ import json
import uuid
import uvicorn # 导入 uvicorn
from fastapi import FastAPI, Depends, Form
from fastapi import FastAPI, Depends, Form, Query
from openai import OpenAI
from starlette.responses import StreamingResponse
from starlette.staticfiles import StaticFiles
from Config import MODEL_API_KEY, MODEL_API_URL, MODEL_NAME
@ -65,16 +66,26 @@ def get_excel(question_id: str = Form(...), question_str: str = Form(...), db: P
return {"success": True, "message": "Excel文件生成成功", "download_url": f"/static/{uuid_str}.xlsx"}
# 获取docx
# http://10.10.21.20:8000/questions/get_docx
@app.post("/questions/get_docx")
def get_docx(question_id: str = Form(...), db: PostgreSQLUtil = Depends(get_db)):
select_sql = """
select * from t_bi_question where id=%s
"""
_data = db.execute_query(select_sql, (question_id,))
sql = _data[0]['sql']
# http://10.10.21.20:8000/questions/get_docx?question_id_get=af15d834-e7f5-46b4-a0f6-15f1f888f443
@app.api_route("/questions/get_docx", methods=["POST", "GET"])
async def get_docx(
question_id: str = Form(None, description="问题IDPOST请求"), # POST 请求参数
question_id_get: str = Query(None, description="问题IDGET请求"), # GET 请求参数
db: PostgreSQLUtil = Depends(get_db)
):
# 根据请求方式获取 question_id
if question_id is not None: # POST 请求
question_id = question_id
elif question_id_get is not None: # GET 请求
question_id = question_id_get
else:
return {"success": False, "message": "缺少问题ID参数"}
# 根据问题ID获取查询sql
sql = get_question_sql_by_id(db, question_id)
# 4、生成word报告
prompt = '''
请根据以下 JSON 数据整理出2000字左右的话描述当前数据情况要求
@ -105,26 +116,56 @@ def get_docx(question_id: str = Form(...), db: PostgreSQLUtil = Depends(get_db))
stream=True # 启用流式输出
)
# 初始化变量用于存储流式输出的内容
summary = ""
# 处理流式输出
for chunk in response:
if chunk.choices[0].delta.content: # 检查是否有内容
chunk_content = chunk.choices[0].delta.content
print(chunk_content, end="", flush=True) # 实时打印到控制台
summary += chunk_content # 将内容拼接到 summary 中
# 最终 summary 为完整的 Markdown 内容
print("\n\n流式输出完成summary 已拼接为完整字符串。")
# 生成 Word 文档
# 生成 Word 文档的文件名
uuid_str = str(uuid.uuid4())
filename = f"static/{uuid_str}.docx"
markdown_to_docx(summary, output_file=filename)
# 返回静态文件URL
return {"success": True, "message": "Word文件生成成功", "download_url": f"/static/{uuid_str}.docx"}
# 定义一个生成器函数,用于逐字返回流式结果
async def generate_stream():
summary = ""
try:
for chunk in response:
if chunk.choices[0].delta.content: # 检查是否有内容
chunk_content = chunk.choices[0].delta.content
# 逐字拆分并返回
for char in chunk_content:
print(char, end="", flush=True) # 逐字输出到控制台
yield char.encode("utf-8") # 将字符编码为 UTF-8 字节
summary += char # 将内容拼接到 summary 中
# 流式传输完成后,生成 Word 文档
markdown_to_docx(summary, output_file=filename)
# 返回最终的 JSON 数据
final_response = json.dumps({
"success": True,
"message": "Word文件生成成功",
"download_url": f"/static/{uuid_str}.docx"
})
print(final_response) # 输出最终 JSON 到控制台
yield final_response.encode("utf-8") # 将 JSON 数据编码为 UTF-8 字节
except Exception as e:
# 如果发生异常,返回错误信息
error_response = json.dumps({
"success": False,
"message": f"生成Word文件失败: {str(e)}"
})
print(error_response) # 输出错误信息到控制台
yield error_response.encode("utf-8") # 将错误信息编码为 UTF-8 字节
finally:
# 确保资源释放
if "response" in locals():
response.close()
# 使用 StreamingResponse 返回流式结果
return StreamingResponse(
generate_stream(),
media_type="text/plain; charset=utf-8", # 明确指定字符编码为 UTF-8
headers={
"Cache-Control": "no-cache", # 禁用缓存
"Content-Type": "text/plain; charset=utf-8", # 设置内容类型和字符编码
}
)
# 确保直接运行脚本时启动 FastAPI 应用
if __name__ == "__main__":

Loading…
Cancel
Save