diff --git a/AI/AiService/CallDeepSeekStream.py b/AI/AiService/CallDeepSeekStream.py index f491d850..905a4d99 100644 --- a/AI/AiService/CallDeepSeekStream.py +++ b/AI/AiService/CallDeepSeekStream.py @@ -2,22 +2,118 @@ import uvicorn from fastapi import FastAPI, Body from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse, PlainTextResponse -import asyncio import socket from openai import OpenAI -from MarkdownToJsonUtil import * +import markdown_to_json +import json +import asyncio # 阿里云中用来调用 deepseek v3 的密钥 MODEL_API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc" -MODEL_NAME = "deepseek-v3" -#MODEL_NAME = "qwen-plus" +#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", ) +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 字符串转换为字典 + return json.loads(json_content) + + +def extract_level1(json_dict): + """ + 提取一级目录，生成指定格式的 JSON 对象列表 + """ + # 获取第一个一级目录的名称 + level1_title = next(iter(json_dict.keys()), None) + if level1_title: + return [{"type": "cover", "data": {"title": level1_title, "text": level1_title}}] + 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(): + # 输出二级目录 + result.append({"type": "transition", "data": {"title": level2_title, "text": level2_title}}) + # 输出三级目录内容 + if isinstance(level2_content, dict): + for level3_title, level3_items in level2_content.items(): + # 确保 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, + "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) + 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: + yield json.dumps(item, ensure_ascii=False) + await asyncio.sleep(0.5) + + # 添加结束标记 + yield '{"type": "end" }' # 获取本机所有 IPv4 地址 def get_local_ips(): ips = [] @@ -35,25 +131,35 @@ def get_local_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格式返回。不要返回 ```markdown 或者 ``` 这样的内容！'} + {'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: - for char in 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 中间件 @@ -89,7 +195,7 @@ async def aippt_outline( @app.post("/api/tools/aippt") # 修改为 POST 方法 async def aippt(content: str = Body(..., embed=True, description="Markdown 内容")): # 使用 Body 接收请求体参数 return StreamingResponse( - getMyJson(content), # 传入 content + ConvertMarkdownToJson(content), # 传入 content media_type="text/plain", # 使用 text/plain 格式 headers={ "Cache-Control": "no-cache", diff --git a/AI/AiService/MarkdownToJsonUtil.py b/AI/AiService/MarkdownToJsonUtil.py deleted file mode 100644 index 764eecb6..00000000 --- a/AI/AiService/MarkdownToJsonUtil.py +++ /dev/null @@ -1,97 +0,0 @@ -import markdown_to_json -import json -import asyncio - - -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 字符串转换为字典 - return json.loads(json_content) - - -def extract_level1(json_dict): - """ - 提取一级目录，生成指定格式的 JSON 对象列表 - """ - # 获取第一个一级目录的名称 - level1_title = next(iter(json_dict.keys()), None) - if level1_title: - return [{"type": "cover", "data": {"title": level1_title, "text": level1_title}}] - 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(): - # 输出二级目录 - result.append({"type": "transition", "data": {"title": level2_title, "text": level2_title}}) - # 输出三级目录内容 - if isinstance(level2_content, dict): - for level3_title, level3_items in level2_content.items(): - items = [{"title": item, "text": item} for item in level3_items] - result.append({ - "type": "content", - "data": { - "title": level3_title, - "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 getMyJson(markdown_content): - """ - 生成一个 AsyncIterable，逐行返回 JSON 字符串 - """ - # 将 Markdown 转换为字典 - json_dict = markdown_to_dict(markdown_content) - - # 提取一级目录 - level1_json = extract_level1(json_dict) - 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: - yield json.dumps(item, ensure_ascii=False) - await asyncio.sleep(0.5) - - # 添加结束标记 - yield '{"type": "end" }' \ No newline at end of file diff --git a/AI/AiService/__pycache__/MarkdownToJsonUtil.cpython-310.pyc b/AI/AiService/__pycache__/MarkdownToJsonUtil.cpython-310.pyc deleted file mode 100644 index 3bfec443..00000000 Binary files a/AI/AiService/__pycache__/MarkdownToJsonUtil.cpython-310.pyc and /dev/null differ