QingLong/AI/AiService/MarkdownToJsonUtil.py

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" }'