personal-experiment/logger.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
日志记录模块
OOP设计的日志记录器
"""

import json
from datetime import datetime
from typing import List, Dict, Any, Optional
from pathlib import Path
from dataclasses import dataclass


@dataclass
class LogEvent:
    """日志事件数据类"""
    timestamp: str
    event_type: str
    message: str
    data: Dict[str, Any]
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {
            'timestamp': self.timestamp,
            'type': self.event_type,
            'message': self.message,
            'data': self.data
        }


class Logger:
    """日志记录器类"""
    
    def __init__(self, log_dir: str = "logs", max_log_files: int = 10):
        """
        初始化日志记录器
        
        Args:
            log_dir: 日志目录
            max_log_files: 最大日志文件数
        """
        self.log_dir = Path(log_dir)
        self.max_log_files = max_log_files
        self.events: List[LogEvent] = []
        self.session_start = datetime.now()
        
        self._init_directory()
        self._cleanup_old_logs()
    
    def _init_directory(self) -> None:
        """初始化日志目录"""
        self.log_dir.mkdir(parents=True, exist_ok=True)
    
    def _cleanup_old_logs(self) -> None:
        """清理旧日志文件"""
        log_files = sorted(
            self.log_dir.glob("network_monitor_*.log"),
            key=lambda x: x.stat().st_mtime,
            reverse=True
        )
        
        for old_log in log_files[self.max_log_files:]:
            old_log.unlink()
    
    def _generate_filename(self) -> str:
        """生成日志文件名"""
        timestamp = self.session_start.strftime("%Y%m%d_%H%M%S")
        return f"network_monitor_{timestamp}.log"
    
    def log(self, event_type: str, message: str, **data) -> None:
        """
        记录日志事件
        
        Args:
            event_type: 事件类型
            message: 事件消息
            **data: 附加数据
        """
        event = LogEvent(
            timestamp=datetime.now().isoformat(),
            event_type=event_type,
            message=message,
            data=data
        )
        self.events.append(event)
    
    def log_ping(self, success: bool, latency: float, error: Optional[str] = None) -> None:
        """
        记录ping事件
        
        Args:
            success: 是否成功
            latency: 延迟（毫秒）
            error: 错误信息
        """
        self.log(
            "ping",
            "Ping执行成功" if success else "Ping执行失败",
            success=success,
            latency_ms=latency,
            error=error
        )
    
    def log_status_change(self, status: str, duration: Optional[float] = None) -> None:
        """
        记录状态变化
        
        Args:
            status: 状态（connected/disconnected）
            duration: 持续时间
        """
        message = f"网络状态变化: {status}"
        if duration is not None:
            message += f" (持续{duration:.1f}秒)"
        
        self.log("status_change", message, status=status, duration_seconds=duration)
    
    def log_error(self, error_type: str, error_message: str) -> None:
        """
        记录错误
        
        Args:
            error_type: 错误类型
            error_message: 错误消息
        """
        self.log("error", f"{error_type}: {error_message}", error_type=error_type)
    
    def log_shutdown(self, reason: str) -> None:
        """
        记录关机事件
        
        Args:
            reason: 关机原因
        """
        self.log("shutdown", f"系统关机: {reason}", reason=reason)
    
    def save(self) -> str:
        """
        保存日志到文件
        
        Returns:
            日志文件路径
        """
        log_file = self.log_dir / self._generate_filename()
        
        session = {
            'session_start': self.session_start.isoformat(),
            'session_end': datetime.now().isoformat(),
            'total_events': len(self.events),
            'events': [event.to_dict() for event in self.events]
        }
        
        with open(log_file, 'w', encoding='utf-8') as f:
            json.dump(session, f, ensure_ascii=False, indent=2)
        
        return str(log_file)
    
    def get_statistics(self) -> Dict[str, Any]:
        """
        获取统计信息
        
        Returns:
            统计信息字典
        """
        ping_events = [e for e in self.events if e.event_type == "ping"]
        successful_pings = [e for e in ping_events if e.data.get("success", False)]
        
        return {
            'total_events': len(self.events),
            'total_pings': len(ping_events),
            'successful_pings': len(successful_pings),
            'failed_pings': len(ping_events) - len(successful_pings),
            'success_rate': self._calculate_success_rate(ping_events, successful_pings),
            'session_duration': (datetime.now() - self.session_start).total_seconds()
        }
    
    def _calculate_success_rate(self, all_pings: List[LogEvent], successful: List[LogEvent]) -> str:
        """计算成功率"""
        if not all_pings:
            return "0%"
        return f"{(len(successful) / len(all_pings) * 100):.2f}%"