computer/libs/python/computer-server/computer_server/handlers/generic.py

"""
Generic handlers for all OSes.

Includes:
- FileHandler

"""

from pathlib import Path
from typing import Dict, Any, Optional
from .base import BaseFileHandler
import base64

def resolve_path(path: str) -> Path:
    """Resolve a path to its absolute path. Expand ~ to the user's home directory.
    
    Args:
        path: The file or directory path to resolve
        
    Returns:
        Path: The resolved absolute path
    """
    return Path(path).expanduser().resolve()

class GenericFileHandler(BaseFileHandler):
    """
    Generic file handler that provides file system operations for all operating systems.
    
    This class implements the BaseFileHandler interface and provides methods for
    file and directory operations including reading, writing, creating, and deleting
    files and directories.
    """
    
    async def file_exists(self, path: str) -> Dict[str, Any]:
        """
        Check if a file exists at the specified path.
        
        Args:
            path: The file path to check
            
        Returns:
            Dict containing 'success' boolean and either 'exists' boolean or 'error' string
        """
        try:
            return {"success": True, "exists": resolve_path(path).is_file()}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def directory_exists(self, path: str) -> Dict[str, Any]:
        """
        Check if a directory exists at the specified path.
        
        Args:
            path: The directory path to check
            
        Returns:
            Dict containing 'success' boolean and either 'exists' boolean or 'error' string
        """
        try:
            return {"success": True, "exists": resolve_path(path).is_dir()}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def list_dir(self, path: str) -> Dict[str, Any]:
        """
        List all files and directories in the specified directory.
        
        Args:
            path: The directory path to list
            
        Returns:
            Dict containing 'success' boolean and either 'files' list of names or 'error' string
        """
        try:
            return {"success": True, "files": [p.name for p in resolve_path(path).iterdir() if p.is_file() or p.is_dir()]}
        except Exception as e:
            return {"success": False, "error": str(e)}
        
    async def read_text(self, path: str) -> Dict[str, Any]:
        """
        Read the contents of a text file.
        
        Args:
            path: The file path to read from
            
        Returns:
            Dict containing 'success' boolean and either 'content' string or 'error' string
        """
        try:
            return {"success": True, "content": resolve_path(path).read_text()}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def write_text(self, path: str, content: str) -> Dict[str, Any]:
        """
        Write text content to a file.
        
        Args:
            path: The file path to write to
            content: The text content to write
            
        Returns:
            Dict containing 'success' boolean and optionally 'error' string
        """
        try:
            resolve_path(path).write_text(content)
            return {"success": True}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def write_bytes(self, path: str, content_b64: str, append: bool = False) -> Dict[str, Any]:
        """
        Write binary content to a file from base64 encoded string.
        
        Args:
            path: The file path to write to
            content_b64: Base64 encoded binary content
            append: If True, append to existing file; if False, overwrite
            
        Returns:
            Dict containing 'success' boolean and optionally 'error' string
        """
        try:
            mode = 'ab' if append else 'wb'
            with open(resolve_path(path), mode) as f:
                f.write(base64.b64decode(content_b64))
            return {"success": True}
        except Exception as e:
            return {"success": False, "error": str(e)}
        
    async def read_bytes(self, path: str, offset: int = 0, length: Optional[int] = None) -> Dict[str, Any]:
        """
        Read binary content from a file and return as base64 encoded string.
        
        Args:
            path: The file path to read from
            offset: Byte offset to start reading from
            length: Number of bytes to read; if None, read entire file from offset
            
        Returns:
            Dict containing 'success' boolean and either 'content_b64' string or 'error' string
        """
        try:
            file_path = resolve_path(path)
            with open(file_path, 'rb') as f:
                if offset > 0:
                    f.seek(offset)
                
                if length is not None:
                    content = f.read(length)
                else:
                    content = f.read()
                
            return {"success": True, "content_b64": base64.b64encode(content).decode('utf-8')}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def get_file_size(self, path: str) -> Dict[str, Any]:
        """
        Get the size of a file in bytes.
        
        Args:
            path: The file path to get size for
            
        Returns:
            Dict containing 'success' boolean and either 'size' integer or 'error' string
        """
        try:
            file_path = resolve_path(path)
            size = file_path.stat().st_size
            return {"success": True, "size": size}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def delete_file(self, path: str) -> Dict[str, Any]:
        """
        Delete a file at the specified path.
        
        Args:
            path: The file path to delete
            
        Returns:
            Dict containing 'success' boolean and optionally 'error' string
        """
        try:
            resolve_path(path).unlink()
            return {"success": True}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def create_dir(self, path: str) -> Dict[str, Any]:
        """
        Create a directory at the specified path.
        
        Creates parent directories if they don't exist and doesn't raise an error
        if the directory already exists.
        
        Args:
            path: The directory path to create
            
        Returns:
            Dict containing 'success' boolean and optionally 'error' string
        """
        try:
            resolve_path(path).mkdir(parents=True, exist_ok=True)
            return {"success": True}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def delete_dir(self, path: str) -> Dict[str, Any]:
        """
        Delete an empty directory at the specified path.
        
        Args:
            path: The directory path to delete
            
        Returns:
            Dict containing 'success' boolean and optionally 'error' string
        """
        try:
            resolve_path(path).rmdir()
            return {"success": True}
        except Exception as e:
            return {"success": False, "error": str(e)}