imetting/backend/app/services/terminal_service.py

from typing import List, Optional, Tuple, Dict, Any
from app.core.database import get_db_connection
from app.models.models import Terminal, CreateTerminalRequest, UpdateTerminalRequest
import datetime

class TerminalService:
    def get_terminals(self, 
                      page: int = 1, 
                      size: int = 20, 
                      keyword: Optional[str] = None, 
                      terminal_type: Optional[str] = None,
                      status: Optional[int] = None) -> Tuple[List[Dict[str, Any]], int]:
        """
        获取终端列表，支持分页和筛选
        """
        with get_db_connection() as conn:
            cursor = conn.cursor(dictionary=True)
            
            where_clauses = []
            params = []
            
            if keyword:
                where_clauses.append("(t.imei LIKE %s OR t.terminal_name LIKE %s)")
                keyword_param = f"%{keyword}%"
                params.extend([keyword_param, keyword_param])
                
            if terminal_type:
                where_clauses.append("t.terminal_type = %s")
                params.append(terminal_type)
                
            if status is not None:
                where_clauses.append("t.status = %s")
                params.append(status)
                
            where_clause = " AND ".join(where_clauses) if where_clauses else "1=1"
            
            # 计算总数
            count_query = f"SELECT COUNT(*) as total FROM terminals t WHERE {where_clause}"
            cursor.execute(count_query, params)
            total = cursor.fetchone()['total']
            
            # 查询列表
            offset = (page - 1) * size
            list_query = f"""
                SELECT 
                    t.*,
                    u.username as creator_username,
                    dd.label_cn as terminal_type_name
                FROM terminals t
                LEFT JOIN users u ON t.created_by = u.user_id
                LEFT JOIN dict_data dd ON t.terminal_type = dd.dict_code AND dd.dict_type = 'terminal_type'
                WHERE {where_clause}
                ORDER BY t.created_at DESC
                LIMIT %s OFFSET %s
            """
            cursor.execute(list_query, params + [size, offset])
            terminals = cursor.fetchall()
            
            cursor.close()
            return terminals, total

    def get_terminal_by_id(self, terminal_id: int) -> Optional[Dict[str, Any]]:
        """
        根据ID获取终端详情
        """
        with get_db_connection() as conn:
            cursor = conn.cursor(dictionary=True)
            
            query = """
                SELECT 
                    t.*,
                    u.username as creator_username,
                    dd.label_cn as terminal_type_name
                FROM terminals t
                LEFT JOIN users u ON t.created_by = u.user_id
                LEFT JOIN dict_data dd ON t.terminal_type = dd.dict_code AND dd.dict_type = 'terminal_type'
                WHERE t.id = %s
            """
            cursor.execute(query, (terminal_id,))
            terminal = cursor.fetchone()
            
            cursor.close()
            return terminal

    def get_terminal_by_imei(self, imei: str) -> Optional[Dict[str, Any]]:
        """
        根据IMEI获取终端详情
        """
        with get_db_connection() as conn:
            cursor = conn.cursor(dictionary=True)
            cursor.execute("SELECT * FROM terminals WHERE imei = %s", (imei,))
            terminal = cursor.fetchone()
            cursor.close()
            return terminal

    def create_terminal(self, terminal_data: CreateTerminalRequest, user_id: int) -> int:
        """
        创建新终端
        """
        with get_db_connection() as conn:
            cursor = conn.cursor()
            
            query = """
                INSERT INTO terminals (
                    imei, terminal_name, terminal_type, description, status, created_by
                ) VALUES (%s, %s, %s, %s, %s, %s)
            """
            cursor.execute(query, (
                terminal_data.imei,
                terminal_data.terminal_name,
                terminal_data.terminal_type,
                terminal_data.description,
                terminal_data.status,
                user_id
            ))
            
            new_id = cursor.lastrowid
            conn.commit()
            cursor.close()
            return new_id

    def update_terminal(self, terminal_id: int, terminal_data: UpdateTerminalRequest) -> bool:
        """
        更新终端信息
        """
        with get_db_connection() as conn:
            cursor = conn.cursor()
            
            update_fields = []
            params = []
            
            if terminal_data.terminal_name is not None:
                update_fields.append("terminal_name = %s")
                params.append(terminal_data.terminal_name)
                
            if terminal_data.terminal_type is not None:
                update_fields.append("terminal_type = %s")
                params.append(terminal_data.terminal_type)
                
            if terminal_data.description is not None:
                update_fields.append("description = %s")
                params.append(terminal_data.description)
                
            if terminal_data.status is not None:
                update_fields.append("status = %s")
                params.append(terminal_data.status)
                
            if terminal_data.firmware_version is not None:
                update_fields.append("firmware_version = %s")
                params.append(terminal_data.firmware_version)
                
            if terminal_data.mac_address is not None:
                update_fields.append("mac_address = %s")
                params.append(terminal_data.mac_address)
            
            if not update_fields:
                return False
                
            query = f"UPDATE terminals SET {', '.join(update_fields)} WHERE id = %s"
            params.append(terminal_id)
            
            cursor.execute(query, params)
            conn.commit()
            cursor.close()
            return True

    def delete_terminal(self, terminal_id: int) -> bool:
        """
        删除终端
        """
        with get_db_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM terminals WHERE id = %s", (terminal_id,))
            deleted = cursor.rowcount > 0
            conn.commit()
            cursor.close()
            return deleted

    def set_terminal_status(self, terminal_id: int, status: int) -> bool:
        """
        设置终端启用/停用状态
        """
        with get_db_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("UPDATE terminals SET status = %s WHERE id = %s", (status, terminal_id))
            updated = cursor.rowcount > 0
            conn.commit()
            cursor.close()
            return updated

terminal_service = TerminalService()