"""
Purely AI generated code that zippes old log files into neet archives. It works fine but only god and the AI know how ;-)
"""
import os
import shutil
import zipfile
from datetime import datetime
from typing import Optional, Tuple
ARCHIVE_DIR_NAME = "archives"
ARCHIVE_TARGET_BYTES = 10 * 1024 * 1024
ARCHIVE_KEEP_LAST = 10
_SESSION_PREFIX = "session_"
_SESSION_TIME_FORMAT = "session_%a_%b_%d_%H_%M_%S_%Y"
_ARCHIVE_TIME_FORMAT = "%Y%m%d_%H%M%S"
[docs]
def _parse_session_datetime(session_name: str) -> Optional[datetime]:
try:
return datetime.strptime(session_name, _SESSION_TIME_FORMAT)
except ValueError:
return None
[docs]
def _collect_sessions(base_path: str):
sessions = []
for entry in os.listdir(base_path):
if not entry.startswith(_SESSION_PREFIX):
continue
full_path = os.path.join(base_path, entry)
if not os.path.isdir(full_path):
continue
session_dt = _parse_session_datetime(entry)
if session_dt is None:
session_dt = datetime.fromtimestamp(os.path.getmtime(full_path))
sessions.append((session_dt, entry, full_path))
sessions.sort(key=lambda item: item[0])
return sessions
[docs]
def _dir_size_bytes(path: str) -> int:
total = 0
for root, _, files in os.walk(path):
for name in files:
try:
total += os.path.getsize(os.path.join(root, name))
except OSError:
continue
return total
[docs]
def _build_archive_name(start_dt: datetime, end_dt: datetime) -> str:
return f"sessions_{start_dt:{_ARCHIVE_TIME_FORMAT}}__{end_dt:{_ARCHIVE_TIME_FORMAT}}.zip"
[docs]
def _parse_archive_name(name: str) -> Optional[Tuple[datetime, datetime]]:
if not (name.startswith("sessions_") and name.endswith(".zip")):
return None
core = name[len("sessions_"):-4]
if "__" not in core:
return None
start_str, end_str = core.split("__", 1)
try:
start_dt = datetime.strptime(start_str, _ARCHIVE_TIME_FORMAT)
end_dt = datetime.strptime(end_str, _ARCHIVE_TIME_FORMAT)
except ValueError:
return None
return start_dt, end_dt
[docs]
def _select_existing_archive(archive_dir: str) -> Optional[Tuple[str, datetime, datetime, int]]:
candidates = []
for entry in os.listdir(archive_dir):
parsed = _parse_archive_name(entry)
if parsed is None:
continue
start_dt, end_dt = parsed
full_path = os.path.join(archive_dir, entry)
if not os.path.isfile(full_path):
continue
size = os.path.getsize(full_path)
candidates.append((end_dt, start_dt, full_path, size))
if not candidates:
return None
candidates.sort(key=lambda item: item[0])
end_dt, start_dt, full_path, size = candidates[-1]
return full_path, start_dt, end_dt, size
[docs]
def _rename_archive_if_needed(archive_dir: str, current_path: str, start_dt: datetime, end_dt: datetime) -> str:
target_name = _build_archive_name(start_dt, end_dt)
target_path = os.path.join(archive_dir, target_name)
if os.path.abspath(target_path) == os.path.abspath(current_path):
return current_path
if os.path.exists(target_path):
suffix = 1
base, ext = os.path.splitext(target_path)
while os.path.exists(f"{base}_{suffix}{ext}"):
suffix += 1
target_path = f"{base}_{suffix}{ext}"
os.replace(current_path, target_path)
return target_path
[docs]
def _archive_session_dir(zip_path: str, base_path: str, session_path: str) -> None:
with zipfile.ZipFile(zip_path, "a", compression=zipfile.ZIP_DEFLATED) as archive:
for root, _, files in os.walk(session_path):
for name in files:
full_path = os.path.join(root, name)
rel_path = os.path.relpath(full_path, base_path)
archive.write(full_path, rel_path)
shutil.rmtree(session_path)
[docs]
def archive_old_sessions(
base_path: str,
keep_last: int = ARCHIVE_KEEP_LAST,
target_bytes: int = ARCHIVE_TARGET_BYTES,
):
sessions = _collect_sessions(base_path)
if len(sessions) <= keep_last:
return
archive_dir = os.path.join(base_path, ARCHIVE_DIR_NAME)
os.makedirs(archive_dir, exist_ok=True)
current_archive = _select_existing_archive(archive_dir)
if current_archive is None or current_archive[3] >= target_bytes:
current_path = None
current_start = None
current_end = None
current_size = 0
else:
current_path, current_start, current_end, current_size = current_archive
for session_dt, _, session_path in sessions[:-keep_last]:
if current_path is None or current_size >= target_bytes:
current_start = session_dt
current_end = session_dt
current_name = _build_archive_name(current_start, current_end)
current_path = os.path.join(archive_dir, current_name)
current_size = 0
session_size = _dir_size_bytes(session_path)
_archive_session_dir(current_path, base_path, session_path)
current_end = session_dt
current_size += session_size
current_path = _rename_archive_if_needed(archive_dir, current_path, current_start, current_end)
