""" File handling utilities for SkillSync """ import os import shutil from datetime import datetime class FileHandler: """Handles file operations for submissions""" def __init__(self, upload_dir="data/user_data/submissions", max_size_mb=100): self.upload_dir = upload_dir self.max_size = max_size_mb * 1024 * 1024 # Convert to bytes self.allowed_extensions = [".zip", ".pdf", ".ipynb", ".py", ".txt", ".md"] def ensure_upload_dir(self): """Ensure upload directory exists""" os.makedirs(self.upload_dir, exist_ok=True) def validate_file(self, file): """Validate uploaded file""" if file is None: return False, "No file provided" # Get file name file_name = getattr(file, 'name', 'unknown') # Check file extension file_ext = os.path.splitext(file_name)[1].lower() if file_ext not in self.allowed_extensions: return False, f"File type {file_ext} not allowed. Allowed types: {', '.join(self.allowed_extensions)}" # Check file size if available if hasattr(file, 'size') and file.size > self.max_size: return False, f"File too large. Maximum size: {self.max_size // (1024*1024)}MB" return True, "File validated successfully" def save_file(self, file, user_id, assignment_type): """Save uploaded file""" try: self.ensure_upload_dir() # Get original filename original_name = getattr(file, 'name', 'uploaded_file') original_name = os.path.basename(original_name) # Create unique filename timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") new_name = f"{user_id}_{assignment_type}_{timestamp}_{original_name}" # Save file destination = os.path.join(self.upload_dir, new_name) # Copy file to destination if hasattr(file, 'name'): shutil.copy2(file.name, destination) return True, destination except Exception as e: return False, str(e) def get_file_info(self, filepath): """Get information about a file""" if not os.path.exists(filepath): return None stat = os.stat(filepath) return { "path": filepath, "size": stat.st_size, "size_mb": round(stat.st_size / (1024 * 1024), 2), "created": datetime.fromtimestamp(stat.st_ctime).strftime("%Y-%m-%d %H:%M:%S"), "modified": datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M:%S") } def list_submissions(self, user_id=None): """List all submissions""" try: if not os.path.exists(self.upload_dir): return [] files = os.listdir(self.upload_dir) if user_id: files = [f for f in files if f.startswith(str(user_id))] submissions = [] for filename in files: filepath = os.path.join(self.upload_dir, filename) info = self.get_file_info(filepath) if info: submissions.append({ "filename": filename, **info }) return submissions except Exception as e: print(f"Error listing submissions: {e}") return [] def delete_file(self, filepath): """Delete a file""" try: if os.path.exists(filepath): os.remove(filepath) return True, "File deleted successfully" return False, "File not found" except Exception as e: return False, str(e) def get_storage_usage(self): """Get total storage usage""" try: if not os.path.exists(self.upload_dir): return 0 total_size = 0 for filename in os.listdir(self.upload_dir): filepath = os.path.join(self.upload_dir, filename) if os.path.isfile(filepath): total_size += os.path.getsize(filepath) return total_size except Exception as e: print(f"Error calculating storage: {e}") return 0