import os
import pty
import signal
import struct
import fcntl
import termios
import asyncio
import json
import base64
import logging
import time
import uuid
import shutil
import subprocess
import spaces
import httpx
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
from fastapi.responses import HTMLResponse, FileResponse, JSONResponse, StreamingResponse, Response
import uvicorn
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("webterm")
app = FastAPI()
# ──────────────────────────── 全域設定 ────────────────────────────
HOME = os.path.expanduser("~")
ALLOWED_BASES = [os.path.abspath(p) for p in (
"/home/user", "/workspace", "/home/app", HOME
)]
# 斷線後 session 保留的閒置時間（沒有任何分頁接回就會被清除）
SESSION_IDLE_TIMEOUT = 30 * 60 # 30 分鐘
# 每個 session 保留的畫面回放緩衝區大小上限（bytes）
SESSION_BUFFER_MAX = 300_000
# ──────────────────────────── Frontend ────────────────────────────
HTML_PAGE = r"""
Web Terminal + OpenCode
"""
# ──────────────────────────── Routes ──────────────────────────────
@app.get("/")
async def index():
return HTMLResponse(HTML_PAGE)
def _safe_path(path: str) -> str | None:
abs_path = os.path.abspath(path)
for base in ALLOWED_BASES:
if abs_path == base or abs_path.startswith(base + os.sep):
return abs_path
return None
@app.get("/download")
async def download_route(path: str):
safe = _safe_path(path)
if safe is None:
return JSONResponse({"error": "path not allowed"}, status_code=403)
if not os.path.isfile(safe):
return JSONResponse({"error": "file not found"}, status_code=404)
return FileResponse(safe, filename=os.path.basename(safe))
# ──────────────────────────── 背景安裝任務 ────────────────────────────
async def _run(cmd, cwd=None, env=None):
"""非阻塞地执行一个 shell 指令，回传 (returncode, stdout, stderr)。"""
proc = await asyncio.create_subprocess_shell(
cmd, cwd=cwd, env=env,
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc.communicate()
return proc.returncode, stdout.decode(errors="replace"), stderr.decode(errors="replace")
async def install_opencode():
"""安裝 opencode（執行官方安裝腳本）。"""
log.info("正在背景安装 opencode...")
rc, out, err = await _run("curl -fsSL https://opencode.ai/install | bash")
if rc == 0:
log.info("✅ opencode 安裝成功")
else:
log.warning("⚠️ opencode 安裝失敗或已存在: %s", err[-500:])
async def install_editors():
"""安裝 vim / neovim（選用）。"""
log.info("正在安裝 vim / neovim ...")
for prefix in ("", "sudo "):
rc, out, err = await _run(f"{prefix}apt-get update -y && {prefix}apt-get install -y vim neovim")
if rc == 0:
log.info("✅ vim / neovim 安裝成功 (prefix=%r)", prefix)
await _run("update-alternatives --set editor /usr/bin/vim.basic || true")
with open(os.path.join(HOME, ".bashrc"), "a") as f:
f.write('\nexport EDITOR=vim\nalias vi="vim"\n')
return
log.error("❌ vim / neovim 安裝失敗（apt 與 sudo apt 皆失敗），請檢查容器權限與套件來源")
async def run_installpi_script():
"""執行 ~/app/installpi.sh（若存在），並自動設定執行權限。"""
installpi_path = os.path.expanduser("~/app/installpi.sh")
if not os.path.isfile(installpi_path):
log.warning("⚠️ installpi.sh 不存在於 ~/app/installpi.sh，跳過執行")
return False
try:
# 設定執行權限
os.chmod(installpi_path, 0o755)
log.info("✅ 已設定 installpi.sh 執行權限 (755)")
# 執行腳本
log.info("正在執行 ~/app/installpi.sh ...")
rc, out, err = await _run(installpi_path)
if rc == 0:
log.info("✅ installpi.sh 執行成功")
if out:
log.info("輸出: %s", out[:500])
return True
else:
log.error("❌ installpi.sh 執行失敗，returncode=%s", rc)
if err:
log.error("錯誤訊息: %s", err[:500])
return False
except Exception as e:
log.error("❌ 執行 installpi.sh 時發生例外: %s", e)
return False
@spaces.GPU(duration=1, queue=False)
def dummy_gpu():
"""空函數，僅用於讓平台識別 GPU 裝飾器（若無 GPU 可忽略）。"""
return
@app.on_event("startup")
async def startup_event():
dummy_gpu()
# 依序執行：安裝 opencode → 執行 installpi.sh → 安裝編輯器（背景）
await install_opencode()
#await run_installpi_script()
# 安裝編輯器不影響主要流程，放背景執行
asyncio.create_task(install_editors())
# ──────────────────────────── Terminal Session 管理 ──────────────────────────────
# 核心概念：pty（bash 進程）的生命週期跟 WebSocket 連線脫鉤。
# WebSocket 斷線／頁面重新整理時，只是少了一個「訂閱者」，背後的 bash 依然在跑；
# 只有使用者主動關閉分頁（送出 kill 訊息），或閒置太久沒人接回，才會真正砍掉該 session。
SESSIONS: dict[str, "TerminalSession"] = {}
SESSIONS_LOCK = asyncio.Lock()
class TerminalSession:
def __init__(self, session_id: str, pid: int, fd: int, cols: int = 120, rows: int = 30):
self.id = session_id
self.pid = pid
self.fd = fd
self.cols = cols
self.rows = rows
self.buffer = bytearray() # 畫面回放緩衝區，供重連時還原終端機畫面
self.subscribers: dict[int, asyncio.Queue] = {}
self._next_sub_id = 0
self.cleanup_task: asyncio.Task | None = None
def add_subscriber(self):
sid = self._next_sub_id
self._next_sub_id += 1
q: asyncio.Queue = asyncio.Queue()
self.subscribers[sid] = q
if self.cleanup_task:
self.cleanup_task.cancel()
self.cleanup_task = None
return sid, q
def remove_subscriber(self, sid: int, loop: asyncio.AbstractEventLoop):
self.subscribers.pop(sid, None)
if not self.subscribers:
self.cleanup_task = loop.create_task(self._schedule_cleanup(loop))
async def _schedule_cleanup(self, loop: asyncio.AbstractEventLoop):
try:
await asyncio.sleep(SESSION_IDLE_TIMEOUT)
except asyncio.CancelledError:
return
if not self.subscribers:
log.info("session %s 閒置超過 %ds 無人接回，清除", self.id, SESSION_IDLE_TIMEOUT)
await kill_session(self.id, loop)
def feed(self, data: bytes):
self.buffer.extend(data)
overflow = len(self.buffer) - SESSION_BUFFER_MAX
if overflow > 0:
del self.buffer[:overflow]
dead = []
for sid, q in self.subscribers.items():
try:
q.put_nowait(data)
except Exception:
dead.append(sid)
for sid in dead:
self.subscribers.pop(sid, None)
def _set_winsize(fd: int, cols: int, rows: int):
fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
def _fork_shell():
pid, fd = pty.fork()
if pid == 0:
os.environ["TERM"] = "xterm-256color"
os.environ["COLORTERM"] = "truecolor"
os.environ.setdefault("EDITOR", "vim")
opencode_bin = os.path.expanduser("~/.opencode/bin")
current_path = os.environ.get("PATH", "")
if opencode_bin not in current_path:
os.environ["PATH"] = f"{current_path}:{opencode_bin}"
banner = (
'echo -e "\\033[36m歡迎使用 Web Terminal｜vim/nvim 已安裝｜'
'opencode 與 installpi.sh 已執行（請查看日誌）\\033[0m";'
)
bash_cmd = (
f"{banner} "
"export PROMPT_COMMAND='echo -ne \"\\033]0;PWD:$PWD\\007\"'; "
"exec bash --login"
)
os.execvp("bash", ["bash", "-c", bash_cmd])
return pid, fd
async def create_session(cols: int = 120, rows: int = 30) -> TerminalSession:
pid, fd = _fork_shell()
session_id = uuid.uuid4().hex[:12]
sess = TerminalSession(session_id, pid, fd, cols, rows)
_set_winsize(fd, cols, rows)
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
loop = asyncio.get_running_loop()
def _on_read():
try:
data = os.read(fd, 65536)
except (BlockingIOError, OSError):
return
if not data:
# bash 行程已結束（例如使用者在終端機內打了 exit）
try:
loop.remove_reader(fd)
except Exception:
pass
asyncio.create_task(kill_session(session_id, loop, already_dead=True))
return
sess.feed(data)
loop.add_reader(fd, _on_read)
async with SESSIONS_LOCK:
SESSIONS[session_id] = sess
return sess
async def _reap_process(pid: int, loop: asyncio.AbstractEventLoop):
"""非阻塞地終止並回收子行程。"""
for sig in (signal.SIGTERM, signal.SIGKILL):
try:
os.kill(pid, sig)
except ProcessLookupError:
return
for _ in range(20):
await asyncio.sleep(0.1)
try:
wpid, _status = os.waitpid(pid, os.WNOHANG)
if wpid == pid:
return
except ChildProcessError:
return
try:
await loop.run_in_executor(None, os.waitpid, pid, 0)
except Exception:
pass
async def kill_session(session_id: str, loop: asyncio.AbstractEventLoop, already_dead: bool = False):
async with SESSIONS_LOCK:
sess = SESSIONS.pop(session_id, None)
if not sess:
return
try:
loop.remove_reader(sess.fd)
except Exception:
pass
if not already_dead:
await _reap_process(sess.pid, loop)
try:
os.close(sess.fd)
except Exception:
pass
for q in list(sess.subscribers.values()):
try:
q.put_nowait(None)
except Exception:
pass
log.info("session %s 已清除", session_id)
# ──────────────────────────── Terminal WebSocket ──────────────────────────────
@app.websocket("/ws")
async def terminal_ws(websocket: WebSocket):
await websocket.accept()
loop = asyncio.get_running_loop()
requested_id = websocket.query_params.get("session_id")
try:
cols = int(websocket.query_params.get("cols", 120))
rows = int(websocket.query_params.get("rows", 30))
except ValueError:
cols, rows = 120, 30
sess = SESSIONS.get(requested_id) if requested_id else None
reattached = sess is not None
if sess is None:
try:
sess = await create_session(cols, rows)
except OSError as e:
await websocket.send_text(json.dumps({
"type": "system",
"data": f"\r\n\x1b[31m[❌ 無法建立新終端機 (pty.fork 失敗: {e})，請稍後再試或關閉其他分頁]\x1b[0m\r\n",
}))
await websocket.close()
return
# 告知前端這次連線對應的 session id，讓前端存起來，日後斷線重連 / 重新整理都能接回同一個
await websocket.send_text(json.dumps({"type": "session", "id": sess.id, "reattached": reattached}))
if reattached:
if sess.buffer:
try:
await websocket.send_text(json.dumps({
"type": "output",
"data": bytes(sess.buffer).decode("utf-8", errors="replace"),
}))
except Exception:
pass
try:
await websocket.send_text(json.dumps({
"type": "system",
"data": "\r\n\x1b[32m[✅ 已重新連線，畫面已還原]\x1b[0m\r\n",
}))
except Exception:
pass
elif requested_id:
# 前端要求接回某個 session，但伺服器已找不到（逾時被清除或伺服器重啟過）
try:
await websocket.send_text(json.dumps({
"type": "system",
"data": "\r\n\x1b[33m[⚠️ 找不到先前的工作階段（可能已逾時或伺服器重新啟動），已建立新的終端機]\x1b[0m\r\n",
}))
except Exception:
pass
sub_id, queue = sess.add_subscriber()
async def _pty_to_ws():
while True:
chunk = await queue.get()
if chunk is None:
break
try:
await websocket.send_text(
json.dumps({"type": "output", "data": chunk.decode("utf-8", errors="replace")})
)
except Exception:
break
sender = asyncio.create_task(_pty_to_ws())
try:
while True:
raw = await websocket.receive_text()
try:
msg = json.loads(raw)
except json.JSONDecodeError:
continue
mtype = msg.get("type")
if mtype == "input":
try:
os.write(sess.fd, msg["data"].encode("utf-8"))
except OSError:
break
elif mtype == "resize":
try:
sess.cols, sess.rows = msg["cols"], msg["rows"]
_set_winsize(sess.fd, msg["cols"], msg["rows"])
except Exception:
pass
elif mtype == "upload":
filename = os.path.basename(msg.get("filename", "uploaded_file"))
target_dir = msg.get("targetDir", "/home/user")
safe_dir = _safe_path(target_dir) or "/home/user"
filepath = os.path.join(safe_dir, filename)
try:
payload = base64.b64decode(msg["data"])
def _write(path=filepath, data=payload):
with open(path, "wb") as f:
f.write(data)
await asyncio.to_thread(_write)
await websocket.send_text(json.dumps({
"type": "system",
"data": f"\r\n\x1b[32m[✅ 檔案已成功上傳至: {filepath}]\x1b[0m\r\n",
}))
except Exception as e:
await websocket.send_text(json.dumps({
"type": "system",
"data": f"\r\n\x1b[31m[❌ 上傳失敗: {e}]\x1b[0m\r\n",
}))
elif mtype == "kill":
# 使用者主動關閉這個分頁：真正終止此 session（不再保留供重連）
await kill_session(sess.id, loop)
break
except (WebSocketDisconnect, RuntimeError):
pass
except Exception:
log.exception("terminal_ws 循環發生未預期錯誤")
finally:
sender.cancel()
# 注意：這裡「不」殺掉 pty／bash，只是移除這個訂閱者，
# 讓 session 保留在 SESSIONS 裡，等待同一個分頁重連接回。
sess.remove_subscriber(sub_id, loop)
if __name__ == "__main__":
port = int(os.environ.get("PORT", 7860))
print(f"🚀 Terminal running on 0.0.0.0:{port}")
uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")