main
HuangHai 5 months ago
parent 2bd058ce11
commit 3931563bc7

@ -1,14 +1,11 @@
import re
import asyncio
import socket
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
from AiService.MarkdownToJsonConverter import MarkdownToJsonConverter
@ -23,136 +20,14 @@ client = OpenAI(
)
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
def convertMarkdownToJson(markdown_content):
# 创建转换器实例
converter = MarkdownToJsonConverter(client)
# 转换 Markdown 为 JSON
for item in converter.convert_markdown_to_json(markdown_content): # 使用普通 for 循环
yield item
yield f"{item}\n" # 每行数据后添加换行符
# 添加结束标记
yield '{"type": "end" }'
yield '{"type": "end" }\n' # 结束标记后也添加换行符
# 获取本机所有 IPv4 地址

Loading…
Cancel
Save