562 lines
20 KiB
Python
562 lines
20 KiB
Python
"""
|
|
서버/CLI 친화 Logger 모듈 - 파일/콘솔 분리 로깅과 자동 정리
|
|
|
|
권장 로그 레벨:
|
|
- 개발: logging.DEBUG
|
|
- 운영: logging.INFO 또는 logging.WARNING
|
|
|
|
사용 예시:
|
|
```python
|
|
# 기본 사용
|
|
logger = Logger(
|
|
log_file="logs/app.log",
|
|
file_log_level=logging.DEBUG
|
|
)
|
|
|
|
logger.info("서버 시작")
|
|
|
|
# FastAPI에서 사용 (예)
|
|
# from fastapi import FastAPI
|
|
# app = FastAPI()
|
|
# log = Logger(log_file="logs/api.log")
|
|
# @app.get("/")
|
|
# def root():
|
|
# log.info("요청 수신")
|
|
# return {"ok": True}
|
|
```
|
|
"""
|
|
import re
|
|
import logging
|
|
from logging.handlers import TimedRotatingFileHandler
|
|
import os
|
|
import time
|
|
import threading
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
import traceback
|
|
|
|
# 로그 레벨 이름 매핑 (안전하고 호환성 좋은 방법)
|
|
def get_level_name(level):
|
|
"""로그 레벨을 이름으로 변환 (호환성 보장)"""
|
|
level_names = {
|
|
logging.NOTSET: "NOTSET",
|
|
logging.DEBUG: "DEBUG",
|
|
logging.INFO: "INFO",
|
|
logging.WARNING: "WARNING",
|
|
logging.ERROR: "ERROR",
|
|
logging.CRITICAL: "CRITICAL"
|
|
}
|
|
return level_names.get(level, f"Level {level}")
|
|
|
|
|
|
class WindowsSafeRotatingFileHandler(logging.Handler):
|
|
"""
|
|
Windows 호환 로테이팅 파일 핸들러
|
|
|
|
기존 RotatingFileHandler의 문제점:
|
|
- Windows에서 파일이 열려있으면 rename 실패 → 롤링 실패 → 로그 유실
|
|
|
|
해결책:
|
|
- 파일 크기 초과 시 기존 파일을 닫고 새 파일명으로 직접 생성
|
|
- 타임스탬프 기반 파일명으로 충돌 방지
|
|
"""
|
|
|
|
def __init__(self, filename, maxBytes=10*1024*1024, backupCount=50, encoding='utf-8'):
|
|
super().__init__()
|
|
self.baseFilename = os.path.abspath(filename)
|
|
self.maxBytes = maxBytes
|
|
self.backupCount = backupCount
|
|
self.encoding = encoding
|
|
self.stream = None
|
|
self._lock = threading.Lock()
|
|
self._open_file()
|
|
|
|
def _open_file(self):
|
|
"""파일 스트림 열기"""
|
|
try:
|
|
# 디렉토리 생성
|
|
os.makedirs(os.path.dirname(self.baseFilename), exist_ok=True)
|
|
self.stream = open(self.baseFilename, 'a', encoding=self.encoding)
|
|
except Exception as e:
|
|
# 파일 열기 실패 시 stderr로 출력
|
|
import sys
|
|
print(f"[Logger] 파일 열기 실패: {self.baseFilename}, 오류: {e}", file=sys.stderr)
|
|
self.stream = None
|
|
|
|
def _close_file(self):
|
|
"""파일 스트림 닫기"""
|
|
if self.stream:
|
|
try:
|
|
self.stream.flush()
|
|
self.stream.close()
|
|
except Exception:
|
|
pass
|
|
self.stream = None
|
|
|
|
def _should_rollover(self):
|
|
"""롤오버가 필요한지 확인"""
|
|
if self.stream is None:
|
|
return False
|
|
try:
|
|
# 현재 파일 크기 확인
|
|
self.stream.seek(0, 2) # 파일 끝으로 이동
|
|
size = self.stream.tell()
|
|
return size >= self.maxBytes
|
|
except Exception:
|
|
return False
|
|
|
|
def _do_rollover(self):
|
|
"""롤오버 실행 - Windows 안전 방식"""
|
|
self._close_file()
|
|
|
|
try:
|
|
# 타임스탬프 기반 백업 파일명 생성
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
base, ext = os.path.splitext(self.baseFilename)
|
|
backup_name = f"{base}_{timestamp}{ext}"
|
|
|
|
# 기존 파일을 백업 파일로 이름 변경
|
|
if os.path.exists(self.baseFilename):
|
|
try:
|
|
os.rename(self.baseFilename, backup_name)
|
|
except OSError:
|
|
# rename 실패 시 복사 후 삭제 시도
|
|
try:
|
|
import shutil
|
|
shutil.copy2(self.baseFilename, backup_name)
|
|
# 원본 파일 내용 비우기 (삭제 대신)
|
|
with open(self.baseFilename, 'w', encoding=self.encoding) as f:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
# 오래된 백업 파일 정리
|
|
self._cleanup_old_backups()
|
|
|
|
except Exception as e:
|
|
import sys
|
|
print(f"[Logger] 롤오버 실패: {e}", file=sys.stderr)
|
|
|
|
# 새 파일 열기
|
|
self._open_file()
|
|
|
|
def _cleanup_old_backups(self):
|
|
"""오래된 백업 파일 정리"""
|
|
try:
|
|
base_dir = os.path.dirname(self.baseFilename)
|
|
base_name = os.path.basename(self.baseFilename)
|
|
name_without_ext, ext = os.path.splitext(base_name)
|
|
|
|
# 백업 파일 패턴: {name}_{timestamp}.log
|
|
backup_files = []
|
|
for f in os.listdir(base_dir):
|
|
if f.startswith(name_without_ext + '_') and f.endswith(ext):
|
|
full_path = os.path.join(base_dir, f)
|
|
try:
|
|
mtime = os.path.getmtime(full_path)
|
|
backup_files.append((full_path, mtime))
|
|
except Exception:
|
|
pass
|
|
|
|
# 최신순 정렬 후 backupCount 초과분 삭제
|
|
backup_files.sort(key=lambda x: x[1], reverse=True)
|
|
for file_path, _ in backup_files[self.backupCount:]:
|
|
try:
|
|
os.remove(file_path)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
def emit(self, record):
|
|
"""로그 레코드 출력"""
|
|
with self._lock:
|
|
try:
|
|
# 롤오버 체크
|
|
if self._should_rollover():
|
|
self._do_rollover()
|
|
|
|
# 스트림이 없으면 다시 열기 시도
|
|
if self.stream is None:
|
|
self._open_file()
|
|
|
|
if self.stream:
|
|
msg = self.format(record)
|
|
self.stream.write(msg + '\n')
|
|
self.stream.flush()
|
|
except Exception:
|
|
self.handleError(record)
|
|
|
|
def close(self):
|
|
"""핸들러 종료"""
|
|
with self._lock:
|
|
self._close_file()
|
|
super().close()
|
|
|
|
class Logger:
|
|
|
|
def __init__(self, gui_logger=None, log_file="Edit_PartTimer_log.log", logger_name="Edit_PartTimer_log",
|
|
file_log_level=logging.DEBUG, gui_log_level=logging.INFO,
|
|
max_days=3, cleanup_interval=3600):
|
|
"""
|
|
개선된 Logger 초기화 (서버/CLI용)
|
|
:param gui_logger: 선택적 콜백 함수(메시지 수신용)
|
|
:param log_file: 로그 파일 이름
|
|
:param logger_name: 로거 이름
|
|
:param file_log_level: 파일 로거의 로그 레벨
|
|
:param gui_log_level: 콜백 호출 로그 레벨(기본 INFO)
|
|
:param max_days: 로그 파일 보관 일수 (기본 3일)
|
|
:param cleanup_interval: 정리 작업 간격(초, 기본 1시간)
|
|
"""
|
|
self.gui_logger = gui_logger
|
|
self.file_log_level = file_log_level
|
|
self.gui_log_level = gui_log_level
|
|
self.max_days = max_days
|
|
self.cleanup_interval = cleanup_interval
|
|
self.log_dir = Path(log_file).parent
|
|
self.log_base_name = Path(log_file).stem
|
|
|
|
# 로그 디렉토리 생성
|
|
self.log_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 멀티프로세스 환경에서의 로그 파일명 충돌 방지를 위해 PID 추가
|
|
# 윈도우에서는 실행 중인 파일의 이름 변경(롤링) 시 PermissionError가 발생할 수 있으므로
|
|
# 프로세스별로 별도의 로그 파일을 사용하도록 함.
|
|
pid = os.getpid()
|
|
|
|
# 원본 로그 파일명에서 확장자 분리
|
|
if log_file.endswith('.log'):
|
|
base_name = log_file[:-4]
|
|
ext = '.log'
|
|
else:
|
|
base_name = log_file
|
|
ext = '.log'
|
|
|
|
# PID가 포함된 실제 로그 파일명 생성 (예: app_1234.log)
|
|
# 단, 메인 프로세스라고 판단되거나 특정 조건에서는 원본 이름을 유지할 수도 있으나
|
|
# 교착 상태 해결을 위해 무조건 PID를 붙이는 것이 안전함.
|
|
self.actual_log_file = self.log_dir / f"{Path(base_name).stem}_{pid}{ext}"
|
|
|
|
# cleanup을 위한 기본 이름 패턴 설정 (PID 부분 제외하고 매칭)
|
|
# self.log_base_name은 원본 파일의 stem을 유지하여 모든 PID 로그를 관리 대상으로 함
|
|
self.log_base_name = Path(log_file).stem
|
|
|
|
# 로그 설정
|
|
self.logger = logging.getLogger(f"{logger_name}_{pid}") # 로거 이름도 유니크하게
|
|
self.logger.setLevel(file_log_level)
|
|
# 상위 로거로의 전파 방지(중복 출력 방지)
|
|
self.logger.propagate = False
|
|
|
|
# 기존 핸들러 제거 (중복 방지)
|
|
for handler in self.logger.handlers[:]:
|
|
self.logger.removeHandler(handler)
|
|
|
|
# 포맷 설정
|
|
self.simple_format = "[%(asctime)s] [%(levelname)s] %(message)s"
|
|
self.detailed_format = (
|
|
"[%(asctime)s] [%(threadName)s] [%(levelname)s] "
|
|
"[%(filename)s:%(funcName)s:%(lineno)d] %(message)s"
|
|
)
|
|
# 콜백 전용 간결한 포맷 (시간:분:초 + 메시지만)
|
|
# 필요 시 log()에서 Formatter를 생성하여 사용
|
|
|
|
# 핸들러 추가
|
|
self._add_console_handler(file_log_level)
|
|
# 실제 파일명(PID 포함)으로 핸들러 생성
|
|
self._add_file_handler(str(self.actual_log_file), file_log_level)
|
|
|
|
# 자동 정리 스레드 시작
|
|
self._start_cleanup_thread()
|
|
|
|
def _add_console_handler(self, level):
|
|
"""콘솔 핸들러 추가"""
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(level)
|
|
formatter = logging.Formatter(
|
|
self.detailed_format if level <= logging.DEBUG else self.simple_format
|
|
)
|
|
console_handler.setFormatter(formatter)
|
|
self.logger.addHandler(console_handler)
|
|
|
|
def _add_file_handler(self, log_file, level):
|
|
"""파일 크기 기반 로테이팅 + 수명 관리 (Windows 호환)"""
|
|
# 확장자가 .log가 아니면 .log로 변경
|
|
if not log_file.endswith('.log'):
|
|
base_name, _ = os.path.splitext(log_file)
|
|
log_file = base_name + '.log'
|
|
|
|
# Windows 호환 커스텀 핸들러 사용
|
|
# RotatingFileHandler는 Windows에서 파일 잠금 문제로 롤링 실패 가능
|
|
file_handler = WindowsSafeRotatingFileHandler(
|
|
log_file,
|
|
maxBytes=10 * 1024 * 1024, # 10MB
|
|
backupCount=50,
|
|
encoding="utf-8",
|
|
)
|
|
|
|
file_handler.setLevel(level)
|
|
formatter = logging.Formatter(
|
|
self.detailed_format if level <= logging.DEBUG else self.simple_format
|
|
)
|
|
file_handler.setFormatter(formatter)
|
|
self.logger.addHandler(file_handler)
|
|
|
|
# 핸들러 참조 저장 (정리 작업용)
|
|
self.file_handler = file_handler
|
|
|
|
def _start_cleanup_thread(self):
|
|
"""자동 정리 스레드 시작"""
|
|
def cleanup_worker():
|
|
while True:
|
|
try:
|
|
self._cleanup_old_logs()
|
|
time.sleep(self.cleanup_interval)
|
|
except Exception as e:
|
|
# 정리 작업 실패 시에도 계속 동작
|
|
pass
|
|
|
|
cleanup_thread = threading.Thread(target=cleanup_worker, daemon=True)
|
|
cleanup_thread.start()
|
|
|
|
def _cleanup_old_logs(self):
|
|
"""오래된 로그(> max_days) 및 날짜별 최대 10개 초과분 정리"""
|
|
cutoff_date = datetime.now() - timedelta(days=self.max_days)
|
|
log_pattern = f"{self.log_base_name}*.log*" # .log, .log.1 등 모두 포함
|
|
|
|
files = []
|
|
for log_file in self.log_dir.glob(log_pattern):
|
|
try:
|
|
if log_file.name == f"{self.log_base_name}.log":
|
|
continue
|
|
file_mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
|
|
files.append((log_file, file_mtime))
|
|
except Exception:
|
|
continue
|
|
|
|
# 1) 보존 기간 초과 파일 삭제
|
|
for log_file, mtime in files:
|
|
if mtime < cutoff_date:
|
|
try:
|
|
log_file.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
# 2) 날짜별 최대 5개 유지 (최신순 보존)
|
|
from collections import defaultdict
|
|
by_day = defaultdict(list)
|
|
for log_file, mtime in files:
|
|
day_key = mtime.strftime('%Y-%m-%d')
|
|
by_day[day_key].append((log_file, mtime))
|
|
|
|
for day_key, items in by_day.items():
|
|
# 최신순으로 정렬
|
|
items.sort(key=lambda x: x[1], reverse=True)
|
|
for (log_file, _mtime) in items[5:]:
|
|
try:
|
|
if log_file.exists():
|
|
log_file.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
def log(self, message, level=logging.INFO, exc_info=False):
|
|
"""로그 메시지 기록"""
|
|
if exc_info:
|
|
message = f"{message}\n{traceback.format_exc()}"
|
|
|
|
# 호출 위치 정보를 동적으로 추출
|
|
caller_frame = logging.currentframe().f_back
|
|
record = self.logger.makeRecord(
|
|
self.logger.name, level, caller_frame.f_code.co_filename,
|
|
caller_frame.f_lineno, message, None, None, caller_frame.f_code.co_name
|
|
)
|
|
|
|
# 파일/콘솔 핸들러에 메시지 전달
|
|
if level >= self.file_log_level:
|
|
self.logger.handle(record)
|
|
|
|
# 선택적 콜백으로 전달 (간결 포맷 적용)
|
|
if self.gui_logger and level >= self.gui_log_level:
|
|
gui_formatter = logging.Formatter(
|
|
fmt="[%(asctime)s] %(message)s",
|
|
datefmt="%H:%M:%S"
|
|
)
|
|
formatted_message = gui_formatter.format(record)
|
|
try:
|
|
self.gui_logger(formatted_message)
|
|
except Exception:
|
|
# 콜백 오류는 로깅 실패로 간주하지 않음
|
|
pass
|
|
|
|
def set_gui_logger(self, gui_logger, gui_log_level=None):
|
|
"""
|
|
로그 콜백 함수를 설정합니다.
|
|
선택적으로 콜백 로그 레벨도 변경할 수 있습니다.
|
|
"""
|
|
self.gui_logger = gui_logger
|
|
|
|
if gui_log_level is not None:
|
|
self.gui_log_level = gui_log_level
|
|
|
|
def set_gui_log_level(self, level):
|
|
"""콜백 로그 레벨을 동적으로 변경합니다"""
|
|
self.gui_log_level = level
|
|
level_name = get_level_name(level)
|
|
self.logger.info(f"콜백 로그 레벨이 {level_name}로 변경되었습니다")
|
|
|
|
def set_file_log_level(self, level):
|
|
"""파일 로그 레벨을 동적으로 변경합니다"""
|
|
self.file_log_level = level
|
|
self.logger.setLevel(level)
|
|
for handler in self.logger.handlers:
|
|
if isinstance(handler, (logging.FileHandler, TimedRotatingFileHandler)):
|
|
handler.setLevel(level)
|
|
level_name = get_level_name(level)
|
|
self.logger.info(f"파일 로그 레벨이 {level_name}로 변경되었습니다")
|
|
|
|
def get_log_levels(self):
|
|
"""현재 로그 레벨 정보 반환"""
|
|
return {
|
|
"file_level": get_level_name(self.file_log_level),
|
|
"gui_level": get_level_name(self.gui_log_level),
|
|
"logger_level": get_level_name(self.logger.level)
|
|
}
|
|
|
|
def get_log_info(self):
|
|
"""로그 파일 정보 반환"""
|
|
try:
|
|
log_files = list(self.log_dir.glob(f"{self.log_base_name}*.log"))
|
|
total_size = sum(f.stat().st_size for f in log_files)
|
|
|
|
return {
|
|
"log_dir": str(self.log_dir),
|
|
"total_files": len(log_files),
|
|
"total_size_mb": total_size / (1024 * 1024),
|
|
"max_days": self.max_days,
|
|
"files": [f.name for f in log_files]
|
|
}
|
|
except Exception:
|
|
return {"error": "로그 정보 조회 실패"}
|
|
|
|
def force_cleanup(self):
|
|
"""수동 정리 실행"""
|
|
self._cleanup_old_logs()
|
|
|
|
# 파이썬 표준 로깅 인터페이스 지원
|
|
def debug(self, message, *args, **kwargs):
|
|
"""DEBUG 레벨 로그"""
|
|
if args:
|
|
message = message % args
|
|
self.log(message, level=logging.DEBUG, exc_info=kwargs.get('exc_info', False))
|
|
|
|
def info(self, message, *args, **kwargs):
|
|
"""INFO 레벨 로그"""
|
|
if args:
|
|
message = message % args
|
|
self.log(message, level=logging.INFO, exc_info=kwargs.get('exc_info', False))
|
|
|
|
def warning(self, message, *args, **kwargs):
|
|
"""WARNING 레벨 로그"""
|
|
if args:
|
|
message = message % args
|
|
self.log(message, level=logging.WARNING, exc_info=kwargs.get('exc_info', False))
|
|
|
|
def error(self, message, *args, **kwargs):
|
|
"""ERROR 레벨 로그"""
|
|
if args:
|
|
message = message % args
|
|
self.log(message, level=logging.ERROR, exc_info=kwargs.get('exc_info', False))
|
|
|
|
def critical(self, message, *args, **kwargs):
|
|
"""CRITICAL 레벨 로그"""
|
|
if args:
|
|
message = message % args
|
|
self.log(message, level=logging.CRITICAL, exc_info=kwargs.get('exc_info', False))
|
|
|
|
def exception(self, message, *args, **kwargs):
|
|
"""ERROR 레벨로 예외 정보와 함께 로그"""
|
|
if args:
|
|
message = message % args
|
|
self.log(message, level=logging.ERROR, exc_info=True)
|
|
|
|
# 전역 기본 로거 인스턴스
|
|
_default_logger = None
|
|
|
|
def get_default_logger():
|
|
"""기본 로거 인스턴스 반환"""
|
|
global _default_logger
|
|
if _default_logger is None:
|
|
_default_logger = Logger(log_file="logs/default.log")
|
|
return _default_logger
|
|
|
|
def set_default_logger(logger):
|
|
"""기본 로거 설정"""
|
|
global _default_logger
|
|
_default_logger = logger
|
|
|
|
# 편의 함수들 - 실제 구현
|
|
def debug(msg, *args, **kwargs):
|
|
"""편의 함수 - DEBUG 레벨 로그"""
|
|
get_default_logger().debug(msg, *args, **kwargs)
|
|
|
|
def info(msg, *args, **kwargs):
|
|
"""편의 함수 - INFO 레벨 로그"""
|
|
get_default_logger().info(msg, *args, **kwargs)
|
|
|
|
def warning(msg, *args, **kwargs):
|
|
"""편의 함수 - WARNING 레벨 로그"""
|
|
get_default_logger().warning(msg, *args, **kwargs)
|
|
|
|
def error(msg, *args, **kwargs):
|
|
"""편의 함수 - ERROR 레벨 로그"""
|
|
get_default_logger().error(msg, *args, **kwargs)
|
|
|
|
def critical(msg, *args, **kwargs):
|
|
"""편의 함수 - CRITICAL 레벨 로그"""
|
|
get_default_logger().critical(msg, *args, **kwargs)
|
|
|
|
def exception(msg, *args, **kwargs):
|
|
"""편의 함수 - 예외 정보와 함께 로그"""
|
|
get_default_logger().exception(msg, *args, **kwargs)
|
|
|
|
|
|
# 구조화된 로깅을 위한 추가 클래스
|
|
class StructuredLogger(Logger):
|
|
"""JSON 형태의 구조화된 로깅을 지원하는 로거"""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def log_structured(self, event, level=logging.INFO, **context):
|
|
"""구조화된 로그 기록"""
|
|
import json
|
|
|
|
log_data = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"event": event,
|
|
"level": get_level_name(level),
|
|
**context
|
|
}
|
|
|
|
message = json.dumps(log_data, ensure_ascii=False, separators=(',', ':'))
|
|
self.log(message, level)
|
|
|
|
def log_performance(self, operation, duration, **context):
|
|
"""성능 로그 기록"""
|
|
self.log_structured(
|
|
"performance",
|
|
level=logging.INFO,
|
|
operation=operation,
|
|
duration_ms=round(duration * 1000, 2),
|
|
**context
|
|
)
|
|
|
|
def log_error_with_context(self, error, **context):
|
|
"""에러와 컨텍스트를 함께 로그"""
|
|
self.log_structured(
|
|
"error",
|
|
level=logging.ERROR,
|
|
error_type=type(error).__name__,
|
|
error_message=str(error),
|
|
**context
|
|
)
|