scripts/log_manager.py - master - lexis/SF_calendar

scripts/log_manager.py
164 строки6 KB
Lexislex
📚 Обновлена документация с новыми функциями
06 сен 2025, 10:20
06 сен 2025, 10:20e470d9f
#!/usr/bin/env python3
"""
Скрипт для управления логами SF Calendar Bot
"""
import os
import sys
import shutil
import gzip
import argparse
from datetime import datetime, timedelta
from pathlib import Path

# Добавляем путь к корневой директории
sys.path.append(os.path.dirname(os.path.dirname(__file__)))

def get_logs_dir():
    """Получение директории с логами"""
    return os.path.join(os.path.dirname(os.path.dirname(__file__)), 'logs')

def get_log_size(log_file):
    """Получение размера лог-файла в MB"""
    if os.path.exists(log_file):
        return os.path.getsize(log_file) / (1024 * 1024)
    return 0

def compress_old_logs(days_old=7, dry_run=False):
    """Сжатие старых лог-файлов"""
    logs_dir = get_logs_dir()
    if not os.path.exists(logs_dir):
        print(f"Директория логов не найдена: {logs_dir}")
        return
    
    cutoff_date = datetime.now() - timedelta(days=days_old)
    compressed_count = 0
    
    for file_path in Path(logs_dir).glob("*.log*"):
        if file_path.is_file():
            # Пропускаем уже сжатые файлы
            if file_path.suffix == '.gz':
                continue
                
            # Проверяем дату модификации
            file_mtime = datetime.fromtimestamp(file_path.stat().st_mtime)
            if file_mtime < cutoff_date:
                compressed_file = f"{file_path}.gz"
                
                if not dry_run:
                    with open(file_path, 'rb') as f_in:
                        with gzip.open(compressed_file, 'wb') as f_out:
                            shutil.copyfileobj(f_in, f_out)
                    os.remove(file_path)
                    print(f"Сжат: {file_path.name} -> {compressed_file}")
                else:
                    print(f"Будет сжат: {file_path.name} -> {compressed_file}")
                
                compressed_count += 1
    
    print(f"Обработано файлов: {compressed_count}")

def clean_old_logs(days_old=30, dry_run=False):
    """Удаление старых лог-файлов"""
    logs_dir = get_logs_dir()
    if not os.path.exists(logs_dir):
        print(f"Директория логов не найдена: {logs_dir}")
        return
    
    cutoff_date = datetime.now() - timedelta(days=days_old)
    deleted_count = 0
    freed_space = 0
    
    for file_path in Path(logs_dir).glob("*.log*"):
        if file_path.is_file():
            file_mtime = datetime.fromtimestamp(file_path.stat().st_mtime)
            if file_mtime < cutoff_date:
                file_size = file_path.stat().st_size
                
                if not dry_run:
                    file_path.unlink()
                    print(f"Удален: {file_path.name} ({file_size / 1024 / 1024:.2f} MB)")
                else:
                    print(f"Будет удален: {file_path.name} ({file_size / 1024 / 1024:.2f} MB)")
                
                deleted_count += 1
                freed_space += file_size
    
    print(f"Удалено файлов: {deleted_count}")
    print(f"Освобождено места: {freed_space / 1024 / 1024:.2f} MB")

def show_log_stats():
    """Показать статистику логов"""
    logs_dir = get_logs_dir()
    if not os.path.exists(logs_dir):
        print(f"Директория логов не найдена: {logs_dir}")
        return
    
    total_size = 0
    file_count = 0
    
    print("📊 Статистика логов:")
    print("-" * 50)
    
    for file_path in sorted(Path(logs_dir).glob("*.log*")):
        if file_path.is_file():
            file_size = file_path.stat().st_size
            file_mtime = datetime.fromtimestamp(file_path.stat().st_mtime)
            total_size += file_size
            file_count += 1
            
            print(f"{file_path.name:30} {file_size / 1024 / 1024:8.2f} MB  {file_mtime.strftime('%Y-%m-%d %H:%M')}")
    
    print("-" * 50)
    print(f"Всего файлов: {file_count}")
    print(f"Общий размер: {total_size / 1024 / 1024:.2f} MB")

def monitor_logs():
    """Мониторинг логов в реальном времени"""
    logs_dir = get_logs_dir()
    main_log = os.path.join(logs_dir, 'sf_calendar.log')
    
    if not os.path.exists(main_log):
        print(f"Основной лог-файл не найден: {main_log}")
        return
    
    print(f"Мониторинг логов: {main_log}")
    print("Нажмите Ctrl+C для выхода")
    print("-" * 50)
    
    try:
        with open(main_log, 'r', encoding='utf-8') as f:
            # Переходим в конец файла
            f.seek(0, 2)
            
            while True:
                line = f.readline()
                if line:
                    print(line.rstrip())
                else:
                    import time
                    time.sleep(0.1)
    except KeyboardInterrupt:
        print("\nМониторинг остановлен")

def main():
    parser = argparse.ArgumentParser(description='Управление логами SF Calendar Bot')
    parser.add_argument('action', choices=['stats', 'compress', 'clean', 'monitor'],
                       help='Действие для выполнения')
    parser.add_argument('--days', type=int, default=7,
                       help='Количество дней для операций (по умолчанию: 7)')
    parser.add_argument('--dry-run', action='store_true',
                       help='Показать что будет сделано без выполнения')
    
    args = parser.parse_args()
    
    if args.action == 'stats':
        show_log_stats()
    elif args.action == 'compress':
        compress_old_logs(args.days, args.dry_run)
    elif args.action == 'clean':
        clean_old_logs(args.days, args.dry_run)
    elif args.action == 'monitor':
        monitor_logs()

if __name__ == '__main__':
    main()