BLD_sync/log_utils.py

# log_utils.py
# 统一的日志工具模块

import logging
import os
import sys
import shutil
import re
from datetime import datetime
from typing import Optional
import ntfy_utils

# ================= 全局 logger 实例 =================
_logger = None

# ================= 日志格式常量 =================
LOG_FORMAT = '%(asctime)s [%(levelname)s] %(message)s'
DATE_FORMAT = '%Y-%m-%d %H:%M:%S'


class LoggerManager:
    """统一日志管理器"""

    def __init__(self, name, log_prefix="app", log_dir="log"):
        """
        初始化日志管理器

        Args:
            name: logger 名称
            log_prefix: 日志文件前缀（如 app, sync, migration）
            log_dir: 日志目录
        """
        global _logger

        # 创建日志目录
        log_path = os.path.join(os.getcwd(), log_dir)
        os.makedirs(log_path, exist_ok=True)

        # 创建带时间戳的日志文件
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        log_file = os.path.join(log_path, f"{log_prefix}_{timestamp}.log")

        # 创建 logger
        _logger = logging.getLogger(name)
        _logger.setLevel(logging.INFO)
        _logger.handlers = []

        # 文件处理器
        file_handler = logging.FileHandler(log_file, encoding='utf-8')
        file_formatter = logging.Formatter(LOG_FORMAT, DATE_FORMAT)
        file_handler.setFormatter(file_formatter)
        _logger.addHandler(file_handler)

        # 控制台处理器（强制 UTF-8 避免 GBK 编码错误）
        #sys.stdout.reconfigure(encoding='utf-8', errors='replace')
        console_handler = logging.StreamHandler(sys.stdout)
        console_formatter = logging.Formatter(LOG_FORMAT, DATE_FORMAT)
        console_handler.setFormatter(console_formatter)
        _logger.addHandler(console_handler)

        _logger.info(f"日志文件: {log_file}")

        # 归档旧日志
        self.archive_old_logs(log_path, log_file)

    def archive_old_logs(self, log_dir: str, current_log_file: str):
        """将 log 目录下的旧日志移动到 Archive/YYYY-MM/ 子目录

        Args:
            log_dir: 日志目录路径
            current_log_file: 当前正在使用的日志文件路径（不会被移动）
        """
        archive_base = os.path.join(log_dir, "Archive")
        os.makedirs(archive_base, exist_ok=True)

        # 遍历 log 根目录下的所有 .log 文件
        for filename in os.listdir(log_dir):
            if not filename.endswith('.log'):
                continue

            file_path = os.path.join(log_dir, filename)

            # 跳过当前正在使用的日志文件
            if file_path == current_log_file:
                continue

            # 跳过 Archive 目录本身
            if os.path.isdir(file_path):
                continue

            # 从文件名提取日期信息（格式：prefix_YYYYMMDD_HHMMSS.log）
            year_month = self._extract_year_month_from_filename(filename)

            # 如果无法从文件名提取日期，使用文件修改时间
            if not year_month:
                try:
                    stat = os.stat(file_path)
                    mtime = datetime.fromtimestamp(stat.st_mtime)
                    year_month = mtime.strftime("%Y-%m")
                except:
                    year_month = "unknown"

            # 创建年月子目录
            month_dir = os.path.join(archive_base, year_month)
            os.makedirs(month_dir, exist_ok=True)

            # 移动文件到对应的年月目录
            dest_path = os.path.join(month_dir, filename)
            try:
                shutil.move(file_path, dest_path)
                _logger.info(f"已归档: {filename} -> Archive/{year_month}/")
            except Exception as e:
                _logger.warning(f"归档失败 {filename}: {e}")

    def _extract_year_month_from_filename(self, filename: str) -> Optional[str]:
        """从日志文件名中提取年月信息

        支持的格式：
        - prefix_YYYYMMDD_HHMMSS.log
        - incremental_YYYYMMDD_HHMMSS.log
        - full_sync_YYYYMMDD_HHMMSS.log
        - excel_sync_YYYYMMDD_HHMMSS.log

        Returns:
            年月字符串 (格式: YYYY-MM) 或 None
        """
        # 匹配 YYYYMMDD 模式
        match = re.search(r'(\d{4})(\d{2})\d{2}_\d{6}', filename)
        if match:
            year = match.group(1)
            month = match.group(2)
            return f"{year}-{month}"

        return None

    @staticmethod
    def get_logger():
        """获取全局 logger 实例"""
        return _logger


# ================= 统一的日志辅助函数 =================
# 无需 logger 参数，内部使用全局 _logger

def log_success(message):
    """成功消息 - ✅ 发送 ntfy 通知 (default)"""
    _logger.info(f"✅ [成功] {message}")
    ntfy_utils.send_ntfy(f"✅ [成功] {message}", title="同步任务成功", priority="default", tags=["white_check_mark"])


def log_error(message, exc_info=False):
    """错误消息 - ❌ 发送 ntfy 通知 (high)"""
    _logger.error(f"❌ [错误] {message}", exc_info=exc_info)
    ntfy_utils.send_error(f"❌ [错误] {message}")


def log_warning(message):
    """警告消息 - ⚠️  不发送 ntfy 通知"""
    _logger.warning(f"⚠️  [警告] {message}")


def log_critical(message, exc_info=False):
    """严重错误 - 🔥 发送 ntfy 通知 (urgent)"""
    _logger.critical(f"🔥 [严重] {message}", exc_info=exc_info)
    ntfy_utils.send_critical(f"🔥 [严重] {message}")


def log_start(message):
    """启动消息 - 🚀 发送 ntfy 通知 (default)"""
    _logger.info(f"🚀 [启动] {message}")
    ntfy_utils.send_ntfy(f"🚀 [启动] {message}", title="任务启动", priority="default", tags=["rocket"])


def log_complete(message):
    """完成消息 - 🎯 发送 ntfy 通知 (default)"""
    _logger.info(f"🎯 [完成] {message}")
    ntfy_utils.send_ntfy(f"🎯 [完成] {message}", title="任务完成", priority="default", tags=["checkered_flag"])


def log_stop(message):
    """停止消息 - 🛑 发送 ntfy 通知 (default)"""
    _logger.info(f"🛑 [停止] {message}")
    ntfy_utils.send_ntfy(f"🛑 [停止] {message}", title="服务停止", priority="default", tags=["stop_sign"])


def log_info(message):
    """信息消息 - ℹ️  不发送 ntfy 通知"""
    _logger.info(f"ℹ️  {message}")


def log_processing(message):
    """处理中消息 - 🔄 不发送 ntfy 通知"""
    _logger.info(f"🔄 [处理] {message}")


def log_skip(message):
    """跳过消息 - ⏭️  不发送 ntfy 通知"""
    _logger.info(f"⏭️  [跳过] {message}")


def log_file(message):
    """文件操作消息 - 📂 不发送 ntfy 通知"""
    _logger.info(f"📂 [文件] {message}")


def log_database(message):
    """数据库操作消息 - 💾 不发送 ntfy 通知"""
    _logger.info(f"💾 [数据库] {message}")


def log_sync(message):
    """同步操作消息 - 🔃 不发送 ntfy 通知"""
    _logger.info(f"🔃 [同步] {message}")