app.py

import subprocess
import os
import sys
import asyncio
import pty
import select

# --- CONFIGURATION ---
# Hugging Face Spaces secara default mengekspos port 7860
PORT = 7860 
# ---------------------

def install_dependencies():
    print("Installing dependencies...")
    subprocess.run([sys.executable, "-m", "pip", "install", "websockets", "nest_asyncio"], check=True)
    print("Dependencies installed.")

try:
    import websockets
    import nest_asyncio
except ImportError:
    install_dependencies()
    import websockets
    import nest_asyncio

nest_asyncio.apply()

async def terminal_handler(websocket):
    print("Client connected")
    master_fd, slave_fd = pty.openpty()

    env = os.environ.copy()
    env["TERM"] = "xterm-256color"

    process = subprocess.Popen(
        ["/bin/bash", "-i"],
        preexec_fn=os.setsid,
        stdin=slave_fd,
        stdout=slave_fd,
        stderr=slave_fd,
        env=env
    )
    os.close(slave_fd)
    loop = asyncio.get_event_loop()

    async def read_from_pty():
        while True:
            await asyncio.sleep(0.01)
            try:
                r, w, x = select.select([master_fd], [], [], 0)
                if master_fd in r:
                    output = os.read(master_fd, 10240).decode('utf-8', errors='ignore')
                    if output:
                        await websocket.send(output)
                    else:
                        break
            except:
                break

    async def write_to_pty():
        try:
            async for message in websocket:
                os.write(master_fd, message.encode())
        except:
            pass

    read_task = loop.create_task(read_from_pty())
    write_task = loop.create_task(write_to_pty())
    try:
        await asyncio.gather(read_task, write_task)
    except:
        pass
    finally:
        process.terminate()
        os.close(master_fd)

async def start_server():
    print(f"Starting WebSocket server on port {PORT}...")
    print(f"\033[1;36m[INFO] Server Mode: Interactive PTY directly on HF Space\033[0m")

    # Jalankan server. Hugging Face akan otomatis meneruskan traffic publik ke port ini.
    async with websockets.serve(terminal_handler, "0.0.0.0", PORT):
        print(f"\n\n\033[1;32m >>> SERVER STARTED. CONNECT VIA WSS TO YOUR SPACE URL. \033[0m\n\n")
        await asyncio.Future()

if __name__ == "__main__":
    try:
        asyncio.run(start_server())
    except KeyboardInterrupt:
        print("\nServer stopped.")