clean_up.py

import os
import time
from loguru import logger
from constants import MINUTES_KEEP, APP_TMP_DIR


def purge_tmp_directory(
    max_age_minutes: int = MINUTES_KEEP,
    skip_substrings: list[str] | None = None,
    tmp_dir: str = APP_TMP_DIR,
) -> tuple[int, int]:
    """
    Delete files in `tmp_dir` older than `max_age_minutes`, optionally skipping paths
    that contain any substring in `skip_substrings`. Use APP_TMP_DIR so we only
    touch our own temp files (spectrograms, audio, etc.), not other apps' files in /tmp.

    Args:
        max_age_minutes: Files older than this many minutes will be deleted.
        skip_substrings: If provided, any file path containing one of these substrings
            will NOT be deleted.
        tmp_dir: Directory to scan (default: APP_TMP_DIR).

    Returns:
        (removed, skipped): Number of files removed and skipped.
    """
    skip_substrings = skip_substrings or []
    skipped = 0
    removed = 0

    if not os.path.exists(tmp_dir):
        os.makedirs(tmp_dir, exist_ok=True)
        logger.info(f"Temp directory created: {tmp_dir}")
        return removed, skipped

    now = time.time()

    for root, _, files in os.walk(tmp_dir):
        for name in files:
            path = os.path.join(root, name)

            # Skip explicitly excluded paths
            if any(s in path for s in skip_substrings):
                skipped += 1
                continue

            try:
                age_minutes = (now - os.path.getmtime(path)) / 60.0
            except Exception as e:
                logger.warning(f"Could not read mtime for {path}: {e}")
                skipped += 1
                continue

            if age_minutes <= max_age_minutes:
                skipped += 1
                continue

            try:
                os.remove(path)
                removed += 1
            except Exception as e:
                logger.warning(f"Failed to remove file {path}: {e}")
                skipped += 1

    logger.info(f"Temp cleanup complete in {tmp_dir}. Removed {removed} files, skipped {skipped} files.")
    return removed, skipped


def remove_files_related_to(stem: str, tmp_dir: str = APP_TMP_DIR) -> int:
    deleted = 0

    if not stem:
        return 0
    if not os.path.exists(tmp_dir):
        return 0

    for root, _, files in os.walk(tmp_dir):
        for filename in files:
            if stem not in filename:
                continue
            full_path = os.path.join(root, filename)
            try:
                os.remove(full_path)
                deleted += 1
            except Exception as e:
                logger.warning(f"Failed to delete file {full_path}: {e}")

    logger.info(f"Deleted {deleted} files in {tmp_dir} related to '{stem}'.")
    return deleted


def cleanup_previous_run(
    sample_stem: str,
    tmp_dir: str = APP_TMP_DIR,
    max_age_minutes: int = MINUTES_KEEP,
):
    try:
        remove_files_related_to(sample_stem, tmp_dir=tmp_dir)
    except Exception as e:
        print(f"Failed to delete last run with id {sample_stem}: {e}")
    purge_tmp_directory(max_age_minutes=max_age_minutes, tmp_dir=tmp_dir)