Ed_
60396f03f8
Applied 236 return type annotations to functions with no return values across 100+ files (core modules, tests, scripts, simulations). Added Phase 4 to python_style_refactor track for remaining 597 items (untyped params, vars, and functions with return values). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
import os
|
|
import shutil
|
|
from datetime import datetime, timedelta
|
|
from log_registry import LogRegistry
|
|
|
|
class LogPruner:
|
|
"""
|
|
Handles the automated deletion of old and insignificant session logs.
|
|
Ensures that only whitelisted or significant sessions (based on size/content)
|
|
are preserved long-term.
|
|
"""
|
|
|
|
def __init__(self, log_registry: LogRegistry, logs_dir: str) -> None:
|
|
"""
|
|
Initializes the LogPruner.
|
|
|
|
Args:
|
|
log_registry: An instance of LogRegistry to check session data.
|
|
logs_dir: The path to the directory containing session sub-directories.
|
|
"""
|
|
self.log_registry = log_registry
|
|
self.logs_dir = logs_dir
|
|
|
|
def prune(self) -> None:
|
|
"""
|
|
Prunes old and small session directories from the logs directory.
|
|
|
|
Deletes session directories that meet the following criteria:
|
|
1. The session start time is older than 24 hours (based on data from LogRegistry).
|
|
2. The session name is NOT in the whitelist provided by the LogRegistry.
|
|
3. The total size of all files within the session directory is less than 2KB (2048 bytes).
|
|
"""
|
|
now = datetime.now()
|
|
cutoff_time = now - timedelta(hours=24)
|
|
# Ensure the base logs directory exists.
|
|
if not os.path.isdir(self.logs_dir):
|
|
return
|
|
# Get sessions that are old and not whitelisted from the registry
|
|
old_sessions_to_check = self.log_registry.get_old_non_whitelisted_sessions(cutoff_time)
|
|
# Prune sessions if their size is less than 2048 bytes
|
|
for session_info in old_sessions_to_check:
|
|
session_id = session_info['session_id']
|
|
session_path = session_info['path']
|
|
if not session_path or not os.path.isdir(session_path):
|
|
continue
|
|
# Calculate total size of files in the directory
|
|
total_size = 0
|
|
try:
|
|
for entry in os.scandir(session_path):
|
|
if entry.is_file():
|
|
total_size += entry.stat().st_size
|
|
except OSError:
|
|
continue
|
|
# Prune if the total size is less than 2KB (2048 bytes)
|
|
if total_size < 2048: # 2KB
|
|
try:
|
|
shutil.rmtree(session_path)
|
|
# print(f"Pruned session '{session_id}' (Size: {total_size} bytes)")
|
|
except OSError:
|
|
pass
|