QingLong/AI/AiService/HuangHaiTest.py

import asyncio
import json
import re
from CommonUtil import *


# 流式生成数据的函数
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):
    """
    遍历输出所有二级目录及其三级目录，并生成指定格式的 JSON 对象
    """
    # 使用正则表达式匹配所有二级目录和三级目录及其内容
    matches = re.findall(r'^(##|###)\s+(.+)$|^-\s+(.+)$', markdown_content, re.MULTILINE)

    # 遍历匹配结果，组织层级关系
    current_level2 = None
    current_level3 = None
    level3_items = []  # 用于存储三级目录下的内容

    for level, title, item in matches:
        if level == '##':  # 二级目录
            if current_level2:
                print()  # 输出空行分隔不同的二级目录
            current_level2 = title
            # 生成二级目录的 JSON 对象
            level2_json = {
                "type": "transition",
                "data": {
                    "title": current_level2,
                    "text": current_level2
                }
            }
            print(json.dumps(level2_json, ensure_ascii=False))
        elif level == '###':  # 三级目录
            if current_level3 and level3_items:
                # 生成三级目录的 JSON 对象
                level3_json = {
                    "type": "content",
                    "data": {
                        "title": current_level3,
                        "text": current_level3,
                        "items": level3_items
                    }
                }
                print(json.dumps(level3_json, ensure_ascii=False))
                level3_items = []  # 清空三级目录下的内容
            current_level3 = title
        elif item:  # 三级目录下的内容
            # 去掉内容中的 "-" 符号
            cleaned_item = item.lstrip('- ').strip()
            # 生成三级目录下内容的 JSON 对象
            item_json = {"title": cleaned_item, "text": cleaned_item}
            level3_items.append(item_json)

    # 处理最后一个三级目录
    if current_level3 and level3_items:
        level3_json = {
            "type": "content",
            "data": {
                "title": current_level3,
                "text": current_level3,
                "items": level3_items
            }
        }
        print(json.dumps(level3_json, ensure_ascii=False))

# 运行应用
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)
    ret = {"type": "cover", "data": {"title": level1_title, "text": ""}}
    print(json.dumps(ret, ensure_ascii=False))

    # 二级名称列表
    contents = extract_level2_titles(markdown_content)
    ret = {"type": "contents", "data": {"items": contents}}
    print(json.dumps(ret, ensure_ascii=False))

    # 二级目录和三级目录
    extract_level2_and_level3(markdown_content)