|
|
import re
|
|
|
|
|
|
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 = "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": ""}}]
|
|
|
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 = []
|
|
|
for item in level3_items:
|
|
|
# 如果 item 是字符串，直接作为 title 和 text
|
|
|
if isinstance(item, str):
|
|
|
items.append({"title": item, "text": item})
|
|
|
# 如果 item 是字典，提取 title 和 text
|
|
|
elif isinstance(item, dict):
|
|
|
items.append({
|
|
|
"title": item.get("title", ""),
|
|
|
"text": item.get("text", "")
|
|
|
})
|
|
|
else:
|
|
|
# 如果 level3_items 不是列表，直接作为 title 和 text
|
|
|
items = [{"title": str(level3_items), "text": str(level3_items)}]
|
|
|
result.append({
|
|
|
"type": "content",
|
|
|
"data": {
|
|
|
"title": level3_title,
|
|
|
"text": 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": []}}
|
|
|
|
|
|
|
|
|
def expand_text_with_ai(json_dict):
|
|
|
"""
|
|
|
调用 AI 扩写每个有 title 属性的节点，生成一句话描述并放到 text 属性上
|
|
|
"""
|
|
|
|
|
|
# 递归处理字典
|
|
|
def process_dict(d):
|
|
|
if isinstance(d, dict):
|
|
|
for key, value in d.items():
|
|
|
if key == "title" and isinstance(value, str):
|
|
|
# 基于全文和当前 title 生成一句话描述
|
|
|
prompt = f"基于以下内容，为标题 '{value}' 生成一句话描述，言简意赅，不要使用标点符号：\n{json.dumps(json_dict, ensure_ascii=False)}"
|
|
|
# 调用 AI 生成描述
|
|
|
response = client.chat.completions.create(
|
|
|
model=MODEL_NAME,
|
|
|
messages=[
|
|
|
{"role": "system", "content": "你是一个教学经验丰富的基础教育教师"},
|
|
|
{"role": "user", "content": prompt}
|
|
|
],
|
|
|
temperature=0.7
|
|
|
)
|
|
|
# 提取生成的描述
|
|
|
description = response.choices[0].message.content.strip()
|
|
|
# 去掉标点符号
|
|
|
description = re.sub(r'[^\w\s]', '', description)
|
|
|
# 将描述赋值给 text 属性
|
|
|
d["text"] = description
|
|
|
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
|
|
|
|
|
|
|
|
|
async def ConvertMarkdownToJson(markdown_content):
|
|
|
"""
|
|
|
生成一个 AsyncIterable，逐行返回 JSON 字符串
|
|
|
"""
|
|
|
# 将 Markdown 转换为字典
|
|
|
json_dict = markdown_to_dict(markdown_content)
|
|
|
|
|
|
# 调用 AI 扩写每个有 title 属性的节点
|
|
|
json_dict = expand_text_with_ai(json_dict)
|
|
|
|
|
|
# 提取一级目录
|
|
|
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)
|
|
|
print(contents_json)
|
|
|
yield json.dumps(contents_json, ensure_ascii=False)
|
|
|
await asyncio.sleep(0.5)
|
|
|
|
|
|
# 提取二级目录及其三级目录内容
|
|
|
level2_and_level3_json = extract_level2_and_level3(json_dict)
|
|
|
print(level2_and_level3_json)
|
|
|
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 = []
|
|
|
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、每部分都有生成完整的一、二、三级内容，不能省略。'}
|
|
|
],
|
|
|
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)
|
|
|
|
|
|
# 读取Sample.md
|
|
|
with open("Sample.md", "r", encoding="utf-8") as f:
|
|
|
markdown_content = f.read()
|
|
|
json_dict = markdown_to_dict(markdown_content)
|
|
|
|
|
|
# 调用 AI 扩写每个有 title 属性的节点
|
|
|
json_dict = expand_text_with_ai(json_dict)
|
|
|
|
|
|
print(json_dict)
|
