QingLong/AI/AiService/CallDeepSeekStream.py

from fastapi import FastAPI, Query, Body
from fastapi.responses import StreamingResponse, PlainTextResponse
from fastapi.middleware.cors import CORSMiddleware
import socket
from openai import OpenAI
import asyncio
import json

app = FastAPI()

# 添加 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 阿里云中用来调用 deepseek v3 的密钥
MODEL_API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc"
MODEL_NAME = "deepseek-v3"

# 初始化 OpenAI 客户端
client = OpenAI(
    api_key=MODEL_API_KEY,
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 获取本机所有 IPv4 地址
def get_local_ips():
    ips = []
    hostname = socket.gethostname()
    try:
        # 获取所有 IP 地址
        addrs = socket.getaddrinfo(hostname, None, family=socket.AF_INET)  # 只获取 IPv4 地址
        for addr in addrs:
            ip = addr[4][0]
            if ip not in ips:
                ips.append(ip)
    except Exception as e:
        print(f"获取 IP 地址失败: {e}")
    return ips


# 解析 Markdown 内容并生成 JSON 数据
def parse_markdown_to_json(content: str):
    sections = []
    current_section = None
    current_subsection = None

    for line in content.split("\n"):
        # 忽略无用的标记
        if line.strip() in ["```markdown", "```"]:
            continue
        if line.startswith("# "):  # 一级标题（忽略，作为整体标题）
            continue
        elif line.startswith("## "):  # 二级标题
            if current_section:
                sections.append(current_section)
            current_section = {"title": line[3:].strip(), "items": []}
            current_subsection = None
        elif line.startswith("### "):  # 三级标题
            if current_section:
                current_subsection = {"title": line[4:].strip(), "text": ""}
                current_section["items"].append(current_subsection)
        elif line.strip() and (line.startswith("- ") or (line[0].isdigit() and line[1] == '.')):  # 列表项或数字加点
            if not current_subsection:  # 如果 current_subsection 为 None，则创建一个默认子章节
                current_subsection = {"title": "", "text": ""}
                if current_section:
                    current_section["items"].append(current_subsection)
            current_subsection["text"] += line.strip() + " "
        elif line.strip():  # 普通文本
            if current_subsection:
                current_subsection["text"] += line.strip() + " "

    if current_section:
        sections.append(current_section)
    return sections

# 生成流式 JSON 数据（中文输出，逐行返回）


# 根路由，返回提示信息
@app.get("/")
def root():
    return PlainTextResponse("Hello ApiStream")

# 流式返回数据（支持 GET 和 POST 方法）
@app.api_route("/api/tools/aippt_outline", methods=["GET", "POST"])
async def aippt_outline(
    course_name: str = Query(None, description="课程名称（GET 方法使用）"),  # 从查询参数中获取 course_name
    course_name_body: str = Body(None, embed=True, description="课程名称（POST 方法使用）")  # 从请求体中获取 course_name
):
    # 检查 course_name 是否为空
    if not course_name and not course_name_body:
        return PlainTextResponse("请提供课程名称,例：course_name=三角形面积")

    # 优先使用 POST 请求体中的 course_name
    course_name = course_name_body if course_name_body else course_name

    # 返回流式响应
    return StreamingResponse(
        generate_stream(course_name),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "Access-Control-Allow-Origin": "*",
            "X-Accel-Buffering": "no"
        }
    )

# 流式生成数据的函数
async def generate_stream(course_name: str):
    # 调用阿里云 API，启用流式响应
    stream = client.chat.completions.create(
        model=MODEL_NAME,
        messages=[
            {'role': 'system', 'content': '你是一个教学经验丰富的基础教育教师'},
            {'role': 'user', 'content': '帮我设计一下' + course_name + '的课件提纲，用markdown格式返回。不要返回 ```markdown 或者 ```  这样的内容！'}
        ],
        stream=True,  # 启用流式响应
        timeout=6000,
    )

    # 逐字返回数据
    for chunk in stream:
        if chunk.choices[0].delta.content:
            for char in chunk.choices[0].delta.content:
                yield char.encode("utf-8")
                await asyncio.sleep(0.05)  # 控制逐字输出的速度

# 生成流式 JSON 数据（中文输出，逐行返回）
async def generate_json_stream(content: str):  # 新增 content 参数
    # 解析 Markdown 内容
    sections = parse_markdown_to_json(content)

    # 封面（使用第一个章节的标题作为封面标题）
    first_section_title = sections[0]["title"] if sections else "默认标题"
    yield json.dumps({"type": "cover", "data": {"title": first_section_title, "text": "课程简介"}}, ensure_ascii=False) + "\n"
    await asyncio.sleep(0.5)

    # 目录（使用所有章节的标题）
    yield json.dumps({"type": "contents", "data": {"items": [section["title"] for section in sections]}}, ensure_ascii=False) + "\n"
    await asyncio.sleep(0.5)

    # 逐章节生成内容
    for section in sections:
        # 过渡（使用章节标题动态生成过渡文本）
        yield json.dumps({"type": "transition", "data": {"title": section["title"], "text": section["title"]}}, ensure_ascii=False) + "\n"
        await asyncio.sleep(0.5)

        # 具体内容（直接使用解析后的章节内容）
        yield json.dumps({"type": "content", "data": {"title": section["title"], "items": section["items"]}}, ensure_ascii=False) + "\n"
        await asyncio.sleep(0.5)

    # 结束
    yield json.dumps({"type": "end"}, ensure_ascii=False) + "\n"

# 从文件中读取 Markdown 并生成流式 JSON（中文输出，逐行返回）
@app.get("/api/tools/aippt")
async def aippt(content: str = Query(..., description="Markdown 内容")):  # 新增 content 参数
    return StreamingResponse(
        generate_json_stream(content),  # 传入 content
        media_type="text/plain",  # 使用 text/plain 格式
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "Access-Control-Allow-Origin": "*",
            "X-Accel-Buffering": "no",
            "Content-Type": "text/plain; charset=utf-8"  # 明确设置 Content-Type
        }
    )

# 运行应用
if __name__ == "__main__":
    import uvicorn

    # 获取本机所有 IPv4 地址
    ips = get_local_ips()
    if not ips:
        print("无法获取本机 IP 地址，使用默认地址 127.0.0.1")
        ips = ["127.0.0.1"]

    # 打印所有 IP 地址
    print("服务将在以下 IP 地址上运行：")
    for ip in ips:
        print(f"http://{ip}:5173")

    # 启动 FastAPI 应用，绑定到所有 IP 地址
    uvicorn.run(app, host="0.0.0.0", port=5173)