multi_camera/app/camera_manager.py

"""
摄像头管理器
处理摄像头配置和URL生成
"""
import logging
from typing import Any, Dict, List, Optional

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from .config import BASE_URL, CAMERA_URL, CAMERAS

logger = logging.getLogger(__name__)


class CameraManager:
    """摄像头管理器"""
    
    def __init__(self, cameras: Optional[List[Dict[str, Any]]] = None) -> None:
        """
        初始化摄像头管理器
        
        Args:
            cameras: 摄像头配置列表，如果为None则使用全局配置
        """
        self.base_url: str = BASE_URL
        self.camera_url: str = CAMERA_URL
        self.cameras: List[Dict[str, Any]] = cameras if cameras is not None else CAMERAS.copy()
        
        # 初始化请求会话，配置连接池和重试策略
        self.session: requests.Session = requests.Session()
        
        # 配置重试策略
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["HEAD", "GET", "OPTIONS"]
        )
        
        # 配置适配器
        adapter = HTTPAdapter(
            max_retries=retry_strategy,
            pool_connections=10,
            pool_maxsize=10
        )
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
        
        # 配置请求头
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
            'Accept': 'application/json, text/plain, */*',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Content-Type': 'application/json',
            'Connection': 'keep-alive',
        })
    
    def get_camera_url(self, camera_id: int, camera_number: str = 'mixed') -> str:
        """
        根据摄像头ID和编号生成URL
        
        Args:
            camera_id: 摄像头ID
            camera_number: 摄像头编号，默认为'mixed'
            
        Returns:
            摄像头URL
            
        Raises:
            ValueError: 摄像头ID不存在
        """
        camera = next((c for c in self.cameras if c['id'] == camera_id), None)
        if not camera:
            raise ValueError(f"摄像头ID {camera_id} 不存在")
        
        room = camera['room']
        if camera_number == 'mixed':
            return f"{self.camera_url}?room={room}&camera=mixed"
        else:
            return f"{self.camera_url}?room={room}&camera=camera-{camera_number}"
    
    def refresh_camera(self, camera_id: int) -> bool:
        """
        刷新指定摄像头（模拟操作）
        
        Args:
            camera_id: 摄像头ID
            
        Returns:
            是否成功
        """
        logger.info(f"刷新摄像头 {camera_id}")
        return True
    
    def get_all_cameras(self) -> List[Dict[str, Any]]:
        """
        返回所有摄像头配置
        
        Returns:
            摄像头配置列表
        """
        return self.cameras
    
    def check_connection(self, timeout: int = 5) -> bool:
        """
        检查连接状态
        
        Args:
            timeout: 超时时间（秒）
            
        Returns:
            连接是否正常
        """
        try:
            response = self.session.get(
                self.base_url, 
                timeout=timeout,
                allow_redirects=True
            )
            response.raise_for_status()
            return response.status_code == 200
        except requests.exceptions.Timeout as e:
            logger.warning(f"连接超时: {e}")
            return False
        except requests.exceptions.ConnectionError as e:
            logger.warning(f"连接错误: {e}")
            return False
        except requests.exceptions.HTTPError as e:
            logger.warning(f"HTTP错误: {e}")
            return False
        except requests.exceptions.RequestException as e:
            logger.error(f"请求异常: {e}")
            return False
    
    def get_camera_by_id(self, camera_id: int) -> Optional[Dict[str, Any]]:
        """
        根据ID获取摄像头配置
        
        Args:
            camera_id: 摄像头ID
            
        Returns:
            摄像头配置，如果不存在返回None
        """
        return next((c for c in self.cameras if c['id'] == camera_id), None)
    
    def close(self) -> None:
        """
        关闭会话
        """
        self.session.close()