#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 数据库服务层 负责景区、产品、风格、受众等数据的查询和管理 """ import logging import time import traceback from typing import Dict, Any, Optional, List, Tuple import mysql.connector from mysql.connector import pooling from core.config import ConfigManager logger = logging.getLogger(__name__) class DatabaseService: """数据库服务类""" def __init__(self, config_manager: ConfigManager): """ 初始化数据库服务 Args: config_manager: 配置管理器 """ self.config_manager = config_manager self.db_pool = self._init_db_pool() def _init_db_pool(self): """初始化数据库连接池""" # 获取数据库配置 raw_db_config = self.config_manager.get_raw_config('database') # 处理环境变量 db_config = self._process_env_vars(raw_db_config) # 连接尝试配置 connection_attempts = [ {"desc": "使用配置文件中的设置", "config": db_config}, {"desc": "使用明确的密码", "config": {**db_config, "password": "password"}}, {"desc": "使用空密码", "config": {**db_config, "password": ""}}, {"desc": "使用auth_plugin", "config": {**db_config, "auth_plugin": "mysql_native_password"}} ] # 尝试不同的连接方式 for attempt in connection_attempts: try: # 打印连接信息（不包含密码） connection_info = {k: v for k, v in attempt["config"].items() if k != 'password'} logger.info(f"尝试连接数据库 ({attempt['desc']}): {connection_info}") # 创建连接池 # 从配置中分离MySQL连接池支持的参数和不支持的参数 config = attempt["config"].copy() # MySQL连接池不支持的参数，需要移除 unsupported_params = [ 'max_retry_attempts', 'query_timeout', 'soft_delete_field', 'active_record_value' ] for param in unsupported_params: config.pop(param, None) # 设置连接池参数，使用配置文件中的值或默认值 pool_size = config.pop('pool_size', 10) pool = pooling.MySQLConnectionPool( pool_name=f"database_service_pool_{int(time.time())}", pool_size=pool_size, **config ) # 测试连接 with pool.get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT 1") cursor.fetchall() logger.info(f"数据库连接池初始化成功 ({attempt['desc']})") return pool except Exception as e: error_details = traceback.format_exc() logger.error(f"数据库连接尝试 ({attempt['desc']}) 失败: {e}\n{error_details}") logger.error("所有数据库连接尝试都失败了") return None def _process_env_vars(self, config: Dict[str, Any]) -> Dict[str, Any]: """处理环境变量""" import os processed_config = {} for key, value in config.items(): if isinstance(value, str) and value.startswith("${") and value.endswith("}"): env_var = value[2:-1] processed_config[key] = os.environ.get(env_var, "") else: processed_config[key] = value return processed_config def get_scenic_spot_by_id(self, spot_id: int) -> Optional[Dict[str, Any]]: """根据ID获取单个景区信息""" if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM scenicSpot WHERE id = %s AND isDelete = 0", (spot_id,) ) result = cursor.fetchone() if result: logger.info(f"找到景区信息: ID={spot_id}, 名称={result['name']}") return result else: logger.warning(f"未找到景区信息: ID={spot_id}") return None except Exception as e: logger.error(f"查询景区信息失败: {e}") return None def get_product_by_id(self, product_id: int) -> Optional[Dict[str, Any]]: """根据ID获取单个产品信息""" if not self.db_pool: return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute("SELECT * FROM product WHERE id = %s AND isDelete = 0", (product_id,)) result = cursor.fetchone() if result: logger.info(f"找到产品信息: ID={product_id}, 名称={result['productName']}") return result else: logger.warning(f"未找到产品信息: ID={product_id}") return None except Exception as e: logger.error(f"查询产品信息失败: {e}") return None def get_style_by_id(self, style_id: int) -> Optional[Dict[str, Any]]: """ 根据ID获取风格信息 Args: style_id: 风格ID Returns: 风格信息字典，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM contentStyle WHERE id = %s AND isDelete = 0", (style_id,) ) result = cursor.fetchone() if result: logger.info(f"找到风格信息: ID={style_id}, 名称={result['styleName']}") return result else: logger.warning(f"未找到风格信息: ID={style_id}") return None except Exception as e: logger.error(f"查询风格信息失败: {e}") return None def get_audience_by_id(self, audience_id: int) -> Optional[Dict[str, Any]]: """ 根据ID获取受众信息 Args: audience_id: 受众ID Returns: 受众信息字典，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM targetAudience WHERE id = %s AND isDelete = 0", (audience_id,) ) result = cursor.fetchone() if result: logger.info(f"找到受众信息: ID={audience_id}, 名称={result['audienceName']}") return result else: logger.warning(f"未找到受众信息: ID={audience_id}") return None except Exception as e: logger.error(f"查询受众信息失败: {e}") return None def get_scenic_spots_by_ids(self, spot_ids: List[int]) -> List[Dict[str, Any]]: """ 根据ID列表批量获取景区信息 Args: spot_ids: 景区ID列表 Returns: 景区信息列表 """ if not self.db_pool or not spot_ids: logger.warning("数据库连接池未初始化或景区ID列表为空") return self._get_fallback_scenic_spots(spot_ids) try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: # 构建IN查询 placeholders = ','.join(['%s'] * len(spot_ids)) query = f"SELECT * FROM scenicSpot WHERE id IN ({placeholders}) AND isDelete = 0" cursor.execute(query, spot_ids) results = cursor.fetchall() # 检查是否所有ID都找到了对应的记录 found_ids = {result['id'] for result in results} missing_ids = set(spot_ids) - found_ids if missing_ids: logger.warning(f"部分景区ID未找到: {missing_ids}") # 添加兜底数据 fallback_spots = self._get_fallback_scenic_spots(list(missing_ids)) results.extend(fallback_spots) logger.info(f"批量查询景区信息: 请求{len(spot_ids)}个，找到{len([r for r in results if r['id'] in spot_ids])}个") return results except Exception as e: logger.error(f"批量查询景区信息失败: {e}") return self._get_fallback_scenic_spots(spot_ids) def _get_fallback_scenic_spots(self, spot_ids: List[int]) -> List[Dict[str, Any]]: """ 获取景区的兜底数据 """ fallback_spots = [] for spot_id in spot_ids: fallback_spots.append({ 'id': spot_id, 'name': f'景区{spot_id}', 'address': '默认地址', 'advantage': '优美的自然风光', 'highlight': '值得一游的景点', 'isPublic': 1, 'isDelete': 0, '_is_fallback': True }) logger.info(f"使用兜底景区数据: {len(fallback_spots)}个") return fallback_spots def get_products_by_ids(self, productIds: List[int]) -> List[Dict[str, Any]]: """ 根据ID列表批量获取产品信息 Args: productIds: 产品ID列表 Returns: 产品信息列表 """ if not self.db_pool or not productIds: logger.warning("数据库连接池未初始化或产品ID列表为空") return self._get_fallback_products(productIds) try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: # 构建IN查询 placeholders = ','.join(['%s'] * len(productIds)) query = f"SELECT * FROM product WHERE id IN ({placeholders}) AND isDelete = 0" cursor.execute(query, productIds) results = cursor.fetchall() # 检查是否所有ID都找到了对应的记录 found_ids = {result['id'] for result in results} missing_ids = set(productIds) - found_ids if missing_ids: logger.warning(f"部分产品ID未找到: {missing_ids}") # 添加兜底数据 fallback_products = self._get_fallback_products(list(missing_ids)) results.extend(fallback_products) logger.info(f"批量查询产品信息: 请求{len(productIds)}个，找到{len([r for r in results if r['id'] in productIds])}个") return results except Exception as e: logger.error(f"批量查询产品信息失败: {e}") return self._get_fallback_products(productIds) def _get_fallback_products(self, productIds: List[int]) -> List[Dict[str, Any]]: """ 获取产品的兜底数据 """ fallback_products = [] for product_id in productIds: fallback_products.append({ 'id': product_id, 'productName': f'产品{product_id}', 'originPrice': 999, 'realPrice': 888, 'packageInfo': '包含景区门票和导游服务', 'advantage': '性价比高，服务优质', 'isDelete': 0, '_is_fallback': True }) logger.info(f"使用兜底产品数据: {len(fallback_products)}个") return fallback_products def get_styles_by_ids(self, styleIds: List[int]) -> List[Dict[str, Any]]: """ 根据ID列表批量获取风格信息 Args: styleIds: 风格ID列表 Returns: 风格信息列表 """ if not self.db_pool or not styleIds: logger.warning("数据库连接池未初始化或风格ID列表为空") return self._get_fallback_styles(styleIds) try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: # 构建IN查询 placeholders = ','.join(['%s'] * len(styleIds)) query = f"SELECT * FROM contentStyle WHERE id IN ({placeholders}) AND isDelete = 0" cursor.execute(query, styleIds) results = cursor.fetchall() # 检查是否所有ID都找到了对应的记录 found_ids = {result['id'] for result in results} missing_ids = set(styleIds) - found_ids if missing_ids: logger.warning(f"部分风格ID未找到: {missing_ids}") # 添加兜底数据 fallback_styles = self._get_fallback_styles(list(missing_ids)) results.extend(fallback_styles) logger.info(f"批量查询风格信息: 请求{len(styleIds)}个，找到{len([r for r in results if r['id'] in styleIds])}个") return results except Exception as e: logger.error(f"批量查询风格信息失败: {e}") return self._get_fallback_styles(styleIds) def _get_fallback_styles(self, styleIds: List[int]) -> List[Dict[str, Any]]: """ 获取风格的兜底数据 """ fallback_styles = [] for style_id in styleIds: fallback_styles.append({ 'id': style_id, 'styleName': f'风格{style_id}', 'description': '默认风格描述', 'isDelete': 0, '_is_fallback': True }) logger.info(f"使用兜底风格数据: {len(fallback_styles)}个") return fallback_styles def get_audiences_by_ids(self, audienceIds: List[int]) -> List[Dict[str, Any]]: """ 根据ID列表批量获取受众信息 Args: audienceIds: 受众ID列表 Returns: 受众信息列表 """ if not self.db_pool or not audienceIds: logger.warning("数据库连接池未初始化或受众ID列表为空") return self._get_fallback_audiences(audienceIds) try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: # 构建IN查询 placeholders = ','.join(['%s'] * len(audienceIds)) query = f"SELECT * FROM targetAudience WHERE id IN ({placeholders}) AND isDelete = 0" cursor.execute(query, audienceIds) results = cursor.fetchall() # 检查是否所有ID都找到了对应的记录 found_ids = {result['id'] for result in results} missing_ids = set(audienceIds) - found_ids if missing_ids: logger.warning(f"部分受众ID未找到: {missing_ids}") # 添加兜底数据 fallback_audiences = self._get_fallback_audiences(list(missing_ids)) results.extend(fallback_audiences) logger.info(f"批量查询受众信息: 请求{len(audienceIds)}个，找到{len([r for r in results if r['id'] in audienceIds])}个") return results except Exception as e: logger.error(f"批量查询受众信息失败: {e}") return self._get_fallback_audiences(audienceIds) def _get_fallback_audiences(self, audienceIds: List[int]) -> List[Dict[str, Any]]: """ 获取受众的兜底数据 """ fallback_audiences = [] for audience_id in audienceIds: fallback_audiences.append({ 'id': audience_id, 'audienceName': f'受众{audience_id}', 'description': '默认受众描述', 'isDelete': 0, '_is_fallback': True }) logger.info(f"使用兜底受众数据: {len(fallback_audiences)}个") return fallback_audiences def list_all_scenic_spots(self, user_id: Optional[int] = None, is_public: Optional[bool] = None) -> List[Dict[str, Any]]: """ 获取所有景区列表 Args: user_id: 用户ID，如果提供则只返回该用户的景区 is_public: 是否公开，如果提供则过滤公开/私有景区 Returns: 景区列表 """ if not self.db_pool: return [] try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: # 构建查询条件 conditions = ["isDelete = 0"] params = [] if user_id is not None: conditions.append("userId = %s") params.append(user_id) if is_public is not None: conditions.append("isPublic = %s") params.append(1 if is_public else 0) query = f""" SELECT id, name, address, trafficInfo, description, advantage, highlight, isPublic, userId FROM scenicSpot WHERE {' AND '.join(conditions)} ORDER BY name """ cursor.execute(query, params) results = cursor.fetchall() logger.info(f"获取景区列表: 找到{len(results)}个景区") return results except Exception as e: logger.error(f"获取景区列表失败: {e}") return [] def list_all_products(self, user_id: Optional[int] = None, is_public: Optional[bool] = None) -> List[Dict[str, Any]]: """ 获取所有产品列表 Args: user_id: 用户ID，如果提供则只返回该用户的产品 is_public: 是否公开，如果提供则过滤公开/私有产品 Returns: 产品列表 """ if not self.db_pool: return [] try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: # 构建查询条件 conditions = ["isDelete = 0"] params = [] if user_id is not None: conditions.append("userId = %s") params.append(user_id) if is_public is not None: conditions.append("isPublic = %s") params.append(1 if is_public else 0) query = f""" SELECT id, productName, originPrice, realPrice, packageInfo, detailedDescription, keyAdvantages, highlights, usageRules, surcharge, reservation, refund, discounts, isPublic, userId FROM product WHERE {' AND '.join(conditions)} ORDER BY productName """ cursor.execute(query, params) results = cursor.fetchall() logger.info(f"获取产品列表: 找到{len(results)}个产品") return results except Exception as e: logger.error(f"获取产品列表失败: {e}") return [] def list_all_styles(self) -> List[Dict[str, Any]]: """ 获取所有风格列表 Returns: 风格列表 """ if not self.db_pool: return [] try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute("SELECT id, styleName, description FROM contentStyle WHERE isDelete = 0 ORDER BY styleName") results = cursor.fetchall() logger.info(f"获取风格列表: 找到{len(results)}个风格") return results except Exception as e: logger.error(f"获取风格列表失败: {e}") return [] def list_all_audiences(self) -> List[Dict[str, Any]]: """ 获取所有受众列表 Returns: 受众列表 """ if not self.db_pool: return [] try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute("SELECT id, audienceName, description FROM targetAudience WHERE isDelete = 0 ORDER BY audienceName") results = cursor.fetchall() logger.info(f"获取受众列表: 找到{len(results)}个受众") return results except Exception as e: logger.error(f"获取受众列表失败: {e}") return [] def is_available(self) -> bool: """ 检查数据库服务是否可用 Returns: 数据库是否可用 """ return self.db_pool is not None # 名称到ID的反向查询方法 def get_style_id_by_name(self, style_name: str) -> Optional[int]: """ 根据风格名称获取风格ID Args: style_name: 风格名称 Returns: 风格ID，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT id FROM contentStyle WHERE styleName = %s AND isDelete = 0", (style_name,) ) result = cursor.fetchone() if result: return result['id'] else: logger.warning(f"未找到风格: {style_name}") return None except Exception as e: logger.error(f"查询风格ID失败: {e}") return None def get_audience_id_by_name(self, audience_name: str) -> Optional[int]: """ 根据受众名称获取受众ID Args: audience_name: 受众名称 Returns: 受众ID，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT id FROM targetAudience WHERE audienceName = %s AND isDelete = 0", (audience_name,) ) result = cursor.fetchone() if result: return result['id'] else: logger.warning(f"未找到受众: {audience_name}") return None except Exception as e: logger.error(f"查询受众ID失败: {e}") return None def get_scenic_spot_id_by_name(self, spot_name: str) -> Optional[int]: """ 根据景区名称获取景区ID Args: spot_name: 景区名称 Returns: 景区ID，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT id FROM scenicSpot WHERE name = %s AND isDelete = 0", (spot_name,) ) result = cursor.fetchone() if result: return result['id'] else: logger.warning(f"未找到景区: {spot_name}") return None except Exception as e: logger.error(f"查询景区ID失败: {e}") return None def get_product_id_by_name(self, product_name: str) -> Optional[int]: """ 根据产品名称获取产品ID Args: product_name: 产品名称 Returns: 产品ID，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT id FROM product WHERE productName = %s AND isDelete = 0", (product_name,) ) result = cursor.fetchone() if result: return result['id'] else: logger.warning(f"未找到产品: {product_name}") return None except Exception as e: logger.error(f"查询产品ID失败: {e}") return None def get_image_by_id(self, image_id: int) -> Optional[Dict[str, Any]]: """ 根据ID获取图像信息 Args: image_id: 图像ID Returns: 图像信息字典，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM material WHERE id = %s AND materialType = 'image' AND isDelete = 0", (image_id,) ) result = cursor.fetchone() if result: logger.info(f"找到图像信息: ID={image_id}, 名称={result['materialName']}") return result else: logger.warning(f"未找到图像信息: ID={image_id}") return None except Exception as e: logger.error(f"查询图像信息失败: {e}") return None def get_images_by_ids(self, image_ids: List[int]) -> List[Dict[str, Any]]: """ 根据ID列表批量获取图像信息 Args: image_ids: 图像ID列表 Returns: 图像信息列表 """ if not self.db_pool or not image_ids: return [] try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: # 构建IN查询 placeholders = ','.join(['%s'] * len(image_ids)) query = f"SELECT * FROM material WHERE id IN ({placeholders}) AND materialType = 'image' AND isDelete = 0" cursor.execute(query, image_ids) results = cursor.fetchall() logger.info(f"批量查询图像信息: 请求{len(image_ids)}个，找到{len(results)}个") return results except Exception as e: logger.error(f"批量查询图像信息失败: {e}") return [] def get_content_by_id(self, content_id: int) -> Optional[Dict[str, Any]]: """ 根据ID获取内容信息 Args: content_id: 内容ID Returns: 内容信息字典，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM content WHERE id = %s AND isDelete = 0", (content_id,) ) result = cursor.fetchone() if result: logger.info(f"找到内容信息: ID={content_id}, 标题={result['title']}") return result else: logger.warning(f"未找到内容信息: ID={content_id}") return None except Exception as e: logger.error(f"查询内容信息失败: {e}") return None def get_content_by_topic_index(self, topic_index: str) -> Optional[Dict[str, Any]]: """根据主题索引获取内容信息""" if not self.db_pool: return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM content WHERE topicIndex = %s AND isDelete = 0 ORDER BY createTime DESC LIMIT 1", (topic_index,) ) result = cursor.fetchone() if result: logger.info(f"找到内容信息: topicIndex={topic_index}, 标题={result['title']}") return result else: logger.warning(f"未找到内容信息: topicIndex={topic_index}") return None except Exception as e: logger.error(f"查询内容信息失败: {e}") return None def get_images_by_folder_id(self, folder_id: int) -> List[Dict[str, Any]]: """ 根据文件夹ID获取图像列表 Args: folder_id: 文件夹ID Returns: 图像信息列表 """ if not self.db_pool: return [] try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM material WHERE folderId = %s AND materialType = 'image' AND isDelete = 0 ORDER BY createTime DESC", (folder_id,) ) results = cursor.fetchall() logger.info(f"根据文件夹ID获取图像: folderId={folder_id}, 找到{len(results)}个图像") return results except Exception as e: logger.error(f"根据文件夹ID获取图像失败: {e}") return [] def get_folder_by_id(self, folder_id: int) -> Optional[Dict[str, Any]]: """ 根据ID获取文件夹信息 Args: folder_id: 文件夹ID Returns: 文件夹信息字典，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM material_folder WHERE id = %s AND isDelete = 0", (folder_id,) ) result = cursor.fetchone() if result: logger.info(f"找到文件夹信息: ID={folder_id}, 名称={result['folderName']}") return result else: logger.warning(f"未找到文件夹信息: ID={folder_id}") return None except Exception as e: logger.error(f"查询文件夹信息失败: {e}") return None def get_related_images_for_content(self, content_id: int, limit: int = 10) -> List[Dict[str, Any]]: """ 获取与内容相关的图像列表 Args: content_id: 内容ID limit: 限制数量 Returns: 相关图像列表 """ if not self.db_pool: return [] try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: # 获取内容信息 cursor.execute( "SELECT * FROM content WHERE id = %s AND isDelete = 0", (content_id,) ) content = cursor.fetchone() if not content: return [] # 获取相关的图像（这里可以根据业务逻辑调整查询条件） cursor.execute( "SELECT * FROM material WHERE materialType = 'image' AND isDelete = 0 ORDER BY RAND() LIMIT %s", (limit,) ) results = cursor.fetchall() logger.info(f"获取相关内容图像: contentId={content_id}, 找到{len(results)}个图像") return results except Exception as e: logger.error(f"获取相关内容图像失败: {e}") return [] # 模板相关查询方法 def get_all_poster_templates(self) -> List[Dict[str, Any]]: """ 获取所有海报模板 Returns: 模板列表 """ if not self.db_pool: return [] try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM poster_templates ORDER BY created_at" ) results = cursor.fetchall() logger.info(f"获取海报模板列表: 找到{len(results)}个模板") return results except Exception as e: logger.error(f"获取海报模板列表失败: {e}") return [] def get_poster_template_by_id(self, template_id: str) -> Optional[Dict[str, Any]]: """ 根据ID获取海报模板信息 Args: template_id: 模板ID Returns: 模板信息字典，如果未找到则返回None """ if not self.db_pool: logger.error("数据库连接池未初始化") return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM poster_templates WHERE id = %s", (template_id,) ) result = cursor.fetchone() if result: logger.info(f"找到模板信息: ID={template_id}, 名称={result['name']}") return result else: logger.warning(f"未找到模板信息: ID={template_id}") return None except Exception as e: logger.error(f"查询模板信息失败: {e}") return None def get_active_poster_templates(self) -> List[Dict[str, Any]]: """ 获取所有激活的海报模板 Returns: 激活的模板列表 """ if not self.db_pool: return [] try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM poster_templates WHERE is_active = 1 ORDER BY created_at" ) results = cursor.fetchall() logger.info(f"获取激活模板列表: 找到{len(results)}个模板") return results except Exception as e: logger.error(f"获取激活模板列表失败: {e}") return [] def update_template_usage_stats(self, template_id: str, success: bool = True, processing_time: float = 0.0): """ 更新模板使用统计 Args: template_id: 模板ID success: 是否成功 processing_time: 处理时间（秒） """ if not self.db_pool: logger.error("数据库连接池未初始化") return try: conn = self.db_pool.get_connection() cursor = conn.cursor() # 更新或插入统计记录 cursor.execute(""" INSERT INTO templateUsageStats (templateId, usageCount, successCount, errorCount, avgProcessingTime, lastUsedAt) VALUES (%s, 1, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE usageCount = usageCount + 1, successCount = successCount + %s, errorCount = errorCount + %s, avgProcessingTime = (avgProcessingTime * (usageCount - 1) + %s) / usageCount, lastUsedAt = NOW(), updatedAt = NOW() """, ( template_id, 1 if success else 0, 0 if success else 1, processing_time, 1 if success else 0, 0 if success else 1, processing_time )) conn.commit() cursor.close() conn.close() logger.info(f"更新模板使用统计: template_id={template_id}, success={success}, time={processing_time:.3f}s") except Exception as e: logger.error(f"更新模板使用统计失败: {e}") def get_template_usage_stats(self, template_id: str) -> Optional[Dict[str, Any]]: """ 获取模板使用统计 Args: template_id: 模板ID Returns: 使用统计信息 """ if not self.db_pool: return None try: with self.db_pool.get_connection() as conn: with conn.cursor(dictionary=True) as cursor: cursor.execute( "SELECT * FROM templateUsageStats WHERE templateId = %s", (template_id,) ) result = cursor.fetchone() if result: logger.info(f"找到模板使用统计: template_id={template_id}") return result else: logger.warning(f"未找到模板使用统计: template_id={template_id}") return None except Exception as e: logger.error(f"查询模板使用统计失败: {e}") return None