utils/file_utils.py

"""File-system helpers."""

from __future__ import annotations

import os
import re
import secrets
import shutil
import string
import time
from pathlib import Path


def ensure_dir(path: str | Path) -> Path:
    p = Path(path)
    p.mkdir(parents=True, exist_ok=True)
    return p


def get_file_extension(filename: str) -> str:
    _, ext = os.path.splitext(filename)
    return ext.lower()


def format_size(size_bytes: int) -> str:
    units = ("B", "KB", "MB", "GB", "TB")
    val = float(size_bytes)
    for unit in units:
        if val < 1024:
            return f"{val:.2f} {unit}"
        val /= 1024
    return f"{val:.2f} PB"


def format_speed(bytes_per_sec: float) -> str:
    return f"{format_size(int(bytes_per_sec))}/s"


def format_duration(seconds: float) -> str:
    m, s = divmod(int(seconds), 60)
    h, m = divmod(m, 60)
    if h:
        return f"{h}h {m}m {s}s"
    if m:
        return f"{m}m {s}s"
    return f"{s}s"


def random_filename(length: int = 10) -> str:
    alphabet = string.ascii_lowercase + string.digits
    return "".join(secrets.choice(alphabet) for _ in range(length))


def sanitize_filename(name: str, max_length: int = 100) -> str:
    """Remove dangerous characters from a filename.

    Strips path separators, parent-directory sequences, and any
    characters that could cause issues on common filesystems.
    """
    name = name.replace("..", "").replace("/", "").replace("\\", "")
    name = re.sub(r'[^\w\.\-\(\) ]', "", name)
    return name[:max_length]


def cleanup_old_files(directory: str | Path, max_age_minutes: int = 30):
    """Delete files older than max_age_minutes inside *directory*."""
    now = time.time()
    cutoff = now - max_age_minutes * 60
    if not Path(directory).exists():
        return
    for entry in Path(directory).iterdir():
        if entry.is_file() and entry.stat().st_mtime < cutoff:
            entry.unlink(missing_ok=True)
        elif entry.is_dir() and entry.stat().st_mtime < cutoff:
            shutil.rmtree(entry, ignore_errors=True)


def smart_disk_cleanup(
    upload_dir: str | Path,
    min_free_mb: int = 500,
    max_age_minutes: int = 60,
    protected_dirs: list[str] | None = None,
) -> dict:
    """تنظيف ذكي لقرص التخزين — يحذف الملفات القديمة عندما تقل المساحة عن الحد.

    Args:
        upload_dir: مسار مجلد الرفع.
        min_free_mb: الحد الأدنى للمساحة الحرة (ميغابايت).
        max_age_minutes: عمر الملف (بالدقائق) الذي يعتبر قديمًا وقابلاً للحذف.
        protected_dirs: قائمة بأسماء المجلدات المحمية (لا تُحذف).

    Returns:
        dict: ملخص التنظيف (deleted_files, freed_bytes, free_mb, status).
    """
    import shutil
    from pathlib import Path
    from shutil import disk_usage

    result = {
        "deleted_files": 0,
        "deleted_dirs": 0,
        "freed_bytes": 0,
        "free_mb": 0,
        "status": "ok",
        "message": "",
    }

    upload_path = Path(upload_dir)
    if not upload_path.exists():
        result["status"] = "no_directory"
        return result

    # فحص المساحة الحالية
    usage = disk_usage(upload_path)
    free_mb = usage.free / (1024 * 1024)
    result["free_mb"] = round(free_mb, 1)

    # إذا كانت المساحة كافية، لا نقوم بالتنظيف
    if free_mb >= min_free_mb:
        result["message"] = f"Enough space: {free_mb:.0f} MB free (min: {min_free_mb} MB)"
        return result

    protected = set(protected_dirs or [])
    protected.add("_db_fallback")  # مجلد قاعدة البيانات الاحتياطية محمي دائماً

    now = time.time()
    cutoff = now - max_age_minutes * 60

    # جمع جميع الملفات والمجلدات المرشحة للحذف
    for entry in sorted(upload_path.iterdir(), key=lambda x: x.stat().st_mtime):
        if entry.name in protected:
            continue
        if not entry.is_file() and not entry.is_dir():
            continue

        # التحقق من العمر
        if entry.stat().st_mtime > cutoff:
            continue

        # حذف الملف
        try:
            if entry.is_file():
                sz = entry.stat().st_size
                entry.unlink(missing_ok=True)
                result["deleted_files"] += 1
                result["freed_bytes"] += sz
            elif entry.is_dir():
                # تحقق أن المجلد ليس قيد الاستخدام (مجلد مستخدم نشط)
                sz = sum(f.stat().st_size for f in entry.rglob("*") if f.is_file())
                shutil.rmtree(entry, ignore_errors=True)
                result["deleted_dirs"] += 1
                result["freed_bytes"] += sz
        except (OSError, PermissionError):
            continue

        # تحقق مرة أخرى بعد كل حذف — إذا أصبحت المساحة كافية، توقف
        if disk_usage(upload_path).free / (1024 * 1024) >= min_free_mb:
            break

    free_after = disk_usage(upload_path).free / (1024 * 1024)
    result["free_mb"] = round(free_after, 1)
    result["message"] = (
        f"Cleaned {result['deleted_files']} files, "
        f"{result['deleted_dirs']} dirs, "
        f"freed {format_size(result['freed_bytes'])}, "
        f"now {free_after:.0f} MB free"
    )
    return result