SkillSync / file_handler.py
yekkala's picture
Create file_handler.py
7aaa832 verified
"""
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