Spaces:
Runtime error
Runtime error
File size: 4,086 Bytes
22997db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | import os
import sys
import shutil
import logging
from pathlib import Path
import tempfile
class CacheManager:
@staticmethod
def cleanup_temp():
temp_dir = Path(tempfile.gettempdir())
try:
for temp_file in temp_dir.glob("kaggle*"):
if temp_file.is_file():
os.remove(temp_file)
elif temp_file.is_dir():
shutil.rmtree(temp_file)
except Exception as e:
logging.warning(f"Error cleaning temp files: {e}")
@staticmethod
def cleanup_interrupted_downloads():
try:
download_patterns = ["*.part", "*.download", "*.tmp"]
project_root = Path(__file__).parent.parent.parent
for pattern in download_patterns:
for file in project_root.rglob(pattern):
try:
os.remove(file)
logging.info(f"Removed incomplete download: {file}")
except Exception as e:
logging.warning(f"Could not remove {file}: {e}")
except Exception as e:
logging.warning(f"Error cleaning downloads: {e}")
@staticmethod
def cleanup_python_cache():
try:
project_root = Path(__file__).parent.parent.parent
for cache_dir in project_root.rglob("__pycache__"):
shutil.rmtree(cache_dir, ignore_errors=True)
for pyc_file in project_root.rglob("*.pyc"):
os.remove(pyc_file)
for pyo_file in project_root.rglob("*.pyo"):
os.remove(pyo_file)
init_paths = [
project_root / "src" / "services" / "processors" / "__init__.py",
project_root / "src" / "lib" / "__init__.py"
]
for init_path in init_paths:
if init_path.exists():
try:
os.remove(init_path)
logging.info(f"Removed {init_path}")
except Exception as e:
logging.warning(f"Could not remove {init_path}: {e}")
except Exception as e:
logging.warning(f"Error cleaning Python cache: {e}")
@staticmethod
def reset_sys_modules():
try:
to_remove = []
for module in sys.modules:
if 'src.' in module:
to_remove.append(module)
for module in to_remove:
del sys.modules[module]
except Exception as e:
logging.warning(f"Error resetting modules: {e}")
@staticmethod
def cleanup():
try:
CacheManager.cleanup_python_cache()
CacheManager.cleanup_temp()
CacheManager.cleanup_interrupted_downloads()
CacheManager.reset_sys_modules()
logging.info("Cache cleanup completed successfully")
return True
except Exception as e:
logging.error(f"Cache cleanup failed: {e}")
return False
@staticmethod
def cleanup_data_cache():
try:
project_root = Path(__file__).parent.parent.parent
cache_dir = project_root / "data" / "cache"
if cache_dir.exists() and cache_dir.is_dir():
shutil.rmtree(cache_dir)
logging.info(f"Deleted cache directory: {cache_dir}")
else:
logging.info(f"Cache directory not found: {cache_dir}")
except Exception as e:
logging.warning(f"Error deleting cache directory: {e}")
@staticmethod
def cleanup():
try:
CacheManager.cleanup_python_cache()
CacheManager.cleanup_temp()
CacheManager.cleanup_interrupted_downloads()
CacheManager.cleanup_data_cache()
CacheManager.reset_sys_modules()
logging.info("Cache cleanup completed successfully")
return True
except Exception as e:
logging.error(f"Cache cleanup failed: {e}")
return False |