py默认框架

平常工作日常需要频繁写python脚本，留下一个常用的模板

# template.py
import logging
import json
import time
import functools
import os
from typing import Any, Dict, Optional, Union
from pathlib import Path
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
from concurrent.futures import ThreadPoolExecutor
from threading import Lock
from datetime import datetime

# =============== 日志配置部分 ===============
def setup_logger(log_level=logging.INFO, max_size_mb=50, backup_count=5):
    """
    配置全局日志接口
    
    参数:
        log_level: 日志级别，默认为INFO
        max_size_mb: 单个日志文件最大大小(MB)，默认50MB
        backup_count: 备份文件数量，默认5个
    """
    # 创建logs目录
    log_dir = Path("logs")
    if not log_dir.exists():
        log_dir.mkdir(parents=True, exist_ok=True)
    
    # 基本日志文件名
    log_file = log_dir / "app.log"
    
    # 配置日志格式
    formatter = logging.Formatter(
        '%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    
    # 使用TimedRotatingFileHandler按天滚动 + RotatingFileHandler按大小滚动
    # 大小限制handler
    size_handler = RotatingFileHandler(
        filename=log_file,
        maxBytes=max_size_mb * 1024 * 1024,  # 转换MB为字节
        backupCount=backup_count,
        encoding='utf-8'
    )
    size_handler.setFormatter(formatter)
    
    # 时间滚动handler
    time_handler = TimedRotatingFileHandler(
        filename=log_file,
        when='midnight',  # 每天午夜滚动
        interval=1,       # 间隔为1天
        backupCount=backup_count,
        encoding='utf-8'
    )
    time_handler.setFormatter(formatter)
    time_handler.suffix = "%Y%m%d"  # 设置日志文件后缀格式
    
    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    
    # 获取根日志器并配置
    logger = logging.getLogger()
    logger.setLevel(log_level)
    
    # 清除现有处理器，避免重复
    if logger.handlers:
        logger.handlers.clear()
        
    logger.addHandler(size_handler)
    logger.addHandler(time_handler)
    logger.addHandler(console_handler)
    
    logger.info("日志系统初始化完成 - 按日滚动及大小限制（最大：%dMB）", max_size_mb)
    return logger

# =============== 配置管理 ===============
class ConfigManager:
    _instance = None
    _lock = Lock()
    _config: Dict[str, Any] = {}
    
    def __new__(cls):
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
            return cls._instance
    
    @classmethod
    def load_config(cls, config_path: Union[str, Path]) -> None:
        """加载配置文件"""
        try:
            with open(config_path, 'r', encoding='utf-8') as f:
                cls._config = json.load(f)
        except Exception as e:
            logger.error(f"加载配置文件失败: {e}")
            raise
    
    @classmethod
    def get(cls, key: str, default: Any = None) -> Any:
        """获取配置值"""
        return cls._config.get(key, default)
    
    @classmethod
    def set(cls, key: str, value: Any) -> None:
        """设置配置值"""
        cls._config[key] = value

# =============== 性能计时器装饰器 ===============
def timer(func):
    """函数执行时间计时器装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        logger.debug(f"{func.__name__} 执行时间: {end_time - start_time:.3f}秒")
        return result
    return wrapper

# =============== 缓存装饰器 ===============
def cache(ttl: int = 300):  # 默认缓存5分钟
    """
    函数结果缓存装饰器
    
    参数:
        ttl: 缓存生存时间(秒)
    """
    def decorator(func):
        cache_data = {}
        cache_times = {}
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = str((args, sorted(kwargs.items())))
            current_time = time.time()
            
            # 检查缓存是否存在且未过期
            if key in cache_data and current_time - cache_times[key] < ttl:
                return cache_data[key]
            
            # 计算新结果
            result = func(*args, **kwargs)
            cache_data[key] = result
            cache_times[key] = current_time
            return result
        return wrapper
    return decorator

# =============== 异常处理 ===============
class AppError(Exception):
    """应用基础异常类"""
    def __init__(self, message: str, error_code: str = None):
        self.message = message
        self.error_code = error_code
        super().__init__(self.message)

def handle_exception(func):
    """统一异常处理装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except AppError as e:
            logger.error(f"业务异常: {e.message}, 错误码: {e.error_code}")
            raise
        except Exception as e:
            logger.exception("未预期的异常:")
            raise AppError(f"系统错误: {str(e)}", "SYS_ERROR")
    return wrapper

# =============== 线程池管理 ===============
class ThreadPoolManager:
    _instance = None
    _lock = Lock()
    
    def __new__(cls):
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
                cls._instance.pool = ThreadPoolExecutor(max_workers=os.cpu_count() * 2)
            return cls._instance
    
    def submit(self, fn, *args, **kwargs):
        """提交任务到线程池"""
        return self.pool.submit(fn, *args, **kwargs)

# =============== 文件操作工具 ===============
class FileUtil:
    @staticmethod
    def ensure_dir(directory: Union[str, Path]) -> Path:
        """确保目录存在，如果不存在则创建"""
        path = Path(directory)
        path.mkdir(parents=True, exist_ok=True)
        return path
    
    @staticmethod
    @handle_exception
    def safe_read_json(file_path: Union[str, Path]) -> Dict:
        """安全读取JSON文件"""
        with open(file_path, 'r', encoding='utf-8') as f:
            return json.load(f)
    
    @staticmethod
    @handle_exception
    def safe_write_json(data: Dict, file_path: Union[str, Path]) -> None:
        """安全写入JSON文件"""
        with open(file_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

# =============== 示例使用 ===============
@timer
@cache(ttl=60)
@handle_exception
def example_function(x: int) -> int:
    """示例函数：展示装饰器的使用"""
    time.sleep(1)  # 模拟耗时操作
    if x < 0:
        raise AppError("输入值不能为负数", "INVALID_INPUT")
    return x * 2

if __name__ == "__main__":
    # 初始化日志
    logger = setup_logger()
    
    # 配置管理示例
    config = ConfigManager()
    config.set("app_name", "最佳实践示例")
    
    # 线程池示例
    pool = ThreadPoolManager()
    
    try:
        # 测试示例函数
        result = example_function(10)
        logger.info(f"计算结果: {result}")
        
        # 测试文件操作
        data = {"test": "数据"}
        FileUtil.ensure_dir("data")
        FileUtil.safe_write_json(data, "data/test.json")
        
    except AppError as e:
        logger.error(f"应用错误: {e.message}")
    except Exception as e:
        logger.exception("未知错误:")

单元测试:
可以直接使用ai生成

import unittest
from zz_test import my_fn

# my_fn.py
###
def my_fn():
    return 0
###


class TestMyFunction(unittest.TestCase):
    def test_my_fn_returns_zero(self):
        """测试my_fn函数是否返回0"""
        result = my_fn()
        self.assertEqual(result, 0)
    
    def test_my_fn_returns_integer(self):
        """测试my_fn函数返回值类型是否为整数"""
        result = my_fn()
        self.assertIsInstance(result, int)

if __name__ == '__main__':
    unittest.main()