main
黄海 5 months ago
parent 090cbbb8e0
commit 3e4e486dbe

@ -1,104 +1,139 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import time import time
from pathlib import Path from pathlib import Path
from typing import Iterator from typing import Iterator, Optional
import dashscope import dashscope
from dashscope import Generation from dashscope import Generation
from dashscope.api_entities.dashscope_response import DashScopeAPIResponse from dashscope.api_entities.dashscope_response import DashScopeAPIResponse
class TestDeepSeek: class MarkdownGenerator:
# 模型版本常量 """Markdown教学大纲生成器"""
R1 = "deepseek-r1"
V3 = "deepseek-v3" # 固定配置项
# API密钥建议从环境变量读取 DEFAULT_TEMPLATE = Path(r"D:\dsWork\QingLong\AI\md-file\readme\default.md")
DEFAULT_OUTPUT_DIR = Path(r"D:\dsWork\QingLong\AI\md-file\readme")
MODEL_R1 = "deepseek-r1"
API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc" API_KEY = "sk-01d13a39e09844038322108ecdbd1bbc"
@staticmethod def __init__(
def get_system_prompt() -> str: self,
"""读取系统提示文件""" course_name: str,
md_path = Path(r"D:\dsWork\QingLong\PptGenerator\md-file\readme\default.md") output_path: Optional[Path] = None,
if not md_path.exists(): template_path: Optional[Path] = None
raise FileNotFoundError(f"提示文件不存在: {md_path}") ):
return md_path.read_text(encoding='utf-8') """
初始化生成器
@classmethod
def call_with_stream(cls) -> Iterator[DashScopeAPIResponse]: :param course_name: 课程名称小学数学内角和
"""流式调用API""" :param output_path: 输出文件路径可选
:param template_path: 模板文件路径可选
"""
self.course_name = course_name
self.template_path = template_path or self.DEFAULT_TEMPLATE
self.output_path = output_path or self.DEFAULT_OUTPUT_DIR / f"{course_name}.md"
self._validate_paths()
def _validate_paths(self):
"""路径验证"""
if not self.template_path.exists():
raise FileNotFoundError(f"模板文件不存在: {self.template_path}")
self.output_path.parent.mkdir(parents=True, exist_ok=True)
if not self.output_path.parent.is_dir():
raise NotADirectoryError(f"无效的输出目录: {self.output_path.parent}")
def _load_template(self) -> str:
"""加载模板内容"""
return self.template_path.read_text(encoding='utf-8')
def _generate_stream(self) -> Iterator[DashScopeAPIResponse]:
"""流式生成内容"""
system_prompt = ( system_prompt = (
"返回的教案内容请严格按照指定的markdown语法格式返回用来生成ppt。" "请严格按照以下Markdown格式生成教学大纲\n"
"需要注意:1、最开头的Title:..,Author: ... Date:... 不能省略掉,有用处。" f"{self._load_template()}\n"
"2、第一级用#,第二级用##,第二级是独立的PPT一个页每页中的数据必须有> 开头," "注意:\n"
"如果是页面中的条目,必须用 - 开头,结果不要返回图片!请仔细阅读参考格式:" "1. 使用#/##标记标题\n"
) + cls.get_system_prompt() "2. 每页内容以>开头\n"
"3. 列表项使用-标记\n"
)
try: return Generation.call(
return Generation.call( model=self.MODEL_R1,
model=cls.R1, api_key=self.API_KEY,
api_key=cls.API_KEY, messages=[
messages=[ {"role": "system", "content": system_prompt},
{"role": "system", "content": system_prompt}, {"role": "user", "content": f"请生成《{self.course_name}》教学大纲"}
{"role": "user", "content": "请帮我生成一个小学数学内角和的教案,按标准格式输出。"} ],
], result_format='message',
result_format='message', stream=True,
stream=True, # 启用流式模式 incremental_output=True
incremental_output=True # 启用增量输出 )
)
except Exception as e:
raise RuntimeError(f"API调用失败: {str(e)}") from e
@classmethod def run(self) -> bool:
def main(cls): """执行生成流程"""
"""主执行方法(带实时进度提示)"""
print("🔄 开始生成教案,请稍候...")
start_time = time.time() start_time = time.time()
last_update = start_time
spinner = ['', '', '', '', '', '', '', '', '', ''] spinner = ['', '', '', '', '', '', '', '', '', '']
spinner_idx = 0 content_buffer = []
try: try:
responses = cls.call_with_stream() print(f"🚀 开始生成【{self.course_name}】教学大纲")
full_content = [] responses = self._generate_stream()
has_content = False
for idx, response in enumerate(responses):
for response in responses: # 显示进度
current_time = time.time() print(f"\r{spinner[idx % 10]} 生成中({int(time.time() - start_time)}秒)", end="")
# 更新进度动画(每秒更新)
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: if response.status_code == 200 and response.output:
chunk = response.output.choices[0].message.content if chunk := response.output.choices[0].message.content:
if chunk: content_buffer.append(chunk)
full_content.append(chunk) if len(content_buffer) == 1:
if not has_content: print("\n\n📝 内容生成开始:")
print("\n\n✅ 开始接收内容:") print(chunk, end="", flush=True)
has_content = True
print(chunk, end="", flush=True) # 实时显示内容 # 保存结果
if content_buffer:
# 最终处理 self.output_path.write_text(''.join(content_buffer), encoding='utf-8')
if full_content: print(f"\n\n✅ 生成成功!耗时 {int(time.time() - start_time)}")
md_content = ''.join(full_content) print(f"📂 文件已保存至:{self.output_path}")
print(f"\n\n🎉 生成完成!总耗时: {int(time.time() - start_time)}") return True
return False
# 保存文件
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: except Exception as e:
print(f"\n\n❌ 发生错误: {str(e)}") print(f"\n\n❌ 生成失败:{str(e)}")
print("请参考文档https://help.aliyun.com/zh/model-studio/developer-reference/error-code") return False
finally:
print("\n🔚 程序执行结束")
def generate_document(
course_name: str,
output_path: Optional[str] = None,
template_path: Optional[str] = None
):
"""
生成文档入口函数
:param course_name: 课程名称
:param output_path: 输出路径可选
:param template_path: 模板路径可选
"""
try:
generator = MarkdownGenerator(
course_name=course_name,
output_path=Path(output_path) if output_path else None,
template_path=Path(template_path) if template_path else None
)
return generator.run()
except Exception as e:
print(f"❌ 初始化失败:{str(e)}")
return False
if __name__ == "__main__": if __name__ == "__main__":
dashscope.api_key = TestDeepSeek.API_KEY # 使用示例(实际使用时参数可写在此处)
TestDeepSeek.main() dashscope.api_key = MarkdownGenerator.API_KEY
generate_document(
course_name="小学数学三角形内角和",
output_path=r"D:\dsWork\QingLong\AI\md-file\readme\5.md"
)
Loading…
Cancel
Save