"""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