TravelContentCreator/domain/poster/poster_service.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
海报服务 V2 - 轻量版

特点:
1. 不依赖数据库
2. 不依赖旧的 core/utils 模块
3. 直接调用模板渲染
4. 支持 URL/路径/Base64 三种图片输入
"""

import logging
import base64
import importlib
from io import BytesIO
from typing import Dict, Any, Optional, List, Tuple
from pathlib import Path

logger = logging.getLogger(__name__)

try:
    from PIL import Image
    PIL_AVAILABLE = True
except ImportError:
    PIL_AVAILABLE = False
    logger.warning("PIL 未安装，海报功能不可用")


class PosterServiceV2:
    """
    轻量海报服务
    
    职责:
    - 管理海报模板
    - 处理图片输入
    - 调用模板渲染
    - 输出结果
    """
    
    # 默认模板配置
    DEFAULT_TEMPLATES = {
        'vibrant': {
            'id': 'vibrant',
            'name': '活力风格',
            'handler_path': 'poster.templates.vibrant_template',
            'class_name': 'VibrantTemplate',
            'description': '适合景点、活动等充满活力的场景',
        },
        'business': {
            'id': 'business',
            'name': '商务风格',
            'handler_path': 'poster.templates.business_template',
            'class_name': 'BusinessTemplate',
            'description': '适合酒店、房地产等商务场景',
        },
        'collage': {
            'id': 'collage',
            'name': '拼贴风格',
            'handler_path': 'poster.templates.collage_template',
            'class_name': 'CollageTemplate',
            'description': '多图拼贴风格',
        },
    }
    
    def __init__(self):
        self._templates = self.DEFAULT_TEMPLATES.copy()
        self._template_instances = {}
        self.logger = logging.getLogger(f"{__name__}.PosterServiceV2")
    
    def get_available_templates(self) -> List[Dict[str, Any]]:
        """获取所有可用模板"""
        return [
            {
                'id': t['id'],
                'name': t['name'],
                'description': t['description'],
            }
            for t in self._templates.values()
        ]
    
    def generate_poster(
        self,
        template_id: str,
        content: Dict[str, Any],
        images: List['Image.Image'] = None,
        images_base64: List[str] = None,
        generate_fabric_json: bool = False,
        **kwargs
    ) -> Dict[str, Any]:
        """
        生成海报
        
        Args:
            template_id: 模板 ID
            content: 海报内容 {title, content, tag, ...}
            images: PIL Image 列表
            images_base64: Base64 图片列表
            generate_fabric_json: 是否生成 Fabric.js JSON
            
        Returns:
            {
                'success': True/False,
                'image': PIL.Image (如果成功),
                'image_base64': str (如果成功),
                'fabric_json': dict (如果 generate_fabric_json=True),
                'error': str (如果失败),
            }
        """
        try:
            self.logger.info(f"开始生成海报，模板: {template_id}")
            
            # 1. 加载模板
            template = self._load_template(template_id)
            if not template:
                return {'success': False, 'error': f'模板 {template_id} 不存在或加载失败'}
            
            # 2. 处理图片输入
            if images is None:
                images = []
            
            if images_base64:
                for b64 in images_base64:
                    img = self._base64_to_image(b64)
                    if img:
                        images.append(img)
            
            if not images:
                self.logger.warning("没有输入图片，使用空白背景")
                # 创建默认背景
                images = [Image.new('RGBA', (1350, 1800), (240, 240, 240, 255))]
            
            # 3. 调用模板渲染
            # 注意: VibrantTemplate._generate_legacy 有 bug，直接把 images 当单图用
            # 这里传第一张图片，后续方案 C 会修复模板
            main_image = images[0] if images else None
            
            result = template.generate(
                images=main_image,  # 传递单张图片 (兼容模板 bug)
                content=content,
                generate_fabric_json=generate_fabric_json,
                **kwargs
            )
            
            # 4. 处理结果
            if isinstance(result, Image.Image):
                # 直接返回图片
                return {
                    'success': True,
                    'image': result,
                    'image_base64': self._image_to_base64(result),
                }
            elif isinstance(result, dict):
                # 返回包含多种格式的结果
                output = {'success': True}
                
                if 'image' in result:
                    output['image'] = result['image']
                    output['image_base64'] = self._image_to_base64(result['image'])
                
                if 'fabric_json' in result:
                    output['fabric_json'] = result['fabric_json']
                
                if 'error' in result:
                    output['success'] = False
                    output['error'] = result['error']
                
                return output
            else:
                return {'success': False, 'error': '模板返回格式异常'}
            
        except Exception as e:
            self.logger.error(f"生成海报失败: {e}", exc_info=True)
            return {'success': False, 'error': str(e)}
    
    def _load_template(self, template_id: str):
        """加载模板实例"""
        # 检查缓存
        if template_id in self._template_instances:
            return self._template_instances[template_id]
        
        # 检查模板是否存在
        if template_id not in self._templates:
            self.logger.error(f"模板不存在: {template_id}")
            return None
        
        template_info = self._templates[template_id]
        handler_path = template_info.get('handler_path')
        class_name = template_info.get('class_name')
        
        try:
            # 动态导入
            module = importlib.import_module(handler_path)
            template_class = getattr(module, class_name)
            
            # 实例化
            instance = template_class()
            
            # 缓存
            self._template_instances[template_id] = instance
            self.logger.info(f"加载模板成功: {template_id}")
            
            return instance
            
        except Exception as e:
            self.logger.error(f"加载模板失败: {template_id}, {e}")
            return None
    
    def _base64_to_image(self, base64_str: str) -> Optional['Image.Image']:
        """Base64 转 PIL Image"""
        try:
            # 移除 data URL 前缀
            if ',' in base64_str:
                base64_str = base64_str.split(',', 1)[1]
            
            image_data = base64.b64decode(base64_str)
            image = Image.open(BytesIO(image_data))
            
            if image.mode != 'RGBA':
                image = image.convert('RGBA')
            
            return image
            
        except Exception as e:
            self.logger.error(f"Base64 转图片失败: {e}")
            return None
    
    def _image_to_base64(self, image: 'Image.Image', format: str = 'PNG') -> str:
        """PIL Image 转 Base64"""
        buffer = BytesIO()
        image.save(buffer, format=format)
        return base64.b64encode(buffer.getvalue()).decode('utf-8')


# 全局单例
_service_instance = None

def get_poster_service() -> PosterServiceV2:
    """获取全局 PosterService 实例"""
    global _service_instance
    if _service_instance is None:
        _service_instance = PosterServiceV2()
    return _service_instance