QingLong/AI/D1_GenerateMarkdown.py

# -*- coding: utf-8 -*-
import time
from pathlib import Path
from typing import Iterator
import dashscope
from dashscope import Generation
from dashscope.api_entities.dashscope_response import DashScopeAPIResponse


class TestDeepSeek:
    # 模型版本常量
    R1 = "deepseek-r1"
    V3 = "deepseek-v3"
    # API密钥（建议从环境变量读取）
    API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc"

    @staticmethod
    def get_system_prompt() -> str:
        """读取系统提示文件"""
        md_path = Path(r"D:\dsWork\QingLong\PptGenerator\md-file\readme\default.md")
        if not md_path.exists():
            raise FileNotFoundError(f"提示文件不存在: {md_path}")
        return md_path.read_text(encoding='utf-8')

    @classmethod
    def call_with_stream(cls) -> Iterator[DashScopeAPIResponse]:
        """流式调用API"""
        system_prompt = (
                            "返回的教案内容请严格按照指定的markdown语法格式返回，用来生成ppt。"
                            "需要注意:1、最开头的Title:..,Author: ... Date:... 不能省略掉，有用处。"
                            "2、第一级用#,第二级用##,第二级是独立的PPT一个页，每页中的数据必须有> 开头,"
                            "如果是页面中的条目，必须用 - 开头,结果不要返回图片！请仔细阅读参考格式："
                        ) + cls.get_system_prompt()

        try:
            return Generation.call(
                model=cls.R1,
                api_key=cls.API_KEY,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": "请帮我生成一个小学数学内角和的教案，按标准格式输出。"}
                ],
                result_format='message',
                stream=True,  # 启用流式模式
                incremental_output=True  # 启用增量输出
            )
        except Exception as e:
            raise RuntimeError(f"API调用失败: {str(e)}") from e

    @classmethod
    def main(cls):
        """主执行方法（带实时进度提示）"""
        print("🔄 开始生成教案，请稍候...")
        start_time = time.time()
        last_update = start_time
        spinner = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']
        spinner_idx = 0

        try:
            responses = cls.call_with_stream()
            full_content = []
            has_content = False

            for response in responses:
                current_time = time.time()
                # 更新进度动画（每秒更新）
                if current_time - last_update > 0.1:
                    print(f"\r{spinner[spinner_idx % 10]} 生成中 [{int(current_time - start_time)}s]", end="")
                    spinner_idx += 1
                    last_update = current_time

                # 处理响应内容
                if response.status_code == 200 and response.output:
                    chunk = response.output.choices[0].message.content
                    if chunk:
                        full_content.append(chunk)
                        if not has_content:
                            print("\n\n✅ 开始接收内容：")
                            has_content = True
                        print(chunk, end="", flush=True)  # 实时显示内容

            # 最终处理
            if full_content:
                md_content = ''.join(full_content)
                print(f"\n\n🎉 生成完成！总耗时: {int(time.time() - start_time)}秒")

                # 保存文件
                output_path = Path(r"D:\dsWork\QingLong\PptGenerator\md-file\readme\5.md")
                output_path.parent.mkdir(parents=True, exist_ok=True)
                output_path.write_text(md_content, encoding='utf-8')
                print(f"💾 结果已保存至: {output_path}")
            else:
                print("\n⚠️ 未收到有效内容，请检查提示词和API配置")

        except Exception as e:
            print(f"\n\n❌ 发生错误: {str(e)}")
            print("请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code")
        finally:
            print("\n🔚 程序执行结束")


if __name__ == "__main__":
    dashscope.api_key = TestDeepSeek.API_KEY
    TestDeepSeek.main()