68453cead8
- 实现 logs/YYYY-MM-DD/ 日期文件夹结构 - 添加 success.log 和 failure.log 专用日志 - 使用 TimedRotatingFileHandler 实现按天切割 - 添加 log_success 和 log_failure 便捷函数 - 集成 markitdown 进行文件转换 - 优化文件存储路径，按项目ID分类存储 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
"""
|
|
Logging Configuration
|
|
日志配置
|
|
"""
|
|
import logging
|
|
import sys
|
|
from datetime import datetime
|
|
from typing import Any
|
|
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
|
|
from pathlib import Path
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
# Log directory
|
|
LOG_DIR = Path("./logs")
|
|
LOG_DIR.mkdir(exist_ok=True)
|
|
|
|
# 日期格式
|
|
LOG_DATE = datetime.now().strftime("%Y-%m-%d")
|
|
|
|
# 当天的日志目录
|
|
CURRENT_LOG_DIR = LOG_DIR / LOG_DATE
|
|
CURRENT_LOG_DIR.mkdir(exist_ok=True)
|
|
|
|
|
|
def get_log_path(filename: str) -> Path:
|
|
"""获取当天的日志文件路径"""
|
|
return CURRENT_LOG_DIR / filename
|
|
|
|
|
|
def setup_logging(name: str = "yg_dataset") -> logging.Logger:
|
|
"""Setup application logging"""
|
|
logger = logging.getLogger(name)
|
|
logger.setLevel(logging.DEBUG if settings.DEBUG else logging.INFO)
|
|
|
|
# Avoid duplicate handlers
|
|
if logger.handlers:
|
|
return logger
|
|
|
|
# Console handler
|
|
console_handler = logging.StreamHandler(sys.stdout)
|
|
console_handler.setLevel(logging.DEBUG if settings.DEBUG else logging.INFO)
|
|
console_formatter = logging.Formatter(
|
|
fmt="%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)d | %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S"
|
|
)
|
|
console_handler.setFormatter(console_formatter)
|
|
logger.addHandler(console_handler)
|
|
|
|
# Main log file handler - app.log
|
|
main_file_handler = TimedRotatingFileHandler(
|
|
get_log_path("app.log"),
|
|
when="midnight",
|
|
interval=1,
|
|
backupCount=30,
|
|
encoding="utf-8"
|
|
)
|
|
main_file_handler.setLevel(logging.INFO)
|
|
main_file_formatter = logging.Formatter(
|
|
fmt="%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)d | %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S"
|
|
)
|
|
main_file_handler.setFormatter(main_file_formatter)
|
|
logger.addHandler(main_file_handler)
|
|
|
|
return logger
|
|
|
|
|
|
# Create default logger
|
|
logger = setup_logging()
|
|
|
|
|
|
# ============== Success Logger ==============
|
|
def get_success_logger() -> logging.Logger:
|
|
"""获取成功日志记录器"""
|
|
success_logger = logging.getLogger("yg_dataset.success")
|
|
if not success_logger.handlers:
|
|
handler = RotatingFileHandler(
|
|
get_log_path("success.log"),
|
|
maxBytes=10 * 1024 * 1024,
|
|
backupCount=30,
|
|
encoding="utf-8"
|
|
)
|
|
handler.setLevel(logging.INFO)
|
|
formatter = logging.Formatter(
|
|
fmt="%(asctime)s | %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S"
|
|
)
|
|
handler.setFormatter(formatter)
|
|
success_logger.addHandler(handler)
|
|
success_logger.setLevel(logging.INFO)
|
|
return success_logger
|
|
|
|
|
|
# ============== Failure Logger ==============
|
|
def get_failure_logger() -> logging.Logger:
|
|
"""获取失败日志记录器"""
|
|
failure_logger = logging.getLogger("yg_dataset.failure")
|
|
if not failure_logger.handlers:
|
|
handler = RotatingFileHandler(
|
|
get_log_path("failure.log"),
|
|
maxBytes=10 * 1024 * 1024,
|
|
backupCount=30,
|
|
encoding="utf-8"
|
|
)
|
|
handler.setLevel(logging.WARNING)
|
|
formatter = logging.Formatter(
|
|
fmt="%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)d | %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S"
|
|
)
|
|
handler.setFormatter(formatter)
|
|
failure_logger.addHandler(handler)
|
|
failure_logger.setLevel(logging.WARNING)
|
|
return failure_logger
|
|
|
|
|
|
# ============== Convenience functions ==============
|
|
def log_success(message: str, **kwargs):
|
|
"""记录成功日志"""
|
|
extra_info = " | ".join([f"{k}={v}" for k, v in kwargs.items()]) if kwargs else ""
|
|
full_message = f"{message} | {extra_info}" if extra_info else message
|
|
get_success_logger().info(full_message)
|
|
|
|
|
|
def log_failure(message: str, **kwargs):
|
|
"""记录失败日志"""
|
|
extra_info = " | ".join([f"{k}={v}" for k, v in kwargs.items()]) if kwargs else ""
|
|
full_message = f"{message} | {extra_info}" if extra_info else message
|
|
get_failure_logger().warning(full_message)
|
|
|
|
|
|
class LoggerMixin:
|
|
"""Mixin to add logging capability to classes"""
|
|
|
|
@property
|
|
def log(self) -> logging.Logger:
|
|
"""Get logger for this class"""
|
|
return logging.getLogger(self.__class__.__module__ + "." + self.__class__.__name__)
|