feat: add automatic log archiving by year-month
- Add archive_old_logs() method to LoggerManager in log_utils.py - Logs are automatically moved to Archive/YYYY-MM/ directory when new log is created - Create archive_existing_logs.py script for one-time migration of existing logs - Archive 386 existing log files into organized year-month structure - Keep only current log file in log/ root directory for cleaner management 🤖 Generated with [Qoder][https://lingma.aliyun.com]
This commit is contained in:
92
archive_existing_logs.py
Normal file
92
archive_existing_logs.py
Normal file
@@ -0,0 +1,92 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
一次性脚本:将 log/ 根目录下的所有历史日志按年月移动到 Archive/ 目录
|
||||
|
||||
使用方法:
|
||||
python archive_existing_logs.py
|
||||
"""
|
||||
import os
|
||||
import shutil
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
|
||||
def extract_year_month_from_filename(filename: str) -> Optional[str]:
|
||||
"""从日志文件名中提取年月信息
|
||||
|
||||
支持的格式:
|
||||
- prefix_YYYYMMDD_HHMMSS.log
|
||||
- incremental_YYYYMMDD_HHMMSS.log
|
||||
- full_sync_YYYYMMDD_HHMMSS.log
|
||||
- excel_sync_YYYYMMDD_HHMMSS.log
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
def archive_existing_logs():
|
||||
"""归档现有的所有日志文件到 Archive/YYYY-MM/ 目录"""
|
||||
log_dir = os.path.join(os.getcwd(), "log")
|
||||
archive_base = os.path.join(log_dir, "Archive")
|
||||
|
||||
if not os.path.exists(log_dir):
|
||||
print(f"日志目录不存在: {log_dir}")
|
||||
return
|
||||
|
||||
os.makedirs(archive_base, exist_ok=True)
|
||||
|
||||
moved_count = 0
|
||||
skipped_count = 0
|
||||
|
||||
for filename in os.listdir(log_dir):
|
||||
if not filename.endswith('.log'):
|
||||
continue
|
||||
|
||||
src = os.path.join(log_dir, filename)
|
||||
|
||||
# 跳过目录
|
||||
if os.path.isdir(src):
|
||||
continue
|
||||
|
||||
# 提取年月信息
|
||||
year_month = extract_year_month_from_filename(filename)
|
||||
|
||||
# 如果无法从文件名提取,使用文件修改时间
|
||||
if not year_month:
|
||||
try:
|
||||
stat = os.stat(src)
|
||||
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)
|
||||
|
||||
# 移动文件
|
||||
dst = os.path.join(month_dir, filename)
|
||||
|
||||
try:
|
||||
shutil.move(src, dst)
|
||||
moved_count += 1
|
||||
print(f"[OK] {filename} -> Archive/{year_month}/")
|
||||
except Exception as e:
|
||||
print(f"[FAIL] 移动失败 {filename}: {e}")
|
||||
skipped_count += 1
|
||||
|
||||
print(f"\n完成!")
|
||||
print(f" 已归档: {moved_count} 个文件")
|
||||
print(f" 失败: {skipped_count} 个文件")
|
||||
print(f" 归档路径: {archive_base}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
archive_existing_logs()
|
||||
76
log_utils.py
76
log_utils.py
@@ -4,7 +4,10 @@
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import shutil
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
import ntfy_utils
|
||||
|
||||
# ================= 全局 logger 实例 =================
|
||||
@@ -57,6 +60,79 @@ class LoggerManager:
|
||||
|
||||
_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 实例"""
|
||||
|
||||
Reference in New Issue
Block a user