Blame for file_manager.py

import os
import shutil
import zipfile
import json
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Optional, Tuple
from werkzeug.utils import secure_filename
from config import config
import humanize

class FileManager:
    def __init__(self):
        self.server_dir = config.SERVER_DIR
        self.backups_dir = config.BACKUPS_DIR
        self.uploads_dir = config.UPLOADS_DIR
        
        # Ensure directories exist
        self.backups_dir.mkdir(parents=True, exist_ok=True)
        self.uploads_dir.mkdir(parents=True, exist_ok=True)
        
        # Editable file extensions
        self.editable_extensions = {'.properties', '.json', '.txt', '.yml', '.yaml', '.toml', '.cfg', '.conf'}
        
        # Protected files/folders
        self.protected = {'fabric-server-launch.jar', 'server.jar', 'libraries'}
    
    def list_directory(self, relative_path: str = "") -> Dict:
        """List directory contents"""
        safe_path = self._get_safe_path(relative_path)
        
        if not safe_path.exists():
            return {'success': False, 'message': 'Directory not found'}
        
        if not safe_path.is_dir():
            return {'success': False, 'message': 'Not a directory'}
        
        items = []
        for item in sorted(safe_path.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower())):
            stat = item.stat()
            items.append({
                'name': item.name,
                'is_dir': item.is_dir(),
                'size': humanize.naturalsize(stat.st_size) if item.is_file() else '-',
                'size_bytes': stat.st_size if item.is_file() else 0,
                'modified': datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M'),
                'editable': item.suffix.lower() in self.editable_extensions if item.is_file() else False,
                'protected': item.name in self.protected,
                'path': str(item.relative_to(self.server_dir))
            })
        
        return {
            'success': True,
            'path': relative_path,
            'parent': str(Path(relative_path).parent) if relative_path else None,
            'items': items
        }
    
    def _get_safe_path(self, relative_path: str) -> Path:
        """Get safe absolute path within server directory"""
        # Normalize and secure the path
        clean_path = Path(relative_path).as_posix().lstrip('/')
        full_path = (self.server_dir / clean_path).resolve()
        
        # Ensure path is within server directory
        try:
            full_path.relative_to(self.server_dir)
        except ValueError:
            return self.server_dir
        
        return full_path
    
    def read_file(self, relative_path: str) -> Dict:
        """Read file contents"""
        safe_path = self._get_safe_path(relative_path)
        
        if not safe_path.exists():
            return {'success': False, 'message': 'File not found'}
        
        if not safe_path.is_file():
            return {'success': False, 'message': 'Not a file'}
        
        if safe_path.suffix.lower() not in self.editable_extensions:
            return {'success': False, 'message': 'File type not editable'}
        
        try:
            with open(safe_path, 'r', encoding='utf-8') as f:
                content = f.read()
            
            return {
                'success': True,
                'path': relative_path,
                'content': content,
                'size': humanize.naturalsize(safe_path.stat().st_size)
            }
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def write_file(self, relative_path: str, content: str) -> Dict:
        """Write file contents"""
        safe_path = self._get_safe_path(relative_path)
        
        if safe_path.name in self.protected:
            return {'success': False, 'message': 'Cannot modify protected file'}
        
        if safe_path.suffix.lower() not in self.editable_extensions:
            return {'success': False, 'message': 'File type not editable'}
        
        try:
            # Create backup first
            if safe_path.exists():
                backup_name = f"{safe_path.name}.bak.{datetime.now().strftime('%Y%m%d_%H%M%S')}"
                shutil.copy2(safe_path, safe_path.parent / backup_name)
            
            with open(safe_path, 'w', encoding='utf-8') as f:
                f.write(content)
            
            return {'success': True, 'message': 'File saved successfully'}
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def create_file(self, relative_path: str, filename: str) -> Dict:
        """Create a new empty file"""
        safe_dir = self._get_safe_path(relative_path)
        filename = secure_filename(filename)
        
        if not filename:
            return {'success': False, 'message': 'Invalid filename'}
        
        filepath = safe_dir / filename
        
        if filepath.exists():
            return {'success': False, 'message': 'File already exists'}
        
        try:
            filepath.touch()
            return {'success': True, 'message': f'File {filename} created'}
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def create_directory(self, relative_path: str, dirname: str) -> Dict:
        """Create a new directory"""
        safe_dir = self._get_safe_path(relative_path)
        dirname = secure_filename(dirname)
        
        if not dirname:
            return {'success': False, 'message': 'Invalid directory name'}
        
        dirpath = safe_dir / dirname
        
        if dirpath.exists():
            return {'success': False, 'message': 'Directory already exists'}
        
        try:
            dirpath.mkdir(parents=True)
            return {'success': True, 'message': f'Directory {dirname} created'}
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def delete_item(self, relative_path: str) -> Dict:
        """Delete file or directory"""
        safe_path = self._get_safe_path(relative_path)
        
        if safe_path.name in self.protected:
            return {'success': False, 'message': 'Cannot delete protected item'}
        
        if not safe_path.exists():
            return {'success': False, 'message': 'Item not found'}
        
        try:
            if safe_path.is_dir():
                shutil.rmtree(safe_path)
            else:
                safe_path.unlink()
            
            return {'success': True, 'message': 'Item deleted'}
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def rename_item(self, relative_path: str, new_name: str) -> Dict:
        """Rename file or directory"""
        safe_path = self._get_safe_path(relative_path)
        new_name = secure_filename(new_name)
        
        if safe_path.name in self.protected:
            return {'success': False, 'message': 'Cannot rename protected item'}
        
        if not safe_path.exists():
            return {'success': False, 'message': 'Item not found'}
        
        new_path = safe_path.parent / new_name
        
        if new_path.exists():
            return {'success': False, 'message': 'An item with that name already exists'}
        
        try:
            safe_path.rename(new_path)
            return {'success': True, 'message': 'Item renamed'}
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def create_backup(self, name: str = None) -> Dict:
        """Create a backup of the server"""
        if not name:
            name = f"backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        else:
            name = secure_filename(name)
        
        backup_path = self.backups_dir / f"{name}.zip"
        
        try:
            with zipfile.ZipFile(backup_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
                for root, dirs, files in os.walk(self.server_dir):
                    # Skip libraries folder (too large)
                    dirs[:] = [d for d in dirs if d != 'libraries']
                    
                    for file in files:
                        filepath = Path(root) / file
                        arcname = filepath.relative_to(self.server_dir)
                        zipf.write(filepath, arcname)
            
            return {
                'success': True,
                'message': f'Backup created: {name}.zip',
                'size': humanize.naturalsize(backup_path.stat().st_size)
            }
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def list_backups(self) -> List[Dict]:
        """List all backups"""
        backups = []
        for f in sorted(self.backups_dir.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True):
            if f.suffix == '.zip':
                stat = f.stat()
                backups.append({
                    'name': f.name,
                    'size': humanize.naturalsize(stat.st_size),
                    'created': datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S')
                })
        return backups
    
    def restore_backup(self, backup_name: str) -> Dict:
        """Restore from backup"""
        backup_path = self.backups_dir / secure_filename(backup_name)
        
        if not backup_path.exists():
            return {'success': False, 'message': 'Backup not found'}
        
        try:
            # Extract to server directory
            with zipfile.ZipFile(backup_path, 'r') as zipf:
                zipf.extractall(self.server_dir)
            
            return {'success': True, 'message': f'Restored from {backup_name}'}
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def delete_backup(self, backup_name: str) -> Dict:
        """Delete a backup"""
        backup_path = self.backups_dir / secure_filename(backup_name)
        
        if not backup_path.exists():
            return {'success': False, 'message': 'Backup not found'}
        
        try:
            backup_path.unlink()
            return {'success': True, 'message': 'Backup deleted'}
        except Exception as e:
            return {'success': False, 'message': str(e)}
    
    def download_backup(self, backup_name: str) -> Optional[Path]:
        """Get backup path for download"""
        backup_path = self.backups_dir / secure_filename(backup_name)
        if backup_path.exists():
            return backup_path
        return None
    
    def import_world(self, file) -> Dict:
        """Import world from uploaded zip"""
        if not file or not file.filename:
            return {'success': False, 'message': 'No file provided'}
        
        filename = secure_filename(file.filename)
        if not filename.endswith('.zip'):
            return {'success': False, 'message': 'Only .zip files are allowed'}
        
        upload_path = self.uploads_dir / filename
        
        try:
            file.save(str(upload_path))
            
            # Extract and validate
            with zipfile.ZipFile(upload_path, 'r') as zipf:
                names = zipf.namelist()
                
                # Detect structure
                has_world = any('world/' in n or n.startswith('world') for n in names)
                has_level_dat = any('level.dat' in n for n in names)
                
                if not has_level_dat:
                    upload_path.unlink()
                    return {'success': False, 'message': 'No valid world data found (missing level.dat)'}
                
                # Create backup of current world
                if (self.server_dir / 'world').exists():
                    backup_name = f"pre_import_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
                    self.create_backup(backup_name)
                    shutil.rmtree(self.server_dir / 'world')
                
                # Extract
                zipf.extractall(self.server_dir)
            
            upload_path.unlink()
            return {'success': True, 'message': 'World imported successfully'}
            
        except Exception as e:
            if upload_path.exists():
                upload_path.unlink()
            return {'success': False, 'message': str(e)}
    
    def get_server_properties(self) -> Dict[str, str]:
        """Read server.properties as dict"""
        props_path = self.server_dir / 'server.properties'
        props = {}
        
        if props_path.exists():
            with open(props_path, 'r') as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#') and '=' in line:
                        key, value = line.split('=', 1)
                        props[key.strip()] = value.strip()
        
        return props
    
    def update_server_properties(self, updates: Dict[str, str]) -> Dict:
        """Update server.properties"""
        props_path = self.server_dir / 'server.properties'
        
        try:
            # Read current properties
            lines = []
            if props_path.exists():
                with open(props_path, 'r') as f:
                    lines = f.readlines()
            
            # Update properties
            updated_keys = set()
            new_lines = []
            
            for line in lines:
                stripped = line.strip()
                if stripped and not stripped.startswith('#') and '=' in stripped:
                    key = stripped.split('=', 1)[0].strip()
                    if key in updates:
                        new_lines.append(f"{key}={updates[key]}\n")
                        updated_keys.add(key)
                    else:
                        new_lines.append(line)
                else:
                    new_lines.append(line)
            
            # Add new properties
            for key, value in updates.items():
                if key not in updated_keys:
                    new_lines.append(f"{key}={value}\n")
            
            # Write back
            with open(props_path, 'w') as f:
                f.writelines(new_lines)
            
            return {'success': True, 'message': 'Server properties updated'}
        except Exception as e:
            return {'success': False, 'message': str(e)}