| """ |
| 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 |
| 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" |
| |
| |
| file_name = getattr(file, 'name', 'unknown') |
| |
| |
| 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)}" |
| |
| |
| 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() |
| |
| |
| original_name = getattr(file, 'name', 'uploaded_file') |
| original_name = os.path.basename(original_name) |
| |
| |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| new_name = f"{user_id}_{assignment_type}_{timestamp}_{original_name}" |
| |
| |
| destination = os.path.join(self.upload_dir, new_name) |
| |
| |
| 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 |
|
|