backend/workspace_api/service.py

import asyncio
import os
import shutil
import subprocess
from typing import Any, Dict, List


class FileExplorerAPI:
    """Provides file system operations for browser workspace integration."""

    def __init__(self, root_dir: str = "."):
        self.root_dir = os.path.abspath(root_dir)

    def _resolve(self, relative_path: str) -> str:
        full_path = os.path.abspath(
            os.path.join(self.root_dir, relative_path)
        )

        if not full_path.startswith(self.root_dir):
            raise PermissionError("Access outside workspace boundary denied")

        return full_path

    def list_files(self, relative_path: str = "") -> List[Dict[str, Any]]:
        target = self._resolve(relative_path)

        if not os.path.exists(target):
            return []

        # Requested path is a file
        if os.path.isfile(target):
            rel = os.path.relpath(target, self.root_dir)
            return [{
                "name": os.path.basename(target),
                "path": rel,
                "is_directory": False,
                "size": os.path.getsize(target),
            }]

        # Requested path is not a directory
        if not os.path.isdir(target):
            return []

        entries: List[Dict[str, Any]] = []

        for name in sorted(os.listdir(target)):
            if name.startswith("."):
                continue

            if name in (
                "__pycache__",
                "node_modules",
                "venv",
                ".venv",
            ):
                continue

            item_path = os.path.join(target, name)

            entries.append({
                "name": name,
                "path": os.path.relpath(item_path, self.root_dir),
                "is_directory": os.path.isdir(item_path),
                "size": (
                    os.path.getsize(item_path)
                    if os.path.isfile(item_path)
                    else 0
                ),
            })

        return entries

    def read_file(self, relative_path: str) -> str:
        target = self._resolve(relative_path)

        with open(target, "r", encoding="utf-8") as f:
            return f.read()

    def write_file(self, relative_path: str, content: str) -> bool:
        target = self._resolve(relative_path)

        os.makedirs(os.path.dirname(target), exist_ok=True)

        with open(target, "w", encoding="utf-8") as f:
            f.write(content)

        return True

    def delete_file(self, relative_path: str) -> bool:
        target = self._resolve(relative_path)

        if os.path.isfile(target):
            os.remove(target)
            return True

        if os.path.isdir(target):
            shutil.rmtree(target)
            return True

        return False


class TerminalAPI:
    """Executes asynchronous terminal operations."""

    def __init__(self, root_dir: str = "."):
        self.root_dir = os.path.abspath(root_dir)

    async def run_command(self, command: str) -> Dict[str, Any]:
        proc = await asyncio.create_subprocess_shell(
            command,
            cwd=self.root_dir,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        stdout, stderr = await proc.communicate()

        return {
            "exit_code": proc.returncode,
            "stdout": stdout.decode(errors="replace"),
            "stderr": stderr.decode(errors="replace"),
        }


class GitAPI:
    """Git wrapper."""

    def __init__(self, root_dir: str = "."):
        self.root_dir = os.path.abspath(root_dir)

    def _git(self, *args) -> Dict[str, Any]:
        res = subprocess.run(
            ["git", *args],
            cwd=self.root_dir,
            capture_output=True,
            text=True,
        )

        return {
            "exit_code": res.returncode,
            "stdout": res.stdout.strip(),
            "stderr": res.stderr.strip(),
        }

    def status(self):
        return self._git("status", "--porcelain")

    def commit(self, message: str):
        self._git("add", "-A")
        return self._git("commit", "-m", message)

    def current_branch(self):
        result = self._git("branch", "--show-current")
        return result["stdout"] or "main"


class BrowserWorkspaceBackend:
    """Workspace backend."""

    def __init__(self, root_dir: str = "."):
        self.file_api = FileExplorerAPI(root_dir)
        self.terminal_api = TerminalAPI(root_dir)
        self.git_api = GitAPI(root_dir)

    def get_build_status(self):
        return {
            "status": "ready",
            "target": "production",
        }

    def get_preview_url(self, port: int = 3000):
        return f"http://localhost:{port}/preview"

    def deploy(self, environment: str = "staging"):
        return {
            "deployment_id": "dep_12345",
            "environment": environment,
            "status": "deployed",
        }