diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..cca5f05 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,83 @@ +name: CI + +on: + push: + branches: + - main + - dev + pull_request: + branches: + - main + - dev + workflow_dispatch: + +permissions: + contents: read + +jobs: + frontend: + name: Frontend + runs-on: ubuntu-latest + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Node.js + uses: actions/setup-node@v4 + with: + node-version: 20 + cache: npm + cache-dependency-path: web/package-lock.json + + - name: Install frontend dependencies + working-directory: web + run: npm ci + + - name: Lint frontend + working-directory: web + run: npm run lint + + - name: Test frontend + working-directory: web + run: npm run test + + - name: Build frontend + working-directory: web + run: npm run build + + backend: + name: Backend Python ${{ matrix.python-version }} + runs-on: ubuntu-latest + + strategy: + fail-fast: false + matrix: + python-version: + - "3.11" + - "3.12" + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + cache-dependency-path: | + requirements.txt + requirements-dev.txt + + - name: Install backend dependencies + run: python -m pip install -r requirements-dev.txt + + - name: Ruff lint + run: ruff check api src tests main.py + + - name: Ruff format check + run: ruff format --check api src tests main.py + + - name: Test backend with coverage + run: python -m pytest --cov=api --cov=src --cov-report=term-missing diff --git a/.gitignore b/.gitignore index 7d98d3f..b2109b8 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,8 @@ ENV/ *.egg-info/ .installed.cfg *.egg +.coverage* +htmlcov/ # Node.js (前端依赖由 web/.gitignore 管理，这里作为备份) node_modules/ diff --git a/README.md b/README.md index 2178bc9..c5145b3 100644 --- a/README.md +++ b/README.md @@ -161,6 +161,33 @@ output/ppt_20241201_123456/ └── images/ # Slide images ``` +## 🧪 Development Checks + +Recommended local runtime: +- Python 3.11 or 3.12 +- Node.js 20 + +Backend checks: + +```bash +python -m pip install -r requirements-dev.txt +ruff check api src tests main.py +ruff format --check api src tests main.py +python -m pytest --cov=api --cov=src --cov-report=term-missing +``` + +Frontend checks: + +```bash +cd web +npm ci +npm run lint +npm run test +npm run build +``` + +GitHub Actions runs the same default checks on `main` and `dev` pull requests and pushes. Real model API calls are intentionally not part of the default CI because they require private keys and can be flaky. + ## 📋 TODO - [ ] Upgrade generated PPT images into structured, editable PPT content diff --git a/README_zh.md b/README_zh.md index 6cb5748..f6a1a93 100644 --- a/README_zh.md +++ b/README_zh.md @@ -161,6 +161,33 @@ output/ppt_20241201_123456/ └── images/ # 幻灯片图片 ``` +## 🧪 开发检查 + +推荐本地运行环境： +- Python 3.11 或 3.12 +- Node.js 20 + +后端检查： + +```bash +python -m pip install -r requirements-dev.txt +ruff check api src tests main.py +ruff format --check api src tests main.py +python -m pytest --cov=api --cov=src --cov-report=term-missing +``` + +前端检查： + +```bash +cd web +npm ci +npm run lint +npm run test +npm run build +``` + +GitHub Actions 会在 `main` 和 `dev` 的 Pull Request 与 push 上运行这些默认检查。真实模型 API 调用需要私有 Key，且容易受外部服务波动影响，因此不放入默认 CI。 + ## 📋 TODO - [ ] 将生成的 PPT 图片增强为结构化、可编辑的 PPT 内容 diff --git a/api/main.py b/api/main.py index fd70d75..c216434 100644 --- a/api/main.py +++ b/api/main.py @@ -15,7 +15,7 @@ app = FastAPI( title="AI PPT Generator API", description="API for AI-powered PPT generation with WebUI", - version="1.0.0" + version="1.0.0", ) # 配置 CORS 中间件 @@ -50,4 +50,5 @@ async def health_check(): if __name__ == "__main__": import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/api/models.py b/api/models.py index f0db170..27107f3 100644 --- a/api/models.py +++ b/api/models.py @@ -8,8 +8,10 @@ # ============ 上传相关模型 ============ + class UploadResponse(BaseModel): """文件上传响应""" + success: bool content: str filename: str @@ -19,8 +21,10 @@ class UploadResponse(BaseModel): # ============ 生成相关模型 ============ + class ImageApiConfig(BaseModel): """图像模型 API 配置""" + api_key: str = Field(..., description="图像模型 API 密钥") base_url: str = Field(..., description="图像模型 API 基础 URL") model: str = Field("gemini-3-pro-image-preview", description="图像模型名称") @@ -28,27 +32,34 @@ class ImageApiConfig(BaseModel): class TextApiConfig(BaseModel): """文本模型 API 配置""" + api_key: str = Field(..., description="文本模型 API 密钥") base_url: str = Field(..., description="文本模型 API 基础 URL") model: str = Field("gemini-3-pro-preview", description="文本模型名称") format: Literal["gemini", "openai"] = Field("gemini", description="API 格式") thinking: Literal["enabled", "disabled"] = Field("disabled", description="思考模式") - thinking_level: Optional[Literal["low", "high"]] = Field(None, description="旧版 Gemini 思考深度") + thinking_level: Optional[Literal["low", "high"]] = Field( + None, description="旧版 Gemini 思考深度" + ) class ModelProfileConfig(BaseModel): """模型 profile 配置""" + id: Optional[str] = Field(None, description="Profile ID") label: Optional[str] = Field(None, description="显示名称") model: str = Field(..., description="模型名称") base_url: str = Field(..., description="OpenAI-compatible Base URL") api_key: str = Field(..., description="API Key") adapter: str = Field("openai_chat", description="适配器") - thinking: Optional[Literal["enabled", "disabled"]] = Field(None, description="OpenAI-compatible thinking mode") + thinking: Optional[Literal["enabled", "disabled"]] = Field( + None, description="OpenAI-compatible thinking mode" + ) class ModelProfilesConfig(BaseModel): """三角色模型配置""" + prompt_model: ModelProfileConfig image_model: ModelProfileConfig edit_model: Optional[ModelProfileConfig] = None @@ -56,6 +67,7 @@ class ModelProfilesConfig(BaseModel): class GenerationConfig(BaseModel): """生成配置（完整版）""" + # 图像模型配置 image: Optional[ImageApiConfig] = Field(None, description="图像模型配置") # 文本模型配置 @@ -74,49 +86,49 @@ class GenerationConfig(BaseModel): style: str = Field("现代简约商务风格", description="PPT 风格") target_audience: str = Field("专业人士", description="目标受众") user_requirements: str = Field("", description="用户定制要求") - + def get_image_api_key(self) -> str: """获取图像模型 API 密钥""" if self.image: return self.image.api_key return self.api_key or "" - + def get_image_base_url(self) -> str: """获取图像模型 API 基础 URL""" if self.image: return self.image.base_url return self.base_url or "" - + def get_image_model(self) -> str: """获取图像模型名称""" if self.image: return self.image.model return "gemini-3-pro-image-preview" - + def get_text_api_key(self) -> str: """获取文本模型 API 密钥""" if self.text: return self.text.api_key return self.api_key or "" - + def get_text_base_url(self) -> str: """获取文本模型 API 基础 URL""" if self.text: return self.text.base_url return self.base_url or "" - + def get_text_model(self) -> str: """获取文本模型名称""" if self.text: return self.text.model return "gemini-3-pro-preview" - + def get_text_format(self) -> str: """获取文本 API 格式""" if self.text: return self.text.format return "gemini" - + def get_thinking_level(self) -> Optional[str]: """获取思考深度""" if self.text: @@ -132,16 +144,17 @@ def get_thinking(self) -> str: class GenerationRequest(BaseModel): """生成请求""" + content: str = Field(..., description="Markdown 内容") config: GenerationConfig slide_prompts: Optional[List["ConfirmedSlidePrompt"]] = Field( - None, - description="用户确认后的逐页图像 prompt；存在时跳过文本生成阶段" + None, description="用户确认后的逐页图像 prompt；存在时跳过文本生成阶段" ) class SlideOutlineData(BaseModel): """用户可编辑的单页设计大纲""" + page: int = Field(..., ge=1) title: str narrative_goal: str @@ -151,6 +164,7 @@ class SlideOutlineData(BaseModel): class DeckOutlineData(BaseModel): """整套 PPT 设计大纲""" + title: str user_requirements: str = "" design_style: str @@ -160,6 +174,7 @@ class DeckOutlineData(BaseModel): class ConfirmedSlidePrompt(BaseModel): """用户确认后的单页设计和图像 prompt""" + page: int = Field(..., ge=1) title: str content_summary: str @@ -169,12 +184,14 @@ class ConfirmedSlidePrompt(BaseModel): class OutlineRequest(BaseModel): """设计大纲生成请求""" + content: str = Field(..., description="Markdown 内容") config: GenerationConfig class OutlineResponse(BaseModel): """设计大纲响应""" + success: bool outline: Optional[DeckOutlineData] = None message: Optional[str] = None @@ -182,6 +199,7 @@ class OutlineResponse(BaseModel): class PromptPlanRequest(BaseModel): """逐页设计和 prompt 生成请求""" + content: str = Field(..., description="Markdown 内容") config: GenerationConfig outline: DeckOutlineData @@ -189,6 +207,7 @@ class PromptPlanRequest(BaseModel): class PromptPlanResponse(BaseModel): """逐页设计和 prompt 响应""" + success: bool slide_prompts: Optional[List[ConfirmedSlidePrompt]] = None message: Optional[str] = None @@ -202,6 +221,7 @@ class PromptPlanResponse(BaseModel): class SlideData(BaseModel): """单页幻灯片数据""" + id: str page_number: int image_base64: str @@ -210,32 +230,38 @@ class SlideData(BaseModel): class GenerationProgressEvent(BaseModel): """生成进度事件""" + type: Literal["progress"] data: dict class GenerationSlideEvent(BaseModel): """生成幻灯片事件""" + type: Literal["slide"] data: SlideData class GenerationCompleteEvent(BaseModel): """生成完成事件""" + type: Literal["complete"] data: dict class GenerationErrorEvent(BaseModel): """生成错误事件""" + type: Literal["error"] data: dict # ============ 编辑相关模型 ============ + class EditConfig(BaseModel): """编辑配置""" + api_key: Optional[str] = Field(None, description="API 密钥") base_url: Optional[str] = Field(None, description="API 基础 URL") model: str = Field("gpt-image-2", description="图像编辑模型名称") @@ -246,6 +272,7 @@ class EditConfig(BaseModel): class EditRequest(BaseModel): """编辑请求""" + image_base64: str = Field(..., description="原始图片 Base64") instruction: str = Field(..., description="修改指令") config: EditConfig @@ -253,6 +280,7 @@ class EditRequest(BaseModel): class EditResponse(BaseModel): """编辑响应""" + success: bool image_base64: Optional[str] = None message: Optional[str] = None @@ -260,13 +288,16 @@ class EditResponse(BaseModel): # ============ 导出相关模型 ============ + class ExportSlide(BaseModel): """导出的幻灯片""" + image_base64: str class ExportRequest(BaseModel): """导出请求""" + slides: List[ExportSlide] format: Literal["pdf", "pptx"] = Field(..., description="导出格式") aspect_ratio: Literal["16:9", "4:3"] = Field("16:9", description="导出画幅比例") @@ -274,12 +305,14 @@ class ExportRequest(BaseModel): class ExportResponse(BaseModel): """导出响应""" + success: bool message: Optional[str] = None class ModelProfilesResponse(BaseModel): """脱敏模型 profile 响应""" + success: bool profiles: Optional[dict] = None message: Optional[str] = None diff --git a/api/routes/edit.py b/api/routes/edit.py index 6490189..ba334b8 100644 --- a/api/routes/edit.py +++ b/api/routes/edit.py @@ -4,6 +4,7 @@ import sys import base64 +import asyncio import tempfile from pathlib import Path from fastapi import APIRouter, HTTPException @@ -23,54 +24,50 @@ async def edit_image(request: EditRequest): """ 图生图编辑 - + Args: request: 编辑请求，包含原始图片和修改指令 - + Returns: EditResponse: 包含编辑后的图片 """ + temp_input_path = None + temp_output_path = None try: client = ModelRouter(profiles_from_edit_config(request.config)) - + # 解码 base64 图片并保存到临时文件 image_data = base64.b64decode(request.image_base64) - - with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as temp_input: + + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_input: temp_input.write(image_data) temp_input_path = temp_input.name - - with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as temp_output: + + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_output: temp_output_path = temp_output.name - - try: - # 调用图生图 API - output_path = client.edit_image( - prompt=request.instruction, - source_image_path=temp_input_path, - aspect_ratio=request.config.aspect_ratio, - quality=request.config.quality, - output_path=temp_output_path - ) - - # 读取生成的图片并转换为 base64 - with open(output_path, 'rb') as f: - output_image_data = f.read() - output_image_base64 = base64.b64encode(output_image_data).decode('utf-8') - - return EditResponse( - success=True, - image_base64=output_image_base64, - message="编辑成功" - ) - - finally: - # 清理临时文件 + + # 调用图生图 API + output_path = await asyncio.to_thread( + client.edit_image, + prompt=request.instruction, + source_image_path=temp_input_path, + aspect_ratio=request.config.aspect_ratio, + quality=request.config.quality, + output_path=temp_output_path, + ) + + # 读取生成的图片并转换为 base64 + with open(output_path, "rb") as f: + output_image_data = f.read() + output_image_base64 = base64.b64encode(output_image_data).decode("utf-8") + + return EditResponse(success=True, image_base64=output_image_base64, message="编辑成功") + + except Exception as e: + raise HTTPException(status_code=500, detail=f"编辑失败: {str(e)}") + finally: + # 清理临时文件 + if temp_input_path: Path(temp_input_path).unlink(missing_ok=True) + if temp_output_path: Path(temp_output_path).unlink(missing_ok=True) - - except Exception as e: - raise HTTPException( - status_code=500, - detail=f"编辑失败: {str(e)}" - ) diff --git a/api/routes/export.py b/api/routes/export.py index 5701445..944aa3f 100644 --- a/api/routes/export.py +++ b/api/routes/export.py @@ -4,6 +4,7 @@ import sys import base64 +import asyncio import tempfile import shutil from pathlib import Path @@ -25,57 +26,56 @@ async def export_presentation(request: ExportRequest): """ 导出演示文稿 - + Args: request: 导出请求，包含所有幻灯片和格式 - + Returns: FileResponse: 导出的文件 """ try: temp_dir = Path(tempfile.mkdtemp()) image_paths = [] - + # 解码并保存所有图片 for idx, slide in enumerate(request.slides): image_data = base64.b64decode(slide.image_base64) image_path = temp_dir / f"slide_{idx + 1}.png" - - with open(image_path, 'wb') as f: + + with open(image_path, "wb") as f: f.write(image_data) - + image_paths.append(str(image_path)) - + # 根据格式导出 if request.format == "pdf": output_path = temp_dir / "presentation.pdf" exporter = PDFExporter() - exporter.export(image_paths, str(output_path)) - + await asyncio.to_thread(exporter.export, image_paths, str(output_path)) + return FileResponse( path=str(output_path), media_type="application/pdf", filename="presentation.pdf", - background=BackgroundTask(shutil.rmtree, temp_dir, ignore_errors=True) + background=BackgroundTask(shutil.rmtree, temp_dir, ignore_errors=True), ) - + elif request.format == "pptx": output_path = temp_dir / "presentation.pptx" - _export_pptx(image_paths, str(output_path), aspect_ratio=request.aspect_ratio) - + await asyncio.to_thread( + _export_pptx, image_paths, str(output_path), aspect_ratio=request.aspect_ratio + ) + return FileResponse( path=str(output_path), media_type="application/vnd.openxmlformats-officedocument.presentationml.presentation", filename="presentation.pptx", - background=BackgroundTask(shutil.rmtree, temp_dir, ignore_errors=True) + background=BackgroundTask(shutil.rmtree, temp_dir, ignore_errors=True), ) - + else: - raise HTTPException( - status_code=400, - detail=f"不支持的导出格式: {request.format}" - ) - + raise HTTPException(status_code=400, detail=f"不支持的导出格式: {request.format}") + except HTTPException: if "temp_dir" in locals(): shutil.rmtree(temp_dir, ignore_errors=True) @@ -84,16 +84,13 @@ async def export_presentation(request: ExportRequest): except Exception as e: if "temp_dir" in locals(): shutil.rmtree(temp_dir, ignore_errors=True) - raise HTTPException( - status_code=500, - detail=f"导出失败: {str(e)}" - ) + raise HTTPException(status_code=500, detail=f"导出失败: {str(e)}") def _export_pptx(image_paths: list, output_path: str, aspect_ratio: str = "16:9"): """ 导出为 PPTX 格式 - + Args: image_paths: 图片路径列表 output_path: 输出路径 @@ -103,10 +100,10 @@ def _export_pptx(image_paths: list, output_path: str, aspect_ratio: str = "16:9" from pptx.util import Inches except ImportError: raise Exception("需要安装 python-pptx: pip install python-pptx") - + # 创建演示文稿 prs = Presentation() - + # 设置幻灯片尺寸 if aspect_ratio == "4:3": prs.slide_width = Inches(10) @@ -114,20 +111,15 @@ def _export_pptx(image_paths: list, output_path: str, aspect_ratio: str = "16:9" else: prs.slide_width = Inches(10) prs.slide_height = Inches(5.625) - + # 添加每一页 for image_path in image_paths: # 使用空白布局 blank_slide_layout = prs.slide_layouts[6] slide = prs.slides.add_slide(blank_slide_layout) - + # 添加图片，填充整个幻灯片 - slide.shapes.add_picture( - image_path, - 0, 0, - width=prs.slide_width, - height=prs.slide_height - ) - + slide.shapes.add_picture(image_path, 0, 0, width=prs.slide_width, height=prs.slide_height) + # 保存 prs.save(output_path) diff --git a/api/routes/generate.py b/api/routes/generate.py index 1e7cb4d..e6463fe 100644 --- a/api/routes/generate.py +++ b/api/routes/generate.py @@ -21,7 +21,13 @@ from src.model_router import ModelRouter from src.prompt_generator import PromptGenerator from src.models import DeckOutline, PromptData, SlidePrompt -from ..models import GenerationRequest, OutlineRequest, OutlineResponse, PromptPlanRequest, PromptPlanResponse +from ..models import ( + GenerationRequest, + OutlineRequest, + OutlineResponse, + PromptPlanRequest, + PromptPlanResponse, +) from ..profile_resolver import profiles_from_generation_config router = APIRouter(prefix="/api", tags=["generate"]) @@ -54,7 +60,9 @@ def _to_ppt_config(config) -> PPTConfig: def _prompt_data_from_confirmed(request: GenerationRequest, ppt_config: PPTConfig) -> PromptData: slide_prompts = sorted(request.slide_prompts or [], key=lambda item: item.page) if len(slide_prompts) != ppt_config.num_pages: - raise ValueError(f"确认后的逐页设计数量不匹配: 期望{ppt_config.num_pages}页，实际{len(slide_prompts)}页") + raise ValueError( + f"确认后的逐页设计数量不匹配: 期望{ppt_config.num_pages}页，实际{len(slide_prompts)}页" + ) pages = [item.page for item in slide_prompts] expected = list(range(1, ppt_config.num_pages + 1)) @@ -84,11 +92,11 @@ def generate_single_slide_sync( output_path: str, config: PPTConfig, max_retries: int = 3, - retry_delay: float = 2.0 + retry_delay: float = 2.0, ) -> str: """ 同步生成单页幻灯片图片 - + Args: client: AI 客户端 slide_prompt: 幻灯片 prompt 对象 @@ -96,7 +104,7 @@ def generate_single_slide_sync( config: PPT 配置 max_retries: 最大重试次数 retry_delay: 重试延迟（秒） - + Returns: 生成的图片路径 """ @@ -106,7 +114,7 @@ def generate_single_slide_sync( prompt=slide_prompt.prompt, aspect_ratio=config.aspect_ratio, quality=config.quality, - output_path=output_path + output_path=output_path, ) return output_path except Exception as e: @@ -119,7 +127,7 @@ def generate_single_slide_sync( async def generate_stream(request: GenerationRequest) -> AsyncGenerator[str, None]: """ 生成 PPT 的流式响应 - + 使用 Server-Sent Events (SSE) 格式返回进度和结果 """ try: @@ -127,14 +135,14 @@ async def generate_stream(request: GenerationRequest) -> AsyncGenerator[str, Non # 配置 PPT ppt_config = _to_ppt_config(request.config) - + # 创建客户端和 Prompt 生成器 client = ModelRouter(profiles) prompt_generator = PromptGenerator(client) - + # 发送开始事件 yield f"data: {json.dumps({'type': 'progress', 'data': {'status': 'started', 'current': 0, 'total': 0, 'message': '开始生成 PPT'}})}\n\n" - + loop = asyncio.get_event_loop() if request.slide_prompts: @@ -145,30 +153,27 @@ async def generate_stream(request: GenerationRequest) -> AsyncGenerator[str, Non yield f"data: {json.dumps({'type': 'progress', 'data': {'status': 'generating_prompts', 'current': 0, 'total': 0, 'message': '正在生成 Prompt...'}})}\n\n" prompt_data = await loop.run_in_executor( - None, - prompt_generator.generate, - request.content, - ppt_config + None, prompt_generator.generate, request.content, ppt_config ) - + total_slides = len(prompt_data.slide_prompts) - + # 发送 Prompt 生成完成事件 yield f"data: {json.dumps({'type': 'progress', 'data': {'status': 'prompts_ready', 'current': 0, 'total': total_slides, 'message': f'Prompt 生成完成，开始生成 {total_slides} 页图片'}})}\n\n" - + # 创建临时目录 with tempfile.TemporaryDirectory() as temp_dir: # 逐页生成并发送 for idx, slide_prompt in enumerate(prompt_data.slide_prompts): page_num = idx + 1 - + # 发送当前页开始生成的进度 yield f"data: {json.dumps({'type': 'progress', 'data': {'status': 'generating_images', 'current': idx, 'total': total_slides, 'message': f'正在生成第 {page_num}/{total_slides} 页...'}})}\n\n" - + try: # 生成单页图片 temp_output = f"{temp_dir}/slide_{page_num}.png" - + # 在线程池中运行同步的图片生成 await loop.run_in_executor( None, @@ -176,69 +181,62 @@ async def generate_stream(request: GenerationRequest) -> AsyncGenerator[str, Non client, slide_prompt, temp_output, - ppt_config + ppt_config, ) - + # 读取图片并转换为 base64 - with open(temp_output, 'rb') as f: + with open(temp_output, "rb") as f: image_data = f.read() - image_base64 = base64.b64encode(image_data).decode('utf-8') - + image_base64 = base64.b64encode(image_data).decode("utf-8") + # 发送幻灯片事件 slide_data = { - 'type': 'slide', - 'data': { - 'id': f"slide_{page_num}", - 'page_number': page_num, - 'image_base64': image_base64, - 'prompt': slide_prompt.prompt if hasattr(slide_prompt, 'prompt') else str(slide_prompt) - } + "type": "slide", + "data": { + "id": f"slide_{page_num}", + "page_number": page_num, + "image_base64": image_base64, + "prompt": slide_prompt.prompt + if hasattr(slide_prompt, "prompt") + else str(slide_prompt), + }, } yield f"data: {json.dumps(slide_data)}\n\n" - + # 发送进度更新 progress_data = { - 'type': 'progress', - 'data': { - 'status': 'generating_images', - 'current': page_num, - 'total': total_slides, - 'message': f'已生成 {page_num}/{total_slides} 页' - } + "type": "progress", + "data": { + "status": "generating_images", + "current": page_num, + "total": total_slides, + "message": f"已生成 {page_num}/{total_slides} 页", + }, } yield f"data: {json.dumps(progress_data)}\n\n" - + except Exception as e: # 发送错误事件但继续生成其他页面 error_data = { - 'type': 'error', - 'data': { - 'fatal': False, - 'page': page_num, - 'message': f'第 {page_num} 页生成失败: {str(e)}' - } + "type": "error", + "data": { + "fatal": False, + "page": page_num, + "message": f"第 {page_num} 页生成失败: {str(e)}", + }, } yield f"data: {json.dumps(error_data)}\n\n" - + # 发送完成事件 complete_data = { - 'type': 'complete', - 'data': { - 'status': 'completed', - 'message': 'PPT 生成完成' - } + "type": "complete", + "data": {"status": "completed", "message": "PPT 生成完成"}, } yield f"data: {json.dumps(complete_data)}\n\n" - + except Exception as e: # 发送致命错误事件 - error_data = { - 'type': 'error', - 'data': { - 'fatal': True, - 'message': f'生成失败: {str(e)}' - } - } + error_data = {"type": "error", "data": {"fatal": True, "message": f"生成失败: {str(e)}"}} yield f"data: {json.dumps(error_data)}\n\n" @@ -246,12 +244,12 @@ async def generate_stream(request: GenerationRequest) -> AsyncGenerator[str, Non async def generate_ppt(request: GenerationRequest): """ 生成 PPT - + 使用 Server-Sent Events (SSE) 流式返回生成进度和结果 - + Args: request: 生成请求，包含内容和配置 - + Returns: StreamingResponse: SSE 流 """ @@ -261,8 +259,8 @@ async def generate_ppt(request: GenerationRequest): headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", - "X-Accel-Buffering": "no" - } + "X-Accel-Buffering": "no", + }, ) diff --git a/api/routes/upload.py b/api/routes/upload.py index fd998db..eedb064 100644 --- a/api/routes/upload.py +++ b/api/routes/upload.py @@ -24,14 +24,14 @@ def validate_supported_file(filename: str) -> bool: """ 验证文件是否为 Markdown 文件 - + Property 1: File Upload Validation - For any uploaded file, if the file extension is not .md, + For any uploaded file, if the file extension is not .md, the system should reject it. - + Args: filename: 文件名 - + Returns: bool: 是否为有效的 Markdown 文件 """ @@ -44,16 +44,16 @@ def validate_supported_file(filename: str) -> bool: async def upload_file(file: UploadFile = File(...)): """ 上传 Markdown 文件 - + 接收 multipart/form-data 格式的文件上传请求， 验证文件类型并返回文件内容。 - + Args: file: 上传的文件 (multipart/form-data) - + Returns: UploadResponse: 包含文件内容、文件名和文件大小 - + Raises: HTTPException 400: 文件类型无效或编码错误 HTTPException 413: 文件大小超过限制 @@ -61,23 +61,19 @@ async def upload_file(file: UploadFile = File(...)): """ # 验证文件类型 if not validate_supported_file(file.filename): - raise HTTPException( - status_code=400, - detail="仅支持 .md/.txt/.pdf/.docx/.pptx 文件格式" - ) - + raise HTTPException(status_code=400, detail="仅支持 .md/.txt/.pdf/.docx/.pptx 文件格式") + try: # 读取文件内容 content = await file.read() - + # 验证文件大小 file_size = len(content) if file_size > MAX_FILE_SIZE: raise HTTPException( - status_code=413, - detail=f"文件大小超过限制 (最大 {MAX_FILE_SIZE // 1024 // 1024}MB)" + status_code=413, detail=f"文件大小超过限制 (最大 {MAX_FILE_SIZE // 1024 // 1024}MB)" ) - + suffix = Path(file.filename).suffix.lower() with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as temp_file: temp_file.write(content) @@ -91,27 +87,21 @@ async def upload_file(file: UploadFile = File(...)): raise HTTPException(status_code=500, detail=str(exc)) from exc finally: temp_path.unlink(missing_ok=True) - + return UploadResponse( success=True, content=content_str, filename=file.filename, file_size=file_size, - message="文件上传并解析成功" + message="文件上传并解析成功", ) - + except UnicodeDecodeError: - raise HTTPException( - status_code=400, - detail="文件编码错误，请确保文件为 UTF-8 编码" - ) - + raise HTTPException(status_code=400, detail="文件编码错误，请确保文件为 UTF-8 编码") + except HTTPException: # Re-raise HTTP exceptions raise - + except Exception as e: - raise HTTPException( - status_code=500, - detail=f"文件上传失败: {str(e)}" - ) + raise HTTPException(status_code=500, detail=f"文件上传失败: {str(e)}") diff --git a/main.py b/main.py index aea1e1c..cd6a44e 100644 --- a/main.py +++ b/main.py @@ -28,14 +28,14 @@ def main(): # 从 Prompt 文件生成图片 python main.py --from-prompt prompts.json - """ + """, ) - + # 输入参数 input_group = parser.add_mutually_exclusive_group(required=True) input_group.add_argument("-i", "--input", help="输入资料文件路径") input_group.add_argument("--from-prompt", help="从 Prompt JSON 文件生成") - + # PPT 配置 parser.add_argument("-n", "--num-pages", type=int, default=8, help="生成页数") parser.add_argument("--style", default="现代简约商务风格", help="PPT 风格") @@ -43,29 +43,29 @@ def main(): parser.add_argument("--audience", default="专业人士", help="目标受众") parser.add_argument("--ratio", choices=["16:9", "4:3", "1:1"], default="16:9", help="图片比例") parser.add_argument("--quality", choices=["1K", "2K", "4K"], default="2K", help="图片质量") - + # 并发控制 parser.add_argument("--concurrent", type=int, default=3, help="最大并发数") parser.add_argument("--retries", type=int, default=3, help="最大重试次数") - + # 输出选项 parser.add_argument("-o", "--output", help="输出路径") parser.add_argument("--output-dir", default="output", help="输出目录") parser.add_argument("--prompt-only", action="store_true", help="仅生成 Prompt") parser.add_argument("--no-pdf", action="store_true", help="不导出 PDF") - + # API 配置（可选覆盖，默认从 config.yaml 读取） parser.add_argument("--api-key", default=None, help="API 密钥（覆盖配置文件）") parser.add_argument("--base-url", default=None, help="API 地址（覆盖配置文件）") - + args = parser.parse_args() - + # 创建配置（命令行参数覆盖配置文件） api_config = APIConfig( image_api_key=args.api_key, # None 时使用配置文件默认值 - image_base_url=args.base_url + image_base_url=args.base_url, ) - + ppt_config = PPTConfig( language=args.lang if args.lang != "中文" else None, style=args.style if args.style != "现代简约商务风格" else None, @@ -74,59 +74,64 @@ def main(): aspect_ratio=args.ratio if args.ratio != "16:9" else None, quality=args.quality if args.quality != "2K" else None, max_concurrent=args.concurrent if args.concurrent != 3 else None, - max_retries=args.retries if args.retries != 3 else None + max_retries=args.retries if args.retries != 3 else None, ) - + # 创建生成器 generator = PPTGenerator(api_config=api_config, output_dir=args.output_dir) - + try: # 从 Prompt 文件生成 if args.from_prompt: prompt_data = PromptData.load(args.from_prompt) - result = generator.generate_from_prompts(prompt_data, ppt_config, export_pdf=not args.no_pdf) + result = generator.generate_from_prompts( + prompt_data, ppt_config, export_pdf=not args.no_pdf + ) if result.success: print(f"\n🎉 完成！输出目录: {result.project_dir}") return 0 if result.success else 1 - + # 读取输入文件 input_path = Path(args.input) if not input_path.exists(): print(f"❌ 输入文件不存在: {args.input}") return 1 - - with open(input_path, 'r', encoding='utf-8') as f: + + with open(input_path, "r", encoding="utf-8") as f: source_material = f.read() - + if not source_material.strip(): print("❌ 输入文件为空") return 1 - + print(f"📄 读取资料: {len(source_material)} 字符") - + # 仅生成 Prompt if args.prompt_only: output_path = args.output or "prompts.json" - prompt_data = generator.generate_prompts_only(source_material, ppt_config, output_path=output_path) + prompt_data = generator.generate_prompts_only( + source_material, ppt_config, output_path=output_path + ) print(f"\n🎉 完成！Prompt 已保存到: {output_path}") return 0 - + # 完整生成 result = generator.generate(source_material, ppt_config, export_pdf=not args.no_pdf) - + if result.success: print(f"\n🎉 完成！输出目录: {result.project_dir}") return 0 if result.success else 1 - + except KeyboardInterrupt: - print(f"\n\n🛑 程序已被用户中断") - print(f" 💡 提示: 已完成的图片已保存在输出目录中") - print(f" 💡 可以使用 --from-prompt 参数从已生成的 prompts.json 继续") + print("\n\n🛑 程序已被用户中断") + print(" 💡 提示: 已完成的图片已保存在输出目录中") + print(" 💡 可以使用 --from-prompt 参数从已生成的 prompts.json 继续") return 130 # SIGINT 退出码 - + except Exception as e: print(f"\n❌ 错误: {e}") import traceback + traceback.print_exc() return 1 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..bd3dbb1 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,33 @@ +[tool.ruff] +target-version = "py311" +line-length = 100 +extend-exclude = [ + "web", + "output", + "test_output", + "docs/assets", +] + +[tool.ruff.format] +quote-style = "double" +indent-style = "space" +line-ending = "auto" + +[tool.ruff.lint.per-file-ignores] +"api/routes/*.py" = ["E402"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-ra" + +[tool.coverage.run] +branch = true +source = ["api", "src"] +omit = [ + "tests/*", +] + +[tool.coverage.report] +show_missing = true +skip_covered = true +fail_under = 45 diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..02a588b --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,5 @@ +-r requirements.txt + +pytest>=8.0.0 +pytest-cov>=5.0.0 +ruff>=0.5.0 diff --git a/src/__init__.py b/src/__init__.py index 101f9e2..f0c6327 100644 --- a/src/__init__.py +++ b/src/__init__.py @@ -9,6 +9,11 @@ __version__ = "0.1.0" __all__ = [ - "PPTConfig", "APIConfig", "PPTGenerator", "PromptTemplates", - "load_sample_material", "get_output_dir", "get_doc_dir" + "PPTConfig", + "APIConfig", + "PPTGenerator", + "PromptTemplates", + "load_sample_material", + "get_output_dir", + "get_doc_dir", ] diff --git a/src/client.py b/src/client.py index c562da5..6a67936 100644 --- a/src/client.py +++ b/src/client.py @@ -3,7 +3,6 @@ 封装 Gemini 和 OpenAI 格式的 API 调用 """ -from typing import Literal from google import genai from google.genai import types from PIL import Image @@ -15,26 +14,25 @@ class AIClient: """AI 客户端，封装文本和图像生成 API""" - + def __init__(self, config: APIConfig = None): self.config = config or APIConfig() self.timeout_config = get_timeout_config() - + # Gemini 客户端（用于图像生成） self._gemini_client = genai.Client( - api_key=self.config.image_api_key, - http_options={'baseUrl': self.config.image_base_url} + api_key=self.config.image_api_key, http_options={"baseUrl": self.config.image_base_url} ) - + # OpenAI 格式客户端（用于文本生成，可选） self._openai_client = None if self.config.text_api_format == "openai": self._openai_client = OpenAI( api_key=self.config.text_api_key, base_url=self.config.text_base_url, - timeout=self.timeout_config['text_generation'] + timeout=self.timeout_config["text_generation"], ) - + def generate_text(self, prompt: str, system_instruction: str = None) -> str: """ 调用文本模型生成内容 @@ -44,58 +42,56 @@ def generate_text(self, prompt: str, system_instruction: str = None) -> str: return self._generate_text_openai(prompt, system_instruction) else: return self._generate_text_gemini(prompt, system_instruction) - + def _generate_text_gemini(self, prompt: str, system_instruction: str = None) -> str: """使用 Gemini API 生成文本""" config = types.GenerateContentConfig() - + # 设置系统指令 if system_instruction: config.system_instruction = system_instruction - + # 设置思考配置（仅支持 Gemini 3+ 系列） if self.config.text_thinking_level and self._supports_thinking(self.config.text_model): config.thinking_config = types.ThinkingConfig( thinking_level=self.config.text_thinking_level ) - + response = self._gemini_client.models.generate_content( - model=self.config.text_model, - contents=prompt, - config=config + model=self.config.text_model, contents=prompt, config=config ) return response.text - + def _supports_thinking(self, model_name: str) -> bool: """检查模型是否支持思考功能""" if not model_name: return False - + model_lower = model_name.lower() - + # 支持思考功能的模型模式 thinking_patterns = [ - "gemini-3", # gemini-3-xxx - "gemini-4", # gemini-4-xxx (未来版本) - "gemini-5", # gemini-5-xxx (未来版本) + "gemini-3", # gemini-3-xxx + "gemini-4", # gemini-4-xxx (未来版本) + "gemini-5", # gemini-5-xxx (未来版本) ] - + # 检查是否匹配支持思考的模型 for pattern in thinking_patterns: if pattern in model_lower: return True - + return False - + def _generate_text_openai(self, prompt: str, system_instruction: str = None) -> str: """使用 OpenAI 格式 API 生成文本""" messages = [] - + if system_instruction: messages.append({"role": "system", "content": system_instruction}) - + messages.append({"role": "user", "content": prompt}) - + kwargs = { "model": self.config.text_model, "messages": messages, @@ -104,109 +100,107 @@ def _generate_text_openai(self, prompt: str, system_instruction: str = None) -> kwargs["extra_body"] = {"thinking": {"type": self.config.text_thinking}} response = self._openai_client.chat.completions.create(**kwargs) - + return response.choices[0].message.content - + def generate_image( self, prompt: str, aspect_ratio: str = "16:9", quality: str = "2K", - output_path: str = "output.png" + output_path: str = "output.png", ) -> str: """文生图（使用 Gemini API）""" generation_config = self._build_image_config(aspect_ratio, quality) full_prompt = f"{PromptTemplates.get_image_generation_prefix()}{prompt}" - + # 在API调用级别控制超时，而不是任务级别 try: response = self._gemini_client.models.generate_content( - model=self.config.image_model, - contents=[full_prompt], - config=generation_config + model=self.config.image_model, contents=[full_prompt], config=generation_config ) except Exception as e: # 将API级别的超时转换为普通异常，让重试机制处理 raise Exception(f"API调用失败: {e}") - + return self._save_image_from_response(response, output_path) - + def edit_image( self, prompt: str, source_image_path: str, aspect_ratio: str = "16:9", quality: str = "2K", - output_path: str = "output.png" + output_path: str = "output.png", ) -> str: """图生图（使用 Gemini API）""" source_image = Image.open(source_image_path) generation_config = self._build_image_config(aspect_ratio, quality) full_prompt = f"{PromptTemplates.get_image_generation_prefix()}{prompt}" - + response = self._gemini_client.models.generate_content( model=self.config.image_model, contents=[full_prompt, source_image], - config=generation_config + config=generation_config, ) - + return self._save_image_from_response(response, output_path) - + def _build_image_config(self, aspect_ratio: str, quality: str) -> types.GenerateContentConfig: """构建图片生成配置""" generation_config = types.GenerateContentConfig(response_modalities=["TEXT", "IMAGE"]) - + # 构建 ImageConfig 对象而不是字典，避免 Pydantic 警告 image_config_params = {} if aspect_ratio: image_config_params["aspect_ratio"] = aspect_ratio if quality and quality != "auto": image_config_params["image_size"] = quality - + if image_config_params: generation_config.image_config = types.ImageConfig(**image_config_params) - + return generation_config - + def _save_image_from_response(self, response, output_path: str) -> str: """从响应中提取并保存图片，支持二进制和 Markdown 链接两种格式""" import re import requests from pathlib import Path - + text_content = [] - + for part in response.parts: # 方式1: 二进制图片数据 if part.inline_data is not None: image = part.as_image() image.save(output_path) return output_path - + # 方式2: 文本内容（可能包含 Markdown 链接） if part.text is not None: text_content.append(part.text) - + # 如果没有二进制数据，尝试从文本中提取图片链接 if text_content: - full_text = ' '.join(text_content) - + full_text = " ".join(text_content) + # 提取 Markdown 格式的图片链接: ![alt](url) - markdown_pattern = r'!\[.*?\]\((https?://[^\)]+)\)' + markdown_pattern = r"!\[.*?\]\((https?://[^\)]+)\)" matches = re.findall(markdown_pattern, full_text) - + if matches: image_url = matches[0] try: resp = requests.get(image_url, timeout=30) resp.raise_for_status() Path(output_path).parent.mkdir(parents=True, exist_ok=True) - with open(output_path, 'wb') as f: + with open(output_path, "wb") as f: f.write(resp.content) return output_path except Exception as e: raise Exception(f"下载图片失败: {e}") - + raise Exception(f"未能生成图片，返回文本: {full_text[:200]}...") - + raise Exception("未能生成图片：响应中没有图片数据") diff --git a/src/config.py b/src/config.py index f88eae9..d3ac1f3 100644 --- a/src/config.py +++ b/src/config.py @@ -3,7 +3,6 @@ 定义所有配置类，支持从 YAML 文件加载 """ -import os from dataclasses import dataclass, asdict from typing import Dict, Any, Literal, Optional from pathlib import Path @@ -18,7 +17,7 @@ def load_yaml_config() -> Dict[str, Any]: """加载 YAML 配置文件""" if CONFIG_FILE.exists(): - with open(CONFIG_FILE, 'r', encoding='utf-8') as f: + with open(CONFIG_FILE, "r", encoding="utf-8") as f: return yaml.safe_load(f) or {} return {} @@ -45,23 +44,24 @@ def reload_config(): @dataclass class APIConfig: """API 配置""" + # 图像生成 API 配置 image_api_key: str = None image_base_url: str = None image_model: str = None - + # 文本生成 API 配置 text_api_format: Literal["gemini", "openai"] = None # API 格式 text_model: str = None text_base_url: str = None # 可选，单独的文本 API 地址 - text_api_key: str = None # 可选，单独的文本 API 密钥 + text_api_key: str = None # 可选，单独的文本 API 密钥 text_thinking: Optional[Literal["enabled", "disabled"]] = None # OpenAI-compatible 思考模式 text_thinking_level: Optional[Literal["low", "high"]] = None # 旧版 Gemini 思考深度 - + def __post_init__(self): """从配置文件加载默认值""" config = get_config().get("api", {}) - + # 图像生成 API 配置 image_config = config.get("image", {}) if self.image_api_key is None: @@ -70,7 +70,7 @@ def __post_init__(self): self.image_base_url = image_config.get("base_url", "") if self.image_model is None: self.image_model = image_config.get("model", "gemini-3-pro-image-preview") - + # 文本生成 API 配置 text_config = config.get("text", {}) if self.text_api_format is None: @@ -85,18 +85,18 @@ def __post_init__(self): self.text_thinking = text_config.get("thinking", "disabled") if self.text_thinking_level is None: self.text_thinking_level = text_config.get("thinking_level") # 可选 - + # 如果文本 API 没有单独配置，使用图像 API 的配置 if not self.text_api_key: self.text_api_key = self.image_api_key if not self.text_base_url: self.text_base_url = self.image_base_url - + @property def api_key(self) -> str: """向后兼容：返回图像 API 密钥""" return self.image_api_key - + @property def base_url(self) -> str: """向后兼容：返回图像 API 地址""" @@ -106,6 +106,7 @@ def base_url(self) -> str: @dataclass class PPTConfig: """PPT 生成配置""" + language: str = None style: str = None target_audience: str = None @@ -116,7 +117,7 @@ class PPTConfig: max_concurrent: int = None max_retries: int = None retry_delay: float = None - + def __post_init__(self): """从配置文件加载默认值""" config = get_config().get("ppt", {}) @@ -140,13 +141,13 @@ def __post_init__(self): self.max_retries = config.get("max_retries", 3) if self.retry_delay is None: self.retry_delay = config.get("retry_delay", 2.0) - + def to_dict(self) -> Dict[str, Any]: """转换为字典""" return asdict(self) - + @classmethod - def from_dict(cls, data: Dict[str, Any]) -> 'PPTConfig': + def from_dict(cls, data: Dict[str, Any]) -> "PPTConfig": """从字典创建""" return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__}) @@ -177,7 +178,7 @@ def get_timeout_config() -> Dict[str, int]: "text_generation": config.get("text_generation", 120), "image_generation": config.get("image_generation", 180), "api_call": config.get("api_call", 60), - "batch_buffer": config.get("batch_buffer", 60) + "batch_buffer": config.get("batch_buffer", 60), } @@ -186,6 +187,6 @@ def load_sample_material() -> str: sample_path = get_sample_file() if not sample_path.exists(): raise FileNotFoundError(f"示例文件不存在: {sample_path}") - - with open(sample_path, 'r', encoding='utf-8') as f: + + with open(sample_path, "r", encoding="utf-8") as f: return f.read() diff --git a/src/config_writer.py b/src/config_writer.py index 2e37e1a..474d0bc 100644 --- a/src/config_writer.py +++ b/src/config_writer.py @@ -10,7 +10,9 @@ from .config import CONFIG_FILE, load_yaml_config, reload_config -def save_model_profiles_to_config(profile_data: Dict[str, Any], config_path: Path = CONFIG_FILE) -> Dict[str, Any]: +def save_model_profiles_to_config( + profile_data: Dict[str, Any], config_path: Path = CONFIG_FILE +) -> Dict[str, Any]: config = load_yaml_config() config.setdefault("api", {}) existing_models = config.get("api", {}).get("models", {}) @@ -24,22 +26,30 @@ def save_model_profiles_to_config(profile_data: Dict[str, Any], config_path: Pat return config -def _clean_profile_data(profile_data: Dict[str, Any], existing_models: Dict[str, Any]) -> Dict[str, Any]: +def _clean_profile_data( + profile_data: Dict[str, Any], existing_models: Dict[str, Any] +) -> Dict[str, Any]: cleaned: Dict[str, Any] = {} for role in ("prompt_model", "image_model", "edit_model"): profile = profile_data.get(role) if not profile: continue - existing_profile = existing_models.get(role, {}) if isinstance(existing_models, dict) else {} + existing_profile = ( + existing_models.get(role, {}) if isinstance(existing_models, dict) else {} + ) api_key = profile.get("api_key") or existing_profile.get("api_key", "") cleaned[role] = { "model": profile.get("model", ""), "base_url": profile.get("base_url", ""), "api_key": api_key, - "adapter": profile.get("adapter", "openai_chat" if role == "prompt_model" else "raw_chat_multimodal"), + "adapter": profile.get( + "adapter", "openai_chat" if role == "prompt_model" else "raw_chat_multimodal" + ), } if role == "prompt_model": - cleaned[role]["thinking"] = profile.get("thinking", existing_profile.get("thinking", "disabled")) + cleaned[role]["thinking"] = profile.get( + "thinking", existing_profile.get("thinking", "disabled") + ) if profile.get("id"): cleaned[role]["id"] = profile["id"] if profile.get("label"): diff --git a/src/exporter.py b/src/exporter.py index 1733c39..76b1089 100644 --- a/src/exporter.py +++ b/src/exporter.py @@ -5,92 +5,86 @@ import os from pathlib import Path -from typing import List, Optional +from typing import List from PIL import Image class PDFExporter: """PDF 导出器""" - + @staticmethod - def export( - image_paths: List[str], - output_path: str, - title: str = "AI Generated PPT" - ) -> str: + def export(image_paths: List[str], output_path: str, title: str = "AI Generated PPT") -> str: """ 将多张图片合并为 PDF - + Args: image_paths: 图片路径列表（按顺序） output_path: 输出 PDF 路径 title: PDF 标题 - + Returns: 输出 PDF 路径 """ if not image_paths: raise ValueError("图片列表为空") - + # 过滤有效图片 valid_paths = [p for p in image_paths if p and os.path.exists(p)] - + if not valid_paths: raise ValueError("没有有效的图片") - + # 加载所有图片 images = [] for path in valid_paths: img = Image.open(path) # 转换为 RGB（PDF 不支持 RGBA） - if img.mode != 'RGB': - img = img.convert('RGB') + if img.mode != "RGB": + img = img.convert("RGB") images.append(img) - + # 保存为 PDF first_image = images[0] other_images = images[1:] if len(images) > 1 else [] - + first_image.save( output_path, "PDF", resolution=100.0, save_all=True, append_images=other_images, - title=title + title=title, ) - + print(f"✅ PDF 已生成: {output_path}") print(f" 共 {len(images)} 页") - + return output_path - + @staticmethod def export_from_directory( - images_dir: str, - output_path: str, - pattern: str = "slide_*.png" + images_dir: str, output_path: str, pattern: str = "slide_*.png" ) -> str: """ 从目录导出 PDF - + Args: images_dir: 图片目录 output_path: 输出 PDF 路径 pattern: 文件匹配模式 - + Returns: 输出 PDF 路径 """ images_path = Path(images_dir) - + if not images_path.exists(): raise FileNotFoundError(f"目录不存在: {images_dir}") - + # 按文件名排序 image_paths = sorted([str(p) for p in images_path.glob(pattern)]) - + if not image_paths: raise ValueError(f"目录中没有匹配的图片: {pattern}") - + return PDFExporter.export(image_paths, output_path) diff --git a/src/generator.py b/src/generator.py index 5354148..f7f7624 100644 --- a/src/generator.py +++ b/src/generator.py @@ -5,7 +5,6 @@ from pathlib import Path from datetime import datetime -from typing import Optional from .config import PPTConfig, APIConfig from .client import AIClient @@ -17,85 +16,77 @@ class PPTGenerator: """PPT 生成器主类""" - - def __init__( - self, - api_config: APIConfig = None, - output_dir: str = None - ): + + def __init__(self, api_config: APIConfig = None, output_dir: str = None): """ 初始化生成器 - + Args: api_config: API 配置 output_dir: 输出目录（默认从配置文件读取） """ from .config import get_output_dir - + self.api_config = api_config or APIConfig() self.output_dir = Path(output_dir or get_output_dir()) self.output_dir.mkdir(parents=True, exist_ok=True) - + # 初始化客户端和子模块 self.client = AIClient(self.api_config) self.prompt_generator = PromptGenerator(self.client) self.image_generator = ImageGenerator(self.client) self.pdf_exporter = PDFExporter() - + def generate( - self, - source_material: str, - ppt_config: PPTConfig = None, - export_pdf: bool = True + self, source_material: str, ppt_config: PPTConfig = None, export_pdf: bool = True ) -> GenerationResult: """ 生成 PPT - + Args: source_material: 输入资料 ppt_config: PPT 配置 export_pdf: 是否导出 PDF - + Returns: GenerationResult 对象 """ if ppt_config is None: ppt_config = PPTConfig() - + # 创建项目目录 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") project_dir = self.output_dir / f"ppt_{timestamp}" project_dir.mkdir(parents=True, exist_ok=True) images_dir = project_dir / "images" images_dir.mkdir(exist_ok=True) - + self._print_header(project_dir, ppt_config) - + result = GenerationResult( - project_dir=str(project_dir), - created_at=datetime.now().isoformat() + project_dir=str(project_dir), created_at=datetime.now().isoformat() ) - + try: # Step 1: 保存输入资料 print("\n【Step 1】保存输入资料...") self._save_source_material(source_material, project_dir) - + # Step 2: 生成 Prompt print("\n【Step 2】生成 Prompt...") prompt_data = self.prompt_generator.generate(source_material, ppt_config) prompt_data.save(str(project_dir / "prompts.json")) result.prompt_data = prompt_data - + # Step 3: 生成图片 print("\n【Step 3】并行生成 PPT 页面...") slide_images = self.image_generator.generate_slides( slide_prompts=prompt_data.slide_prompts, output_dir=str(images_dir), - config=ppt_config + config=ppt_config, ) result.slide_image_paths = slide_images - + # Step 4: 导出 PDF if export_pdf: print("\n【Step 4】导出 PDF...") @@ -103,98 +94,94 @@ def generate( if valid_images: pdf_path = str(project_dir / "presentation.pdf") self.pdf_exporter.export(valid_images, pdf_path) - + # 保存结果 result.success = True result.save(str(project_dir / "result.json")) - + self._print_footer(result) - + except Exception as e: result.success = False result.error_message = str(e) result.save(str(project_dir / "result.json")) print(f"\n❌ 生成失败: {e}") - + return result - + def generate_prompts_only( - self, - source_material: str, - ppt_config: PPTConfig = None, - output_path: str = None + self, source_material: str, ppt_config: PPTConfig = None, output_path: str = None ) -> PromptData: """ 仅生成 Prompt（不生成图片） - + Args: source_material: 输入资料 ppt_config: PPT 配置 output_path: 输出路径（可选） - + Returns: PromptData 对象 """ if ppt_config is None: ppt_config = PPTConfig() - + print("=" * 60) print("🎨 生成 PPT Prompt") print("=" * 60) - + prompt_data = self.prompt_generator.generate(source_material, ppt_config) - + if output_path: prompt_data.save(output_path) - + return prompt_data - + def generate_from_prompts( - self, - prompt_data: PromptData, - ppt_config: PPTConfig = None, - export_pdf: bool = True + self, prompt_data: PromptData, ppt_config: PPTConfig = None, export_pdf: bool = True ) -> GenerationResult: """ 从已有的 Prompt 生成图片 - + Args: prompt_data: Prompt 数据 ppt_config: PPT 配置 export_pdf: 是否导出 PDF - + Returns: GenerationResult 对象 """ if ppt_config is None: - ppt_config = PPTConfig.from_dict(prompt_data.config) if prompt_data.config else PPTConfig() - + ppt_config = ( + PPTConfig.from_dict(prompt_data.config) if prompt_data.config else PPTConfig() + ) + # 创建项目目录 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") project_dir = self.output_dir / f"ppt_{timestamp}" project_dir.mkdir(parents=True, exist_ok=True) images_dir = project_dir / "images" images_dir.mkdir(exist_ok=True) - + result = GenerationResult( project_dir=str(project_dir), prompt_data=prompt_data, - created_at=datetime.now().isoformat() + created_at=datetime.now().isoformat(), ) - + try: # 保存 Prompt prompt_data.save(str(project_dir / "prompts.json")) - + # 并行生成每页 PPT print("\n[Step 1] 并行生成 PPT 页面...") slide_images = self.image_generator.generate_slides( slide_prompts=prompt_data.slide_prompts, output_dir=str(images_dir), - config=ppt_config + config=ppt_config, ) result.slide_image_paths = slide_images - + # 导出 PDF if export_pdf: print("\n[Step 2] 导出 PDF...") @@ -202,24 +189,24 @@ def generate_from_prompts( if valid_images: pdf_path = str(project_dir / "presentation.pdf") self.pdf_exporter.export(valid_images, pdf_path) - + result.success = True result.save(str(project_dir / "result.json")) - + except Exception as e: result.success = False result.error_message = str(e) print(f"\n❌ 生成失败: {e}") - + return result - + def _save_source_material(self, material: str, project_dir: Path): """保存输入资料""" material_path = project_dir / "source_material.txt" - with open(material_path, 'w', encoding='utf-8') as f: + with open(material_path, "w", encoding="utf-8") as f: f.write(material) print(f"✅ 资料已保存到: {material_path}") - + def _print_header(self, project_dir: Path, config: PPTConfig): """打印头部信息""" print("=" * 60) @@ -233,12 +220,12 @@ def _print_header(self, project_dir: Path, config: PPTConfig): print(f"📐 比例: {config.aspect_ratio}") print(f"✨ 质量: {config.quality}") print("=" * 60) - + def _print_footer(self, result: GenerationResult): """打印尾部信息""" valid_count = len([p for p in result.slide_image_paths if p]) total_count = len(result.slide_image_paths) - + print("\n" + "=" * 60) print("🎉 PPT 生成完成！") print(f"📁 输出目录: {result.project_dir}") diff --git a/src/image_generator.py b/src/image_generator.py index 8fa0ef8..bcf9028 100644 --- a/src/image_generator.py +++ b/src/image_generator.py @@ -10,78 +10,77 @@ from .client import AIClient from .config import PPTConfig, get_timeout_config -from .models import PromptData, SlidePrompt +from .models import SlidePrompt class ImageGenerator: """图片生成器""" - + def __init__(self, client: AIClient): """ 初始化 - + Args: client: AI 客户端 """ self.client = client - + def generate_slides( - self, - slide_prompts: List[SlidePrompt], - output_dir: str, - config: PPTConfig + self, slide_prompts: List[SlidePrompt], output_dir: str, config: PPTConfig ) -> List[str]: """ 并行生成所有 PPT 页面图片（文生图） - + Args: slide_prompts: 每页的 Prompt 列表 output_dir: 输出目录 config: PPT 配置 - + Returns: 生成的图片路径列表（按顺序） """ print(f"🎨 开始并行生成 {len(slide_prompts)} 页 PPT...") print(f" 并发数: {config.max_concurrent}, 重试次数: {config.max_retries}") - + output_path = Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) - + # 结果列表，保持顺序 results = [None] * len(slide_prompts) - + # 使用线程池并行生成 with ThreadPoolExecutor(max_workers=config.max_concurrent) as executor: futures = {} - + for i, slide in enumerate(slide_prompts): future = executor.submit( self._generate_single_slide, index=i, slide=slide, output_dir=output_path, - config=config + config=config, ) futures[future] = i - + # 收集结果（双层超时控制） timeout_config = get_timeout_config() - single_timeout = timeout_config['image_generation'] - + single_timeout = timeout_config["image_generation"] + # 计算全局超时（用于整体进度监控） import math + max_batches = math.ceil(len(slide_prompts) / config.max_concurrent) theoretical_time = max_batches * single_timeout - global_timeout = theoretical_time + timeout_config['batch_buffer'] - + global_timeout = theoretical_time + timeout_config["batch_buffer"] + print(f" 超时设置: 单个任务{single_timeout}s, 全局超时{global_timeout}s") - + # 收集结果（全局超时控制） import time + start_time = time.time() completed_count = 0 - + for future in as_completed(futures): # 检查全局超时 elapsed = time.time() - start_time @@ -92,71 +91,69 @@ def generate_slides( if not remaining_future.done(): remaining_future.cancel() break - + index = futures[future] completed_count += 1 - + try: # 设置单个任务超时，避免模型响应错误时卡死 result_index, result_path, error = future.result(timeout=single_timeout) if result_path: results[result_index] = result_path - print(f" ✅ 第 {index + 1} 页完成 ({completed_count}/{len(slide_prompts)})") + print( + f" ✅ 第 {index + 1} 页完成 ({completed_count}/{len(slide_prompts)})" + ) else: print(f"❌ 第 {index + 1} 页最终失败: {error}") except TimeoutError: print(f"❌ 第 {index + 1} 页任务超时 ({single_timeout}s)") except Exception as e: print(f"❌ 第 {index + 1} 页异常: {e}") - + # 统计结果 valid_results = [r for r in results if r is not None] print(f"\n✅ 成功生成 {len(valid_results)}/{len(slide_prompts)} 页") - + return results - + def _generate_single_slide( - self, - index: int, - slide: SlidePrompt, - output_dir: Path, - config: PPTConfig + self, index: int, slide: SlidePrompt, output_dir: Path, config: PPTConfig ) -> Tuple[int, Optional[str], Optional[str]]: """ 生成单页 PPT（文生图） - + Args: index: 页面索引 slide: 页面 Prompt output_dir: 输出目录 config: 配置 - + Returns: (索引, 图片路径或None, 错误信息或None) """ page = slide.page output_path = str(output_dir / f"slide_{page:03d}.png") - + for attempt in range(config.max_retries): try: print(f" 📄 生成第 {page} 页 (尝试 {attempt + 1}/{config.max_retries})...") - + # 使用文生图直接生成 self.client.generate_image( prompt=slide.prompt, aspect_ratio=config.aspect_ratio, quality=config.quality, - output_path=output_path + output_path=output_path, ) - + print(f" ✅ 第 {page} 页生成成功") return (index, output_path, None) - + except Exception as e: error_msg = str(e) print(f" ⚠️ 第 {page} 页尝试 {attempt + 1} 失败: {error_msg}") - + if attempt < config.max_retries - 1: time.sleep(config.retry_delay) - + return (index, None, f"重试 {config.max_retries} 次后仍失败") diff --git a/src/model_profiles.py b/src/model_profiles.py index 44dc278..b9bd30f 100644 --- a/src/model_profiles.py +++ b/src/model_profiles.py @@ -131,7 +131,9 @@ def load_profiles_from_env(env_path: Optional[Path] = None) -> Optional[ModelPro ) -def load_default_profiles(config_data: Optional[Dict[str, Any]] = None) -> Optional[ModelProfileSet]: +def load_default_profiles( + config_data: Optional[Dict[str, Any]] = None, +) -> Optional[ModelProfileSet]: """Load profiles from the repository-local config.yaml data.""" if config_data is None: try: @@ -210,7 +212,7 @@ def _parse_env_like_file(path: Path) -> Dict[str, str]: if not line or line.startswith("#"): continue if line.startswith("export "): - line = line[len("export "):].strip() + line = line[len("export ") :].strip() if "=" in line: key, value = line.split("=", 1) elif ":" in line: diff --git a/src/model_router.py b/src/model_router.py index 66709aa..2b6066c 100644 --- a/src/model_router.py +++ b/src/model_router.py @@ -76,9 +76,13 @@ def edit_image( result = self._post_chat(profile, payload) return self._write_normalized_image(result, output_path) - def generate_image_base64(self, prompt: str, aspect_ratio: str = "16:9", quality: str = "2K") -> str: + def generate_image_base64( + self, prompt: str, aspect_ratio: str = "16:9", quality: str = "2K" + ) -> str: profile = self.profiles.image - result = self._post_chat(profile, self._chat_image_payload(profile, prompt, aspect_ratio, quality)) + result = self._post_chat( + profile, self._chat_image_payload(profile, prompt, aspect_ratio, quality) + ) return self.normalizer.normalize(result).base64_data def edit_image_base64( diff --git a/src/models.py b/src/models.py index 24f6c6d..822cb0e 100644 --- a/src/models.py +++ b/src/models.py @@ -6,8 +6,6 @@ import json from dataclasses import dataclass, field, asdict from typing import List, Dict, Any, Optional -from datetime import datetime -from pathlib import Path from pydantic import BaseModel, Field @@ -15,22 +13,24 @@ @dataclass class SlidePrompt: """单页 PPT 的 Prompt""" + page: int title: str content_summary: str prompt: str display_content: str = "" - + def to_dict(self) -> Dict[str, Any]: return asdict(self) - + @classmethod - def from_dict(cls, data: Dict[str, Any]) -> 'SlidePrompt': + def from_dict(cls, data: Dict[str, Any]) -> "SlidePrompt": return cls(**data) class SlideOutline(BaseModel): """用户可审阅的单页设计大纲""" + page: int = Field(..., ge=1) title: str = Field(..., min_length=1) narrative_goal: str = Field(..., min_length=1) @@ -40,6 +40,7 @@ class SlideOutline(BaseModel): class DeckOutline(BaseModel): """整套 PPT 的设计大纲""" + title: str = Field(..., min_length=1) user_requirements: str = Field("", description="已吸收的用户定制需求") design_style: str = Field(..., min_length=1) @@ -49,6 +50,7 @@ class DeckOutline(BaseModel): class SlidePromptPlan(BaseModel): """用户可确认的单页设计和内部图像 prompt""" + page: int = Field(..., ge=1) title: str = Field(..., min_length=1) content_summary: str = Field(..., min_length=1) @@ -58,75 +60,80 @@ class SlidePromptPlan(BaseModel): class SlidePromptPlanSet(BaseModel): """逐页设计和 prompt 结构化结果""" + slide_prompts: List[SlidePromptPlan] = Field(default_factory=list) @dataclass class PromptData: """所有 Prompt 数据""" + slide_prompts: List[SlidePrompt] = field(default_factory=list) created_at: str = "" config: Dict[str, Any] = field(default_factory=dict) - + # 保存原始输入，用于后续检查 source_material: str = "" user_requirements: str = "" - + def to_dict(self) -> Dict[str, Any]: return { - "slide_prompts": [s.to_dict() if isinstance(s, SlidePrompt) else s for s in self.slide_prompts], + "slide_prompts": [ + s.to_dict() if isinstance(s, SlidePrompt) else s for s in self.slide_prompts + ], "created_at": self.created_at, "config": self.config, "source_material": self.source_material, - "user_requirements": self.user_requirements + "user_requirements": self.user_requirements, } - + def save(self, filepath: str): """保存到 JSON 文件""" - with open(filepath, 'w', encoding='utf-8') as f: + with open(filepath, "w", encoding="utf-8") as f: json.dump(self.to_dict(), f, ensure_ascii=False, indent=2) print(f"✅ Prompt 数据已保存到: {filepath}") - + @classmethod - def load(cls, filepath: str) -> 'PromptData': + def load(cls, filepath: str) -> "PromptData": """从 JSON 文件加载""" - with open(filepath, 'r', encoding='utf-8') as f: + with open(filepath, "r", encoding="utf-8") as f: data = json.load(f) - + slide_prompts = [ - SlidePrompt.from_dict(s) if isinstance(s, dict) else s + SlidePrompt.from_dict(s) if isinstance(s, dict) else s for s in data.get("slide_prompts", []) ] - + return cls( slide_prompts=slide_prompts, created_at=data.get("created_at", ""), config=data.get("config", {}), source_material=data.get("source_material", ""), - user_requirements=data.get("user_requirements", "") + user_requirements=data.get("user_requirements", ""), ) @dataclass class GenerationResult: """生成结果""" + project_dir: str slide_image_paths: List[str] = field(default_factory=list) prompt_data: Optional[PromptData] = None created_at: str = "" success: bool = True error_message: str = "" - + def to_dict(self) -> Dict[str, Any]: return { "project_dir": self.project_dir, "slide_image_paths": self.slide_image_paths, "created_at": self.created_at, "success": self.success, - "error_message": self.error_message + "error_message": self.error_message, } - + def save(self, filepath: str): """保存结果到 JSON""" - with open(filepath, 'w', encoding='utf-8') as f: + with open(filepath, "w", encoding="utf-8") as f: json.dump(self.to_dict(), f, ensure_ascii=False, indent=2) diff --git a/src/prompt_generator.py b/src/prompt_generator.py index 0cabef2..52d6b29 100644 --- a/src/prompt_generator.py +++ b/src/prompt_generator.py @@ -16,10 +16,10 @@ class PromptGenerator: """Prompt 生成器""" - + def __init__(self, client: AIClient): self.client = client - + def generate(self, source_material: str, config: PPTConfig) -> PromptData: """生成所有 Prompt""" user_requirements = self._build_user_requirements(config) @@ -34,7 +34,7 @@ def generate(self, source_material: str, config: PPTConfig) -> PromptData: final_prompts = self._review_and_optimize_prompts( initial_prompts, source_material, user_requirements, config ) - + return final_prompts def generate_outline(self, source_material: str, config: PPTConfig) -> DeckOutline: @@ -73,7 +73,9 @@ def generate_prompts_from_outline( self._validate_outline(outline_model, config) outline_json = self._model_to_json(outline_model) - system_instruction = PromptTemplates.get_prompts_from_outline_system(user_requirements, config.num_pages) + system_instruction = PromptTemplates.get_prompts_from_outline_system( + user_requirements, config.num_pages + ) base_user_prompt = PromptTemplates.get_prompts_from_outline_user( source_material, outline_json, @@ -88,7 +90,9 @@ def generate_prompts_from_outline( response = self.client.generate_text(user_prompt, system_instruction) prompt_plan = self._parse_model_response(response, SlidePromptPlanSet) self._validate_prompt_plan(prompt_plan, config) - prompt_data = self._prompt_plan_to_prompt_data(prompt_plan, source_material, user_requirements, config) + prompt_data = self._prompt_plan_to_prompt_data( + prompt_plan, source_material, user_requirements, config + ) print(f"✅ 逐页 Prompt 生成完成，共 {len(prompt_data.slide_prompts)} 页") return prompt_data except Exception as e: @@ -98,7 +102,7 @@ def generate_prompts_from_outline( raise Exception(f"逐页设计生成失败，重试 3 次后仍失败: {last_error}") raise RuntimeError("逐页设计生成异常") - + def _build_user_requirements(self, config: PPTConfig) -> str: """构建用户需求描述""" custom = (config.user_requirements or "").strip() @@ -109,133 +113,149 @@ def _build_user_requirements(self, config: PPTConfig) -> str: - 页数：{config.num_pages} 页 - 图片比例：{config.aspect_ratio} - 图片质量：{config.quality}{custom_line}""" - + def _generate_initial_prompts( self, source_material: str, user_requirements: str, config: PPTConfig ) -> PromptData: """生成初始 Prompt（带重试机制）""" - system_instruction = PromptTemplates.get_initial_prompt_system(user_requirements, config.num_pages) + system_instruction = PromptTemplates.get_initial_prompt_system( + user_requirements, config.num_pages + ) user_prompt = PromptTemplates.get_initial_prompt_user(source_material, config.num_pages) - + max_retries = 3 for attempt in range(max_retries): try: print(f" 尝试生成 Prompt ({attempt + 1}/{max_retries})...") response = self.client.generate_text(user_prompt, system_instruction) - + # 验证响应格式 prompt_data = self._parse_prompt_response(response, config) - + # 验证生成的页数是否正确 if len(prompt_data.slide_prompts) != config.num_pages: - raise ValueError(f"生成页数不匹配: 期望{config.num_pages}页，实际{len(prompt_data.slide_prompts)}页") - + raise ValueError( + f"生成页数不匹配: 期望{config.num_pages}页，实际{len(prompt_data.slide_prompts)}页" + ) + # 验证每页内容是否完整 for i, slide in enumerate(prompt_data.slide_prompts): if not slide.title or not slide.prompt: - raise ValueError(f"第{i+1}页内容不完整: title='{slide.title}', prompt长度={len(slide.prompt)}") - + raise ValueError( + f"第{i + 1}页内容不完整: title='{slide.title}', prompt长度={len(slide.prompt)}" + ) + prompt_data.source_material = source_material[:2000] prompt_data.user_requirements = user_requirements - + print(f"✅ 生成了 {len(prompt_data.slide_prompts)} 个页面 Prompt") return prompt_data - + except Exception as e: print(f" ⚠️ 第 {attempt + 1} 次尝试失败: {e}") if attempt == max_retries - 1: raise Exception(f"Prompt 生成失败，重试 {max_retries} 次后仍失败: {e}") - + import time + time.sleep(2) # 重试间隔 - + def _review_and_optimize_prompts( - self, initial_prompts: PromptData, source_material: str, user_requirements: str, config: PPTConfig + self, + initial_prompts: PromptData, + source_material: str, + user_requirements: str, + config: PPTConfig, ) -> PromptData: """检查和优化 Prompt（带重试机制）""" system_instruction = PromptTemplates.get_review_prompt_system() user_prompt = PromptTemplates.get_review_prompt_user( user_requirements, source_material, - json.dumps(initial_prompts.to_dict(), ensure_ascii=False, indent=2) + json.dumps(initial_prompts.to_dict(), ensure_ascii=False, indent=2), ) - + max_retries = 2 # 优化步骤重试次数 for attempt in range(max_retries): try: print(f" 尝试优化 Prompt ({attempt + 1}/{max_retries})...") response = self.client.generate_text(user_prompt, system_instruction) - + # 验证优化后的格式 optimized = self._parse_prompt_response(response, config) - + # 验证优化后的页数 if len(optimized.slide_prompts) != len(initial_prompts.slide_prompts): - raise ValueError(f"优化后页数变化: 原始{len(initial_prompts.slide_prompts)}页，优化后{len(optimized.slide_prompts)}页") - + raise ValueError( + f"优化后页数变化: 原始{len(initial_prompts.slide_prompts)}页，优化后{len(optimized.slide_prompts)}页" + ) + optimized.source_material = initial_prompts.source_material optimized.user_requirements = initial_prompts.user_requirements print(f"✅ Prompt 优化完成，共 {len(optimized.slide_prompts)} 页") return optimized - + except Exception as e: print(f" ⚠️ 优化第 {attempt + 1} 次尝试失败: {e}") if attempt == max_retries - 1: - print(f"⚠️ 优化失败，使用原始 Prompt") + print("⚠️ 优化失败，使用原始 Prompt") return initial_prompts - + import time + time.sleep(1) # 优化重试间隔短一些 - + def _parse_prompt_response(self, response: str, config: PPTConfig) -> PromptData: """解析 LLM 响应中的 JSON（增强错误处理）""" data = self._parse_json_payload(response) - + # 验证必要字段 if "slide_prompts" not in data: raise ValueError("响应中缺少 'slide_prompts' 字段") - + if not isinstance(data["slide_prompts"], list): raise ValueError("'slide_prompts' 必须是数组") - + slide_prompts = [] for i, item in enumerate(data["slide_prompts"]): if not isinstance(item, dict): - raise ValueError(f"第 {i+1} 个 slide_prompt 不是对象") - + raise ValueError(f"第 {i + 1} 个 slide_prompt 不是对象") + # 验证必要字段 required_fields = ["page", "title", "prompt"] for field in required_fields: if field not in item or not item[field]: - raise ValueError(f"第 {i+1} 个 slide_prompt 缺少或为空: {field}") - - slide_prompts.append(SlidePrompt( - page=item.get("page", i + 1), - title=item.get("title", ""), - content_summary=item.get("content_summary", ""), - prompt=item.get("prompt", "") - )) - + raise ValueError(f"第 {i + 1} 个 slide_prompt 缺少或为空: {field}") + + slide_prompts.append( + SlidePrompt( + page=item.get("page", i + 1), + title=item.get("title", ""), + content_summary=item.get("content_summary", ""), + prompt=item.get("prompt", ""), + ) + ) + return PromptData( slide_prompts=slide_prompts, created_at=datetime.now().isoformat(), - config=config.to_dict() + config=config.to_dict(), ) def _parse_json_payload(self, response: str) -> Any: """从模型响应中提取 JSON 对象""" json_str = None - json_match = re.search(r'```json\s*(\{.*?\})\s*```', response, re.DOTALL) + json_match = re.search(r"```json\s*(\{.*?\})\s*```", response, re.DOTALL) if json_match: json_str = json_match.group(1) else: - code_match = re.search(r'```\s*(\{.*?\})\s*```', response, re.DOTALL) + code_match = re.search(r"```\s*(\{.*?\})\s*```", response, re.DOTALL) if code_match: json_str = code_match.group(1) else: - json_start = response.find('{') - json_end = response.rfind('}') + 1 + json_start = response.find("{") + json_end = response.rfind("}") + 1 if json_start != -1 and json_end > json_start: json_str = response[json_start:json_end] @@ -279,7 +299,9 @@ def _with_correction(self, user_prompt: str, last_error: str) -> str: def _validate_outline(self, outline: DeckOutline, config: PPTConfig) -> None: if len(outline.slides) != config.num_pages: - raise ValueError(f"大纲页数不匹配: 期望{config.num_pages}页，实际{len(outline.slides)}页") + raise ValueError( + f"大纲页数不匹配: 期望{config.num_pages}页，实际{len(outline.slides)}页" + ) pages = [slide.page for slide in outline.slides] expected = list(range(1, config.num_pages + 1)) diff --git a/src/prompts/templates.py b/src/prompts/templates.py index a3217d8..21f475f 100644 --- a/src/prompts/templates.py +++ b/src/prompts/templates.py @@ -85,7 +85,9 @@ def get_prompts_from_outline_system(user_requirements: str, num_pages: int) -> s """ @staticmethod - def get_prompts_from_outline_user(source_material: str, outline_json: str, num_pages: int) -> str: + def get_prompts_from_outline_user( + source_material: str, outline_json: str, num_pages: int + ) -> str: """根据确认大纲生成逐页 prompt 的用户提示""" return f"""请根据已确认的大纲生成 {num_pages} 页 PPT 的逐页设计说明和图像 prompt。 @@ -97,7 +99,7 @@ def get_prompts_from_outline_user(source_material: str, outline_json: str, num_p 请严格输出 JSON。 """ - + @staticmethod def get_initial_prompt_system(user_requirements: str, num_pages: int) -> str: """生成初始 Prompt 的系统指令""" @@ -208,9 +210,7 @@ def get_review_prompt_system() -> str: @staticmethod def get_review_prompt_user( - user_requirements: str, - source_material: str, - prompts_json: str + user_requirements: str, source_material: str, prompts_json: str ) -> str: """检查优化 Prompt 的用户提示""" return f"""请检查并优化以下 Prompt： diff --git a/tests/test_config_writer.py b/tests/test_config_writer.py index 48616b3..bfbe5c1 100644 --- a/tests/test_config_writer.py +++ b/tests/test_config_writer.py @@ -13,8 +13,10 @@ def test_saves_model_profiles_to_config_yaml(self): with tempfile.TemporaryDirectory() as temp_dir: config_path = Path(temp_dir) / "config.yaml" - with patch("src.config_writer.load_yaml_config", return_value={"ppt": {"num_pages": 3}}), \ - patch("src.config_writer.reload_config"): + with ( + patch("src.config_writer.load_yaml_config", return_value={"ppt": {"num_pages": 3}}), + patch("src.config_writer.reload_config"), + ): save_model_profiles_to_config( { "prompt_model": { @@ -44,14 +46,20 @@ def test_empty_api_key_preserves_existing_secret(self): with tempfile.TemporaryDirectory() as temp_dir: config_path = Path(temp_dir) / "config.yaml" - with patch("src.config_writer.load_yaml_config", return_value={ - "api": { - "models": { - "prompt_model": {"api_key": "old-text-key"}, - "image_model": {"api_key": "old-image-key"}, - } - } - }), patch("src.config_writer.reload_config"): + with ( + patch( + "src.config_writer.load_yaml_config", + return_value={ + "api": { + "models": { + "prompt_model": {"api_key": "old-text-key"}, + "image_model": {"api_key": "old-image-key"}, + } + } + }, + ), + patch("src.config_writer.reload_config"), + ): save_model_profiles_to_config( { "prompt_model": { diff --git a/tests/test_edit_route.py b/tests/test_edit_route.py new file mode 100644 index 0000000..a093a85 --- /dev/null +++ b/tests/test_edit_route.py @@ -0,0 +1,71 @@ +import asyncio +import base64 +from pathlib import Path + +import pytest +from fastapi import HTTPException + +from api.models import EditConfig, EditRequest +import api.routes.edit as edit_route + + +def test_edit_route_cleans_input_temp_file_when_output_temp_creation_fails(monkeypatch): + created_inputs = [] + original_named_temporary_file = edit_route.tempfile.NamedTemporaryFile + call_count = 0 + + def failing_second_named_temporary_file(*args, **kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + temp_file = original_named_temporary_file(*args, **kwargs) + created_inputs.append(Path(temp_file.name)) + return temp_file + raise RuntimeError("output temp failed") + + monkeypatch.setattr( + edit_route.tempfile, "NamedTemporaryFile", failing_second_named_temporary_file + ) + + request = EditRequest( + image_base64=base64.b64encode(b"image").decode(), + instruction="make it blue", + config=EditConfig(api_key="key", base_url="https://example.test/v1"), + ) + + with pytest.raises(HTTPException): + asyncio.run(edit_route.edit_image(request)) + + assert created_inputs + assert all(not path.exists() for path in created_inputs) + + +def test_edit_route_runs_model_edit_in_thread(monkeypatch): + thread_calls = [] + + async def fake_to_thread(func, *args, **kwargs): + thread_calls.append(func) + return func(*args, **kwargs) + + class FakeModelRouter: + def __init__(self, profiles): + self.profiles = profiles + + def edit_image(self, *, output_path, **kwargs): + Path(output_path).write_bytes(b"edited") + return output_path + + monkeypatch.setattr(asyncio, "to_thread", fake_to_thread) + monkeypatch.setattr(edit_route, "ModelRouter", FakeModelRouter) + + request = EditRequest( + image_base64=base64.b64encode(b"image").decode(), + instruction="make it blue", + config=EditConfig(api_key="key", base_url="https://example.test/v1"), + ) + + response = asyncio.run(edit_route.edit_image(request)) + + assert response.success is True + assert response.image_base64 == base64.b64encode(b"edited").decode() + assert [getattr(func, "__name__", "") for func in thread_calls] == ["edit_image"] diff --git a/tests/test_export_pptx_ratio.py b/tests/test_export_pptx_ratio.py index 94c8075..888b0b8 100644 --- a/tests/test_export_pptx_ratio.py +++ b/tests/test_export_pptx_ratio.py @@ -9,6 +9,7 @@ from api.models import ExportRequest, ExportSlide from api.routes.export import _export_pptx, export_presentation +import api.routes.export as export_route class ExportPptxRatioTest(unittest.TestCase): @@ -43,6 +44,76 @@ def test_export_response_cleans_temp_directory_after_send(self): asyncio.run(response.background()) self.assertFalse(output_path.parent.exists()) + def test_pdf_export_runs_synchronous_work_in_thread(self): + thread_calls = [] + + async def fake_to_thread(func, *args, **kwargs): + thread_calls.append(func) + return func(*args, **kwargs) + + class FakePDFExporter: + def export(self, image_paths, output_path): + Path(output_path).write_bytes(b"%PDF-1.4") + return output_path + + original_to_thread = asyncio.to_thread + original_exporter = export_route.PDFExporter + asyncio.to_thread = fake_to_thread + export_route.PDFExporter = FakePDFExporter + try: + with tempfile.TemporaryDirectory() as tmp: + image_path = Path(tmp) / "slide.png" + Image.new("RGB", (800, 450), "white").save(image_path) + image_base64 = base64.b64encode(image_path.read_bytes()).decode() + + request = ExportRequest( + slides=[ExportSlide(image_base64=image_base64)], + format="pdf", + ) + response = asyncio.run(export_presentation(request)) + + self.assertTrue(Path(response.path).exists()) + self.assertEqual([getattr(func, "__name__", "") for func in thread_calls], ["export"]) + asyncio.run(response.background()) + finally: + asyncio.to_thread = original_to_thread + export_route.PDFExporter = original_exporter + + def test_pptx_export_runs_synchronous_work_in_thread(self): + thread_calls = [] + + async def fake_to_thread(func, *args, **kwargs): + thread_calls.append(func) + return func(*args, **kwargs) + + def fake_export_pptx(image_paths, output_path, aspect_ratio="16:9"): + Path(output_path).write_bytes(b"pptx") + + original_to_thread = asyncio.to_thread + original_export_pptx = export_route._export_pptx + asyncio.to_thread = fake_to_thread + export_route._export_pptx = fake_export_pptx + try: + with tempfile.TemporaryDirectory() as tmp: + image_path = Path(tmp) / "slide.png" + Image.new("RGB", (800, 450), "white").save(image_path) + image_base64 = base64.b64encode(image_path.read_bytes()).decode() + + request = ExportRequest( + slides=[ExportSlide(image_base64=image_base64)], + format="pptx", + ) + response = asyncio.run(export_presentation(request)) + + self.assertTrue(Path(response.path).exists()) + self.assertEqual( + [getattr(func, "__name__", "") for func in thread_calls], ["fake_export_pptx"] + ) + asyncio.run(response.background()) + finally: + asyncio.to_thread = original_to_thread + export_route._export_pptx = original_export_pptx + if __name__ == "__main__": unittest.main() diff --git a/tests/test_generator.py b/tests/test_generator.py index e480f1b..40215b1 100644 --- a/tests/test_generator.py +++ b/tests/test_generator.py @@ -18,7 +18,7 @@ pytestmark = pytest.mark.skipif( os.getenv("RUN_MODEL_TESTS") != "1", - reason="真实模型集成测试默认跳过；设置 RUN_MODEL_TESTS=1 后运行" + reason="真实模型集成测试默认跳过；设置 RUN_MODEL_TESTS=1 后运行", ) @@ -27,25 +27,25 @@ def test_full_generation(num_pages=None): print("=" * 60) print("🧪 测试：完整 PPT 生成流程") print("=" * 60) - + material = load_sample_material() print(f"📄 加载资料: {len(material)} 字符") - + # 使用默认配置，仅覆盖测试需要的参数 ppt_config = PPTConfig(num_pages=num_pages or 3) - + # 创建生成器（配置从 config.yaml 读取） generator = PPTGenerator(output_dir="test_output") - + result = generator.generate(material, ppt_config) - + print("\n" + "=" * 60) if result.success: print(f"✅ 测试通过！输出目录: {result.project_dir}") else: print(f"❌ 测试失败: {result.error_message}") print("=" * 60) - + return result @@ -54,22 +54,20 @@ def test_prompt_only(num_pages=None): print("=" * 60) print("🧪 测试：仅生成 Prompt") print("=" * 60) - + material = load_sample_material() print(f"📄 加载资料: {len(material)} 字符") - + # 使用默认配置，仅覆盖测试需要的参数 ppt_config = PPTConfig(num_pages=num_pages or 3) - + # 创建生成器 generator = PPTGenerator() - + prompt_data = generator.generate_prompts_only( - material, - ppt_config, - output_path="test_output/test_prompts.json" + material, ppt_config, output_path="test_output/test_prompts.json" ) - + print("\n📄 页面 Prompt:") print("-" * 40) for slide in prompt_data.slide_prompts[:2]: @@ -77,11 +75,11 @@ def test_prompt_only(num_pages=None): print(f"摘要: {slide.content_summary}") prompt_preview = slide.prompt[:150] + "..." if len(slide.prompt) > 150 else slide.prompt print(f"Prompt: {prompt_preview}") - + print("\n" + "=" * 60) print("✅ Prompt 生成测试完成！") print("=" * 60) - + return prompt_data @@ -90,49 +88,46 @@ def test_from_prompts(): print("=" * 60) print("🧪 测试：从 Prompt 文件生成图片") print("=" * 60) - + prompt_path = Path("test_output/test_prompts.json") if not prompt_path.exists(): print("⚠️ 请先运行 --mode prompt 生成 Prompt 文件") return None - + prompt_data = PromptData.load(str(prompt_path)) print(f"📄 加载了 {len(prompt_data.slide_prompts)} 个页面 Prompt") - + # 创建生成器 generator = PPTGenerator(output_dir="test_output") - + result = generator.generate_from_prompts(prompt_data) - + print("\n" + "=" * 60) if result.success: print(f"✅ 测试通过！输出目录: {result.project_dir}") else: print(f"❌ 测试失败: {result.error_message}") print("=" * 60) - + return result if __name__ == "__main__": import argparse - + parser = argparse.ArgumentParser(description="PPT 生成器测试") parser.add_argument( "--mode", choices=["full", "prompt", "from-prompt"], default="full", - help="测试模式: full=完整测试, prompt=仅生成Prompt, from-prompt=从Prompt生成" + help="测试模式: full=完整测试, prompt=仅生成Prompt, from-prompt=从Prompt生成", ) parser.add_argument( - "-n", "--num-pages", - type=int, - default=None, - help="生成页数（默认使用配置文件中的设置）" + "-n", "--num-pages", type=int, default=None, help="生成页数（默认使用配置文件中的设置）" ) - + args = parser.parse_args() - + if args.mode == "full": test_full_generation(args.num_pages) elif args.mode == "prompt": diff --git a/tests/test_image_result_normalizer.py b/tests/test_image_result_normalizer.py index 5b9a39b..a8118cf 100644 --- a/tests/test_image_result_normalizer.py +++ b/tests/test_image_result_normalizer.py @@ -13,15 +13,7 @@ def fetcher(url: str) -> bytes: return png_bytes result = ImageResultNormalizer(fetcher=fetcher).normalize( - { - "choices": [ - { - "message": { - "content": "![image](https://example.test/slide.png)\n" - } - } - ] - } + {"choices": [{"message": {"content": "![image](https://example.test/slide.png)\n"}}]} ) self.assertEqual(result.base64_data, base64.b64encode(png_bytes).decode()) @@ -31,9 +23,7 @@ def fetcher(url: str) -> bytes: def test_accepts_b64_json_from_image_endpoint(self): encoded = base64.b64encode(b"image-bytes").decode() - result = ImageResultNormalizer().normalize( - {"data": [{"b64_json": encoded}]} - ) + result = ImageResultNormalizer().normalize({"data": [{"b64_json": encoded}]}) self.assertEqual(result.base64_data, encoded) self.assertEqual(result.source, "base64") @@ -41,9 +31,7 @@ def test_accepts_b64_json_from_image_endpoint(self): def test_accepts_data_url(self): encoded = base64.b64encode(b"image-bytes").decode() - result = ImageResultNormalizer().normalize( - f"data:image/jpeg;base64,{encoded}" - ) + result = ImageResultNormalizer().normalize(f"data:image/jpeg;base64,{encoded}") self.assertEqual(result.base64_data, encoded) self.assertEqual(result.mime_type, "image/jpeg") diff --git a/tests/test_model_profiles.py b/tests/test_model_profiles.py index 10fd640..a7cc6ad 100644 --- a/tests/test_model_profiles.py +++ b/tests/test_model_profiles.py @@ -1,6 +1,12 @@ import unittest -from src.model_profiles import ModelProfile, ModelProfileSet, load_default_profiles, load_profiles_from_env, resolve_model_profiles +from src.model_profiles import ( + ModelProfile, + ModelProfileSet, + load_default_profiles, + load_profiles_from_env, + resolve_model_profiles, +) class ModelProfileResolutionTest(unittest.TestCase): @@ -49,9 +55,15 @@ def test_explicit_edit_profile_overrides_image_profile(self): def test_public_view_masks_api_keys(self): profiles = ModelProfileSet( - prompt=ModelProfile(role="prompt", model="text", base_url="https://t/v1", api_key="secret"), - image=ModelProfile(role="image", model="image", base_url="https://i/v1", api_key="secret"), - edit=ModelProfile(role="edit", model="image", base_url="https://i/v1", api_key="secret"), + prompt=ModelProfile( + role="prompt", model="text", base_url="https://t/v1", api_key="secret" + ), + image=ModelProfile( + role="image", model="image", base_url="https://i/v1", api_key="secret" + ), + edit=ModelProfile( + role="edit", model="image", base_url="https://i/v1", api_key="secret" + ), ) public = profiles.to_public_dict() diff --git a/tests/test_model_router.py b/tests/test_model_router.py index 60c9052..da83e4e 100644 --- a/tests/test_model_router.py +++ b/tests/test_model_router.py @@ -32,9 +32,7 @@ def test_generate_text_passes_thinking_extra_body(self): ), ) - response = SimpleNamespace( - choices=[SimpleNamespace(message=SimpleNamespace(content="ok"))] - ) + response = SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content="ok"))]) with patch("src.model_router.OpenAI") as openai_cls: create = openai_cls.return_value.chat.completions.create diff --git a/tests/test_profile_resolver.py b/tests/test_profile_resolver.py index 75d6205..2d0dd82 100644 --- a/tests/test_profile_resolver.py +++ b/tests/test_profile_resolver.py @@ -33,12 +33,18 @@ def default_profiles(self): def test_empty_frontend_generation_config_uses_backend_profiles(self): config = GenerationConfig( - text=TextApiConfig(api_key="", base_url="https://oneapi.example/v1", model="DeepSeek-V4-Pro"), - image=ImageApiConfig(api_key="", base_url="https://image.example/v1", model="gpt-image-2"), + text=TextApiConfig( + api_key="", base_url="https://oneapi.example/v1", model="DeepSeek-V4-Pro" + ), + image=ImageApiConfig( + api_key="", base_url="https://image.example/v1", model="gpt-image-2" + ), page_count=1, ) - with patch("api.profile_resolver.load_default_profiles", return_value=self.default_profiles()): + with patch( + "api.profile_resolver.load_default_profiles", return_value=self.default_profiles() + ): profiles = profiles_from_generation_config(config) self.assertEqual(profiles.prompt.api_key, "text-key") @@ -46,12 +52,18 @@ def test_empty_frontend_generation_config_uses_backend_profiles(self): def test_masked_frontend_generation_config_uses_backend_profiles(self): config = GenerationConfig( - text=TextApiConfig(api_key="SET", base_url="https://oneapi.example/v1", model="DeepSeek-V4-Pro"), - image=ImageApiConfig(api_key="SET", base_url="https://image.example/v1", model="gpt-image-2"), + text=TextApiConfig( + api_key="SET", base_url="https://oneapi.example/v1", model="DeepSeek-V4-Pro" + ), + image=ImageApiConfig( + api_key="SET", base_url="https://image.example/v1", model="gpt-image-2" + ), page_count=1, ) - with patch("api.profile_resolver.load_default_profiles", return_value=self.default_profiles()): + with patch( + "api.profile_resolver.load_default_profiles", return_value=self.default_profiles() + ): profiles = profiles_from_generation_config(config) self.assertEqual(profiles.prompt.model, "default-text") @@ -59,12 +71,22 @@ def test_masked_frontend_generation_config_uses_backend_profiles(self): def test_complete_frontend_generation_config_overrides_backend_profiles(self): config = GenerationConfig( - text=TextApiConfig(api_key="override-text-key", base_url="https://text.override/v1", model="override-text"), - image=ImageApiConfig(api_key="override-image-key", base_url="https://image.override/v1", model="override-image"), + text=TextApiConfig( + api_key="override-text-key", + base_url="https://text.override/v1", + model="override-text", + ), + image=ImageApiConfig( + api_key="override-image-key", + base_url="https://image.override/v1", + model="override-image", + ), page_count=1, ) - with patch("api.profile_resolver.load_default_profiles", return_value=self.default_profiles()): + with patch( + "api.profile_resolver.load_default_profiles", return_value=self.default_profiles() + ): profiles = profiles_from_generation_config(config) self.assertEqual(profiles.prompt.api_key, "override-text-key") @@ -73,7 +95,9 @@ def test_complete_frontend_generation_config_overrides_backend_profiles(self): def test_empty_edit_config_uses_backend_profiles(self): config = EditConfig(api_key="", base_url="", model="gpt-image-2") - with patch("api.profile_resolver.load_default_profiles", return_value=self.default_profiles()): + with patch( + "api.profile_resolver.load_default_profiles", return_value=self.default_profiles() + ): profiles = profiles_from_edit_config(config) self.assertEqual(profiles.edit.api_key, "edit-key") diff --git a/tests/test_prompt_planning.py b/tests/test_prompt_planning.py index 595d7da..2a2c932 100644 --- a/tests/test_prompt_planning.py +++ b/tests/test_prompt_planning.py @@ -57,7 +57,9 @@ def _prompt_plan(page_count=3): def test_generate_outline_retries_and_validates_page_count(): bad = _outline(page_count=2) good = _outline(page_count=3) - client = FakeTextClient([json.dumps(bad, ensure_ascii=False), json.dumps(good, ensure_ascii=False)]) + client = FakeTextClient( + [json.dumps(bad, ensure_ascii=False), json.dumps(good, ensure_ascii=False)] + ) generator = PromptGenerator(client) config = PPTConfig(num_pages=3, user_requirements="强调架构风险", target_audience="研发团队") @@ -79,7 +81,9 @@ def test_generate_prompts_from_outline_keeps_display_content_and_exact_pages(): assert len(prompt_data.slide_prompts) == 3 assert prompt_data.slide_prompts[0].display_content.startswith("第 1 页展示") - assert prompt_data.slide_prompts[0].prompt.startswith("你生成的 PPT 其中一页的内容，要图文并茂。") + assert prompt_data.slide_prompts[0].prompt.startswith( + "你生成的 PPT 其中一页的内容，要图文并茂。" + ) def test_generate_prompts_from_outline_rejects_internal_prompt_without_prefix():