#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 海报API路由 - 统一生成接口 """ import logging from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks from typing import List, Dict, Any from core.config import ConfigManager from core.ai import AIAgent from utils.file_io import OutputManager from api.services.poster import PosterService from api.models.poster import ( PosterGenerateRequest, PosterGenerateResponse, ImageUsageRequest, ImageUsageResponse, TemplateListResponse ) # 从依赖注入模块导入依赖 from api.dependencies import get_config, get_ai_agent, get_output_manager logger = logging.getLogger(__name__) # 创建路由 router = APIRouter() # 依赖注入函数 def get_poster_service( config_manager: ConfigManager = Depends(get_config), ai_agent: AIAgent = Depends(get_ai_agent), output_manager: OutputManager = Depends(get_output_manager) ) -> PosterService: """获取海报服务""" return PosterService(ai_agent, config_manager, output_manager) @router.get("/templates", response_model=TemplateListResponse, summary="获取可用模板列表") async def get_templates( poster_service: PosterService = Depends(get_poster_service) ): """ 获取可用的海报模板列表 """ try: templates = await poster_service.get_available_templates() return TemplateListResponse( templates=templates, total_count=len(templates) ) except Exception as e: logger.error(f"获取模板列表失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取模板列表失败: {str(e)}") @router.get("/templates/{template_id}", summary="获取指定模板信息") async def get_template_info( template_id: str, poster_service: PosterService = Depends(get_poster_service) ): """ 获取指定模板的详细信息 """ try: template_info = await poster_service.get_template_info(template_id) if not template_info: raise HTTPException(status_code=404, detail=f"模板 {template_id} 不存在") return template_info except HTTPException: raise except Exception as e: logger.error(f"获取模板信息失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取模板信息失败: {str(e)}") @router.post("/generate", response_model=PosterGenerateResponse, summary="生成海报") async def generate_poster( request: PosterGenerateRequest, poster_service: PosterService = Depends(get_poster_service) ): """ 生成海报 - **content_id**: 内容ID(可选) - **product_id**: 产品ID(可选) - **scenic_spot_id**: 景区ID(可选) - **image_ids**: 图像ID列表(可选) - **template_id**: 模板ID(默认为vibrant) - **generate_collage**: 是否生成拼图 - **poster_content**: 用户提供的海报内容(可选) - **force_llm_generation**: 是否强制使用LLM生成内容(可选) """ try: result = await poster_service.generate_poster( template_id=request.templateId, poster_content=request.posterContent, content_id=request.contentId, product_id=request.productId, scenic_spot_id=request.scenicSpotId, images_base64=request.imagesBase64, num_variations=request.numVariations, force_llm_generation=request.forceLlmGeneration ) return PosterGenerateResponse(**result) except ValueError as e: logger.error(f"参数错误: {e}") raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"生成海报失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"生成海报失败: {str(e)}") @router.post("/image-usage", response_model=ImageUsageResponse, summary="查询图像使用情况") async def get_image_usage( request: ImageUsageRequest, poster_service: PosterService = Depends(get_poster_service) ): """ 查询图像使用情况 """ try: result = poster_service.get_image_usage_info(request.image_ids) return ImageUsageResponse(**result) except Exception as e: logger.error(f"查询图像使用情况失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"查询图像使用情况失败: {str(e)}")