ehentai-fastapi/performance.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
性能优化模块
"""
import asyncio
import time
from typing import Dict, Any
from functools import wraps

from logger import get_logger

logger = get_logger("performance")


def monitor_performance(func):
    """性能监控装饰器"""
    @wraps(func)
    async def async_wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = await func(*args, **kwargs)
            execution_time = time.time() - start_time
            logger.info(f"{func.__name__} 执行完成，耗时: {execution_time:.2f}秒")
            return result
        except Exception as e:
            execution_time = time.time() - start_time
            logger.error(f"{func.__name__} 执行失败，耗时: {execution_time:.2f}秒，错误: {e}")
            raise
    
    @wraps(func)
    def sync_wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = func(*args, **kwargs)
            execution_time = time.time() - start_time
            logger.info(f"{func.__name__} 执行完成，耗时: {execution_time:.2f}秒")
            return result
        except Exception as e:
            execution_time = time.time() - start_time
            logger.error(f"{func.__name__} 执行失败，耗时: {execution_time:.2f}秒，错误: {e}")
            raise
    
    if asyncio.iscoroutinefunction(func):
        return async_wrapper
    else:
        return sync_wrapper


class PerformanceMonitor:
    """性能监控器"""
    
    def __init__(self):
        self.metrics: Dict[str, Any] = {}
        self.start_time = time.time()
    
    def start_timer(self, name: str):
        """开始计时"""
        self.metrics[name] = {"start": time.time()}
    
    def end_timer(self, name: str):
        """结束计时"""
        if name in self.metrics:
            self.metrics[name]["end"] = time.time()
            self.metrics[name]["duration"] = (
                self.metrics[name]["end"] - self.metrics[name]["start"]
            )
            logger.info(f"{name} 耗时: {self.metrics[name]['duration']:.2f}秒")
    
    def get_summary(self) -> Dict[str, Any]:
        """获取性能摘要"""
        total_time = time.time() - self.start_time
        return {
            "total_time": total_time,
            "metrics": self.metrics,
            "summary": f"总运行时间: {total_time:.2f}秒"
        }


# 全局性能监控器
perf_monitor = PerformanceMonitor()