import asyncio import json import re from CommonUtil import * def update_json_with_ai(client, markdown_content, json_obj): """ 更新 JSON 对象，将 title 属性生成描述信息填充到 text 属性中 """ # 如果 json_obj 是字符串，则将其解析为字典 if isinstance(json_obj, str): json_obj = json.loads(json_obj) # 检查是否存在嵌套的 data 对象 if "data" in json_obj and isinstance(json_obj["data"], dict): data = json_obj["data"] if "title" in data: title = data["title"] description = generate_description_with_ai(client, markdown_content, title) data["text"] = description # 将生成的描述信息填充到 text 属性中 elif "title" in json_obj: # 如果 title 直接存在于 json_obj 中 title = json_obj["title"] description = generate_description_with_ai(client, markdown_content, title) json_obj["text"] = description # 递归处理 items 中的子对象 if "items" in json_obj: for item in json_obj["items"]: update_json_with_ai(client, markdown_content, item) return json_obj def generate_description_with_ai(client, markdown_content, title): """ 使用 AI 生成一句话的描述信息 """ # 构造提示词 prompt = f"以下是 Markdown 内容：\n{markdown_content}\n\n请为以下标题生成一句话的描述信息，描述信息应简洁明了，且与标题内容相关，不要包含任何序号（如 1.、2. 等）或 Markdown 语法（如 #、- 等）,也一定不要重复标题：\n{title}" try: # 调用 AI 模型 response = client.chat.completions.create( model=MODEL_NAME, # 根据需要选择合适的模型 messages=[ {"role": "system", "content": "你是一个专业的助手，能够根据上下文生成简洁的描述信息。"}, {"role": "user", "content": prompt} ], max_tokens=50 # 限制生成内容的长度 ) # 提取生成的描述信息 if response.choices and response.choices[0].message.content: description = response.choices[0].message.content.strip() # 去除序号和 Markdown 语法（如 1.、#、- 等） description = re.sub(r'[\d.]+', '', description).strip() # 去除序号 description = re.sub(r'[#-]', '', description).strip() # 去除 Markdown 语法 return description else: print(f"AI 未返回有效描述信息，标题：{title}") return title # 如果 AI 未返回有效描述，则返回标题本身 except Exception as e: print(f"调用 AI 生成描述信息时出错：{e}") return title # 如果调用失败，则返回标题本身 # 流式生成数据的函数 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) # 测试函数 async def test_generate_stream_markdown(): async for chunk in generate_stream_markdown("三角形面积"): print(chunk.decode("utf-8"), end="", flush=True) def extract_level1_title(markdown_content): """ 从 Markdown 字符串中提取一级目录的文本内容 """ # 使用正则表达式匹配一级目录的标题 match = re.search(r'^#\s+(.+)$', markdown_content, re.MULTILINE) if match: return match.group(1) # 返回一级目录的文本内容 return None def extract_level2_titles(markdown_content): """ 从 Markdown 字符串中提取所有二级目录的文本内容，返回一个数组 """ # 使用正则表达式匹配所有二级目录的标题 matches = re.findall(r'^##\s+(.+)$', markdown_content, re.MULTILINE) # 去重，确保每个二级目录标题只出现一次 unique_matches = list(dict.fromkeys(matches)) return unique_matches # 返回去重后的二级目录标题数组 def extract_level2_and_level3(markdown_content): """ 遍历 Markdown 内容，记录所有二级目录、三级目录及其下的 - 内容 """ lines = markdown_content.splitlines() # 将内容按行分割 current_level3 = None level3_items = [] result = [] # 用于存储最终的结构化数据 def save_level3(): """保存当前三级目录及其内容""" nonlocal current_level3, level3_items if current_level3: result[-1]["children"].append({ "title": current_level3, "items": level3_items.copy() }) level3_items.clear() for line in lines: line = line.strip() # 去掉前后空格 if not line: # 跳过空行 continue if line.startswith("## "): # 二级目录 # 先处理之前未完成的三级目录 if current_level3: save_level3() current_level3 = None # 处理二级目录 current_level2 = line[3:].strip() # 去掉 "## " 取标题 result.append({ "title": current_level2, "children": [] # 用于存储三级目录 }) elif line.startswith("### "): # 三级目录 # 遇到新三级目录时，先处理前一个 if current_level3: save_level3() current_level3 = line[4:].strip() # 去掉 "### " 取标题 level3_items = [] elif line.startswith("- "): # 三级目录下的内容项 if current_level3 is None: current_level3 = "未命名章节" # 兜底处理 level3_items.append(line[2:].strip()) # 去掉 "- " 取内容 # 处理最后一个三级目录 if current_level3: save_level3() return result def add_descriptions_to_structure(client, markdown_content, structure): """ 遍历结构化数据，为每个二级和三级文本生成描述信息，并赋值给 text 属性 """ for level2 in structure: # 为二级目录生成描述信息 level2_title = level2["title"] level2["text"] = generate_description_with_ai(client, markdown_content, level2_title) # 遍历三级目录 for level3 in level2["children"]: # 为三级目录生成描述信息 level3_title = level3["title"] level3["text"] = generate_description_with_ai(client, markdown_content, level3_title) return structure # 运行应用 if __name__ == "__main__": # 使用 asyncio.run 运行异步函数 # asyncio.run(test_generate_stream_markdown()) # 读取Sample.md内容 with open("Sample.md", "r", encoding="utf-8") as file: markdown_content = file.read() # 一级名称 level1_title = extract_level1_title(markdown_content) json_obj = {"type": "cover", "data": {"title": level1_title, "text": ""}} # print(json.dumps(json_obj, ensure_ascii=False)) # 更新 JSON 对象 updated_json = update_json_with_ai(client, markdown_content, json_obj) print(json.dumps(updated_json, ensure_ascii=False, indent=2)) # 二级名称列表 contents = extract_level2_titles(markdown_content) json_obj = {"type": "contents", "data": {"items": contents}} print(json.dumps(json_obj, ensure_ascii=False)) # 二级目录和三级目录 result = extract_level2_and_level3(markdown_content) print(json.dumps(result, ensure_ascii=False, indent=2)) result=add_descriptions_to_structure(client, markdown_content, result) print(json.dumps(result, ensure_ascii=False, indent=2))