import uvicorn from fastapi import FastAPI, Body from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse, PlainTextResponse import socket from openai import OpenAI import markdown_to_json import json import asyncio # 阿里云中用来调用 deepseek v3 的密钥 MODEL_API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc" #MODEL_NAME = "deepseek-v3" MODEL_NAME = "qwen-plus" # 初始化 OpenAI 客户端 client = OpenAI( api_key=MODEL_API_KEY, base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", ) import re import re def markdown_to_dict(markdown_content): """ 将 Markdown 内容转换为 Python 字典，并处理【】中的文字 """ # 将 Markdown 转换为 JSON 字符串 json_content = markdown_to_json.jsonify(markdown_content) # 解码 Unicode 转义 json_content = json_content.encode('utf-8').decode('unicode_escape') # 将 JSON 字符串转换为字典 json_dict = json.loads(json_content) # 提取【】中的文字 def extract_brackets(text): match = re.search(r'【(.*?)】', text) return match.group(1) if match else None # 递归处理字典，提取并赋值 title 和 text 属性 def process_dict(d): if isinstance(d, dict): for key, value in d.items(): if isinstance(value, str): # 提取【】中的文字作为 text text = extract_brackets(value) if text: # title 是【】左侧的文字 d[key] = value.split("【")[0].strip() # text 是【】中的内容 d[f"{key}_text"] = text elif isinstance(value, dict): process_dict(value) elif isinstance(value, list): for item in value: process_dict(item) elif isinstance(d, list): for item in d: process_dict(item) process_dict(json_dict) return json_dict def extract_level1(json_dict): """ 提取一级目录，生成指定格式的 JSON 对象列表 """ # 获取第一个一级目录的名称 level1_title = next(iter(json_dict.keys()), None) if level1_title: # 提取 text 属性 level1_text = json_dict.get(f"{level1_title}_text", "") return [{"type": "cover", "data": {"title": level1_title, "text": level1_text}}] return [] def extract_level2_and_level3(json_dict, level1_title=None): """ 提取指定一级目录下的二级目录及其三级目录内容，生成指定格式的 JSON 对象列表 """ # 如果没有指定一级目录，则使用第一个一级目录 if level1_title is None: level1_title = next(iter(json_dict.keys()), None) if level1_title and level1_title in json_dict: result = [] for level2_title, level2_content in json_dict[level1_title].items(): # 提取二级目录的 text 属性 level2_text = json_dict[level1_title].get(f"{level2_title}_text", "") result.append({"type": "transition", "data": {"title": level2_title, "text": level2_text}}) # 输出三级目录内容 if isinstance(level2_content, dict): for level3_title, level3_items in level2_content.items(): # 提取三级目录的 text 属性 level3_text = level2_content.get(f"{level3_title}_text", "") # 确保 level3_items 是列表 if isinstance(level3_items, list): items = [{"title": item, "text": item} for item in level3_items] else: items = [{"title": str(level3_items), "text": str(level3_items)}] result.append({ "type": "content", "data": { "title": level3_title, "text": level3_text, "items": items } }) return result return [] def extract_contents(json_dict, level1_title=None): """ 提取所有二级目录名称，生成目录部分的 JSON 对象 """ # 如果没有指定一级目录，则使用第一个一级目录 if level1_title is None: level1_title = next(iter(json_dict.keys()), None) if level1_title and level1_title in json_dict: # 获取所有二级目录名称 level2_titles = list(json_dict[level1_title].keys()) return {"type": "contents", "data": {"items": level2_titles}} return {"type": "contents", "data": {"items": []}} async def ConvertMarkdownToJson(markdown_content): """ 生成一个 AsyncIterable，逐行返回 JSON 字符串 """ # 将 Markdown 转换为字典 json_dict = markdown_to_dict(markdown_content) # 提取一级目录 level1_json = extract_level1(json_dict) print(level1_json) for item in level1_json: yield json.dumps(item, ensure_ascii=False) await asyncio.sleep(0.5) # 控制逐行输出的速度 # 生成目录部分 contents_json = extract_contents(json_dict) yield json.dumps(contents_json, ensure_ascii=False) await asyncio.sleep(0.5) # 提取二级目录及其三级目录内容 level2_and_level3_json = extract_level2_and_level3(json_dict) for item in level2_and_level3_json: print(item) yield json.dumps(item, ensure_ascii=False) await asyncio.sleep(0.5) # 添加结束标记 yield '{"type": "end" }' # 获取本机所有 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 # 流式生成数据的函数 async def generate_stream_markdown(course_name: str): """ 流式生成 Markdown 数据，并在控制台输出完整的 Markdown 内容 """ # 调用阿里云 API，启用流式响应 stream = client.chat.completions.create( model=MODEL_NAME, messages=[ {'role': 'system', 'content': '你是一个教学经验丰富的基础教育教师'}, {'role': 'user', 'content': '帮我设计一下' + course_name + '的课件提纲，用markdown格式返回。强调1、标签只能返回 #,##,###,-,其它标签一率不可以返回，这个非常重要！2、不要返回 ```markdown 或者 ``` 这样的内容！ 3、每部分都有生成完整的一、二、三级内容，不能省略。4、必须做到：一级、二级、三级都要同步生成一个对于当前部分的一句话描述，尽量简短，不要出现标点符号,与当前级别换行后输出，放到【】这两个符号内部。'} ], stream=True, # 启用流式响应 timeout=6000, ) # 初始化完整的 Markdown 内容 full_markdown = "" # 逐字返回数据 for chunk in stream: if chunk.choices[0].delta.content: chunk_content = chunk.choices[0].delta.content full_markdown += chunk_content # 拼接 Markdown 内容 for char in chunk_content: yield char.encode("utf-8") await asyncio.sleep(0.05) # 控制逐字输出的速度 # 在控制台输出完整的 Markdown 内容 print("\n完整的 Markdown 内容：") print(full_markdown) app = FastAPI() # 添加 CORS 中间件 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 根路由，返回提示信息 @app.get("/") def root(): return PlainTextResponse("Hello ApiStream") @app.post("/api/tools/aippt_outline") # 仅支持 POST 方法 async def aippt_outline( course_name: str = Body(..., embed=True, description="课程名称") # 从请求体中获取 course_name ): # 返回流式响应 return StreamingResponse( generate_stream_markdown(course_name), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "Access-Control-Allow-Origin": "*", "X-Accel-Buffering": "no" } ) @app.post("/api/tools/aippt") # 修改为 POST 方法 async def aippt(content: str = Body(..., embed=True, description="Markdown 内容")): # 使用 Body 接收请求体参数 return StreamingResponse( ConvertMarkdownToJson(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__": # 获取本机所有 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)