|
|
@@ -0,0 +1,259 @@
|
|
|
+"""
|
|
|
+图片管理服务层
|
|
|
+处理图片分类和图片信息的业务逻辑
|
|
|
+"""
|
|
|
+import logging
|
|
|
+import uuid
|
|
|
+import os
|
|
|
+from typing import Optional, List, Dict, Any, Tuple
|
|
|
+from datetime import datetime
|
|
|
+from app.base.async_mysql_connection import get_db_connection
|
|
|
+from app.base.minio_connection import get_minio_manager
|
|
|
+
|
|
|
+logger = logging.getLogger(__name__)
|
|
|
+
|
|
|
+class ImageService:
|
|
|
+ """图片管理服务类"""
|
|
|
+
|
|
|
+ def __init__(self):
|
|
|
+ """初始化服务"""
|
|
|
+ self.minio_manager = get_minio_manager()
|
|
|
+
|
|
|
+ async def get_categories(self) -> List[Dict[str, Any]]:
|
|
|
+ """获取所有图片分类并构建树形结构"""
|
|
|
+ conn = get_db_connection()
|
|
|
+ if not conn:
|
|
|
+ return []
|
|
|
+
|
|
|
+ cursor = conn.cursor()
|
|
|
+ try:
|
|
|
+ sql = "SELECT * FROM t_image_category ORDER BY created_time DESC"
|
|
|
+ cursor.execute(sql)
|
|
|
+ categories = cursor.fetchall()
|
|
|
+
|
|
|
+ # 构建树形结构
|
|
|
+ category_dict = {str(cat['id']): {**cat, 'children': []} for cat in categories}
|
|
|
+ root_nodes = []
|
|
|
+
|
|
|
+ for cat_id, cat_node in category_dict.items():
|
|
|
+ parent_id = str(cat_node['parent_id'])
|
|
|
+ if parent_id == '0':
|
|
|
+ root_nodes.append(cat_node)
|
|
|
+ elif parent_id in category_dict:
|
|
|
+ category_dict[parent_id]['children'].append(cat_node)
|
|
|
+ else:
|
|
|
+ # 如果父节点找不到，也作为根节点（防御性编程）
|
|
|
+ root_nodes.append(cat_node)
|
|
|
+
|
|
|
+ return root_nodes
|
|
|
+ except Exception as e:
|
|
|
+ logger.exception(f"获取图片分类失败: {e}")
|
|
|
+ return []
|
|
|
+ finally:
|
|
|
+ cursor.close()
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ async def add_category(self, data: Dict[str, Any], user_id: str) -> Tuple[bool, str, Optional[str]]:
|
|
|
+ """新增图片分类"""
|
|
|
+ conn = get_db_connection()
|
|
|
+ if not conn:
|
|
|
+ return False, "数据库连接失败", None
|
|
|
+
|
|
|
+ cursor = conn.cursor()
|
|
|
+ try:
|
|
|
+ category_id = str(uuid.uuid4())
|
|
|
+ parent_id = data.get('parent_id', '0')
|
|
|
+ type_name = data.get('type_name')
|
|
|
+ remark = data.get('remark')
|
|
|
+
|
|
|
+ if not type_name:
|
|
|
+ return False, "分类名称不能为空", None
|
|
|
+
|
|
|
+ sql = """
|
|
|
+ INSERT INTO t_image_category (
|
|
|
+ id, type_name, parent_id, remark,
|
|
|
+ created_by, updated_by, created_time, updated_time
|
|
|
+ ) VALUES (%s, %s, %s, %s, %s, %s, NOW(), NOW())
|
|
|
+ """
|
|
|
+ cursor.execute(sql, (category_id, type_name, parent_id, remark, user_id, user_id))
|
|
|
+ conn.commit()
|
|
|
+ return True, "新增成功", category_id
|
|
|
+ except Exception as e:
|
|
|
+ logger.exception(f"新增图片分类失败: {e}")
|
|
|
+ conn.rollback()
|
|
|
+ return False, f"新增失败: {str(e)}", None
|
|
|
+ finally:
|
|
|
+ cursor.close()
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ async def update_category(self, category_id: str, data: Dict[str, Any], user_id: str) -> Tuple[bool, str]:
|
|
|
+ """更新图片分类"""
|
|
|
+ conn = get_db_connection()
|
|
|
+ if not conn:
|
|
|
+ return False, "数据库连接失败"
|
|
|
+
|
|
|
+ cursor = conn.cursor()
|
|
|
+ try:
|
|
|
+ type_name = data.get('type_name')
|
|
|
+ remark = data.get('remark')
|
|
|
+
|
|
|
+ if not type_name:
|
|
|
+ return False, "分类名称不能为空"
|
|
|
+
|
|
|
+ sql = """
|
|
|
+ UPDATE t_image_category
|
|
|
+ SET type_name = %s, remark = %s, updated_by = %s, updated_time = NOW()
|
|
|
+ WHERE id = %s
|
|
|
+ """
|
|
|
+ cursor.execute(sql, (type_name, remark, user_id, category_id))
|
|
|
+ conn.commit()
|
|
|
+ return True, "更新成功"
|
|
|
+ except Exception as e:
|
|
|
+ logger.exception(f"更新图片分类失败: {e}")
|
|
|
+ conn.rollback()
|
|
|
+ return False, f"更新失败: {str(e)}"
|
|
|
+ finally:
|
|
|
+ cursor.close()
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ async def delete_category(self, category_id: str) -> Tuple[bool, str]:
|
|
|
+ """删除图片分类"""
|
|
|
+ conn = get_db_connection()
|
|
|
+ if not conn:
|
|
|
+ return False, "数据库连接失败"
|
|
|
+
|
|
|
+ cursor = conn.cursor()
|
|
|
+ try:
|
|
|
+ # 检查是否有子分类
|
|
|
+ cursor.execute("SELECT COUNT(*) as count FROM t_image_category WHERE parent_id = %s", (category_id,))
|
|
|
+ if cursor.fetchone()['count'] > 0:
|
|
|
+ return False, "该分类下有子分类，不能直接删除"
|
|
|
+
|
|
|
+ # 检查是否有图片
|
|
|
+ cursor.execute("SELECT COUNT(*) as count FROM t_image_info WHERE image_type = %s", (category_id,))
|
|
|
+ if cursor.fetchone()['count'] > 0:
|
|
|
+ return False, "该分类下有图片，不能直接删除"
|
|
|
+
|
|
|
+ cursor.execute("DELETE FROM t_image_category WHERE id = %s", (category_id,))
|
|
|
+ conn.commit()
|
|
|
+ return True, "删除成功"
|
|
|
+ except Exception as e:
|
|
|
+ logger.exception(f"删除图片分类失败: {e}")
|
|
|
+ conn.rollback()
|
|
|
+ return False, f"删除失败: {str(e)}"
|
|
|
+ finally:
|
|
|
+ cursor.close()
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ async def get_images(self, category_id: Optional[str] = None, page: int = 1, page_size: int = 10) -> Dict[str, Any]:
|
|
|
+ """获取图片列表"""
|
|
|
+ conn = get_db_connection()
|
|
|
+ if not conn:
|
|
|
+ return {"total": 0, "list": []}
|
|
|
+
|
|
|
+ cursor = conn.cursor()
|
|
|
+ try:
|
|
|
+ where_clauses = []
|
|
|
+ params = []
|
|
|
+
|
|
|
+ if category_id and category_id != '0':
|
|
|
+ where_clauses.append("image_type = %s")
|
|
|
+ params.append(category_id)
|
|
|
+
|
|
|
+ where_sql = ""
|
|
|
+ if where_clauses:
|
|
|
+ where_sql = "WHERE " + " AND ".join(where_clauses)
|
|
|
+
|
|
|
+ # 获取总数
|
|
|
+ count_sql = f"SELECT COUNT(*) as total FROM t_image_info {where_sql}"
|
|
|
+ cursor.execute(count_sql, tuple(params))
|
|
|
+ total = cursor.fetchone()['total']
|
|
|
+
|
|
|
+ # 获取列表
|
|
|
+ offset = (page - 1) * page_size
|
|
|
+ list_sql = f"""
|
|
|
+ SELECT i.*, c.type_name as category_name, u.username as creator_name
|
|
|
+ FROM t_image_info i
|
|
|
+ LEFT JOIN t_image_category c ON i.image_type = c.id
|
|
|
+ LEFT JOIN t_sys_user u ON i.created_by = u.id
|
|
|
+ {where_sql}
|
|
|
+ ORDER BY i.created_time DESC
|
|
|
+ LIMIT %s OFFSET %s
|
|
|
+ """
|
|
|
+ cursor.execute(list_sql, tuple(params + [page_size, offset]))
|
|
|
+ images = cursor.fetchall()
|
|
|
+
|
|
|
+ return {
|
|
|
+ "total": total,
|
|
|
+ "list": images,
|
|
|
+ "page": page,
|
|
|
+ "page_size": page_size
|
|
|
+ }
|
|
|
+ except Exception as e:
|
|
|
+ logger.exception(f"获取图片列表失败: {e}")
|
|
|
+ return {"total": 0, "list": []}
|
|
|
+ finally:
|
|
|
+ cursor.close()
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ async def add_image(self, data: Dict[str, Any], user_id: str) -> Tuple[bool, str]:
|
|
|
+ """保存图片信息"""
|
|
|
+ conn = get_db_connection()
|
|
|
+ if not conn:
|
|
|
+ return False, "数据库连接失败"
|
|
|
+
|
|
|
+ cursor = conn.cursor()
|
|
|
+ try:
|
|
|
+ image_id = str(uuid.uuid4())
|
|
|
+ image_name = data.get('image_name')
|
|
|
+ image_url = data.get('image_url')
|
|
|
+ image_type = data.get('image_type')
|
|
|
+ description = data.get('description')
|
|
|
+
|
|
|
+ if not image_name or not image_url or not image_type:
|
|
|
+ return False, "缺少必要参数"
|
|
|
+
|
|
|
+ sql = """
|
|
|
+ INSERT INTO t_image_info (
|
|
|
+ id, image_name, image_url, image_type, description,
|
|
|
+ created_by, updated_by, created_time, updated_time
|
|
|
+ ) VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
|
|
|
+ """
|
|
|
+ cursor.execute(sql, (image_id, image_name, image_url, image_type, description, user_id, user_id))
|
|
|
+ conn.commit()
|
|
|
+ return True, "保存成功"
|
|
|
+ except Exception as e:
|
|
|
+ logger.exception(f"保存图片信息失败: {e}")
|
|
|
+ conn.rollback()
|
|
|
+ return False, f"保存失败: {str(e)}"
|
|
|
+ finally:
|
|
|
+ cursor.close()
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ async def delete_image(self, image_id: str) -> Tuple[bool, str]:
|
|
|
+ """删除图片"""
|
|
|
+ conn = get_db_connection()
|
|
|
+ if not conn:
|
|
|
+ return False, "数据库连接失败"
|
|
|
+
|
|
|
+ cursor = conn.cursor()
|
|
|
+ try:
|
|
|
+ cursor.execute("DELETE FROM t_image_info WHERE id = %s", (image_id,))
|
|
|
+ conn.commit()
|
|
|
+ return True, "删除成功"
|
|
|
+ except Exception as e:
|
|
|
+ logger.exception(f"删除图片失败: {e}")
|
|
|
+ conn.rollback()
|
|
|
+ return False, f"删除失败: {str(e)}"
|
|
|
+ finally:
|
|
|
+ cursor.close()
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ async def get_upload_url(self, filename: str, content_type: str) -> Tuple[bool, str, Dict[str, Any]]:
|
|
|
+ """获取 MinIO 预签名上传 URL"""
|
|
|
+ try:
|
|
|
+ data = self.minio_manager.get_upload_url(filename, content_type)
|
|
|
+ return True, "成功获取上传链接", data
|
|
|
+ except Exception as e:
|
|
|
+ logger.exception("生成上传链接失败")
|
|
|
+ return False, f"生成上传链接失败: {str(e)}", {}
|
