import os
import pty
import shlex
import shutil
import socket
import subprocess
import sys
import signal
import struct
import fcntl
import termios
import asyncio
import codecs
import json
import base64
import logging
import mimetypes
import time
import uuid
from urllib.parse import quote
import spaces
import httpx
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, File, UploadFile
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()
# ---------------------------------------------------------------- aichat 自架 REST 中繼
# 這是**外掛**：獨立模組 relay.py，只掛上去，不動下面既有的終端機邏輯。
# 掛不起來也不能讓整個 Space 的終端機跟著壞掉，所以整段包在 try 裡。
try:
import relay # noqa: E402
relay.install_cors(app) # 網頁在別的 Space → 跨網域，要放行預檢
app.include_router(relay.router)
print("[app] aichat relay 已掛上：/api/aichat/*")
except Exception as _relay_err: # pragma: no cover
print(f"[app] relay 掛載失敗（終端機照常運作）：{_relay_err!r}")
# ──────────────────────────── 全域設定 ────────────────────────────
HOME = os.path.expanduser("~")
# repo／工作目錄：HF Space 的程式在 /app，但 HOME 可能是 /root，不能假設 $HOME/app。
REPO_DIR = os.path.dirname(os.path.abspath(__file__))
# 檔案瀏覽／下載不設路徑白名單（此服務本身已提供完整終端機權限）。
# 共用分頁：所有裝置／瀏覽器看到同一組分頁，而且每一頁都是「共用＋長存」的 PTY。
# 分頁清單由伺服器統一管理並落地到 TABS_FILE，所以：
# * 不同裝置、斷線數小時後都接回同一個終端機（opencode 也還在原地跑）
# * 任何一頁都不會因為閒置／關閉瀏覽器被清除
# * 只有使用者明確關閉分頁（DELETE /api/tabs/{id}）才會終止該 PTY
PRIMARY_SESSION_ID = "primary" # 第一頁的固定 id
NEW_SESSION_ID = "__new__" # 舊前端要求「開新分頁」時仍相容
MAX_TABS = int(os.environ.get("WEBTERM_MAX_TABS", "12"))
TABS_FILE = os.path.join(HOME, ".webterm_tabs.json")
# 本機測試時可跳過 opencode / 編輯器安裝
SKIP_INSTALL = os.environ.get("WEBTERM_SKIP_INSTALL", "") == "1"
# ──────────────────────────── Frontend ────────────────────────────
HTML_PAGE = r"""
Web Terminal + OpenCode
就緒
"""
# ──────────────────────────── Routes ──────────────────────────────
@app.get("/")
async def index():
return HTMLResponse(HTML_PAGE)
def _safe_path(path: str) -> str | None:
path = os.path.expanduser(os.path.expandvars(path or HOME))
if "\x00" in path:
return None
return os.path.abspath(path)
@app.get("/files")
async def list_files(path: str = "/home/user/app"):
safe = _safe_path(path)
if safe is None:
return JSONResponse({"error": "path not allowed"}, status_code=403)
if not os.path.isdir(safe):
return JSONResponse({"error": "directory not found"}, status_code=404)
try:
entries = []
with os.scandir(safe) as items:
for item in items:
try:
is_dir = item.is_dir(follow_symlinks=False)
if not is_dir and not item.is_file(follow_symlinks=False):
continue
entries.append({
"name": item.name,
"path": os.path.join(safe, item.name),
"is_dir": is_dir,
"size": 0 if is_dir else item.stat(follow_symlinks=False).st_size,
"modified": item.stat(follow_symlinks=False).st_mtime,
})
except OSError:
continue
entries.sort(key=lambda entry: (not entry["is_dir"], entry["name"].casefold()))
return {"path": safe, "entries": entries}
except OSError as exc:
return JSONResponse({"error": str(exc)}, status_code=500)
@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 os.path.isfile(safe):
return FileResponse(safe, filename=os.path.basename(safe))
if not os.path.isdir(safe):
return JSONResponse({"error": "path not found"}, status_code=404)
# 資料夾串流打包成 .tar.gz，避免一次載入記憶體。
normalized = safe.rstrip(os.sep) or os.sep
name = os.path.basename(normalized) or "root"
parent = os.path.dirname(normalized) or os.sep
member = name if normalized != os.sep else "."
archive = f"{name}.tar.gz"
proc = subprocess.Popen(
["tar", "-czf", "-", "-C", parent, member],
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0,
)
async def stream_archive():
try:
while True:
chunk = await asyncio.to_thread(proc.stdout.read, 1 << 20)
if not chunk:
break
yield chunk
finally:
if proc.poll() is None:
proc.kill()
await asyncio.to_thread(proc.wait)
headers = {"Content-Disposition": f"attachment; filename*=UTF-8''{quote(archive)}"}
return StreamingResponse(stream_archive(), media_type="application/gzip", headers=headers)
# ──────────────────────────── 固定桌面分頁路由 ──────────────────────────────
@app.get("/desktop")
async def desktop_page():
"""noVNC 用戶端；固定桌面分頁的 iframe 會載入這一頁。"""
return HTMLResponse(DESKTOP_PAGE)
@app.get("/api/desktop/status")
async def api_desktop_status():
return desktop_manager.status()
@app.post("/api/desktop/restart")
async def api_desktop_restart():
try:
ok = await desktop_manager.restart()
except Exception as exc:
log.exception("[desktop] 重新啟動失敗")
return JSONResponse({"error": f"無法重新啟動桌面：{exc}"}, status_code=500)
return {"ok": ok, **desktop_manager.status()}
@app.post("/api/desktop/upload")
async def api_desktop_upload(file: UploadFile = File(...)):
"""把檔案上傳到共用桌面的 ~/Desktop，讓 LXDE 內可直接看到。"""
name = os.path.basename(file.filename or "").strip() or "uploaded_file"
if name in (".", "..") or "\x00" in name:
return JSONResponse({"error": "invalid filename"}, status_code=400)
upload_dir = os.path.join(HOME, "Desktop")
os.makedirs(upload_dir, exist_ok=True)
safe = os.path.join(upload_dir, name)
stem, ext = os.path.splitext(name)
counter = 1
while os.path.exists(safe):
safe = os.path.join(upload_dir, f"{stem} ({counter}){ext}")
counter += 1
limit = 200 * 1024 * 1024
written = 0
try:
with open(safe, "wb") as out:
while True:
chunk = await file.read(1 << 20)
if not chunk:
break
written += len(chunk)
if written > limit:
raise ValueError("檔案超過 200MB 上限")
out.write(chunk)
except Exception as exc:
try:
os.remove(safe)
except OSError:
pass
return JSONResponse({"error": str(exc)}, status_code=413 if written > limit else 500)
finally:
await file.close()
return {"ok": True, "path": safe, "size": written}
@app.get("/novnc/{asset_path:path}")
async def novnc_asset(asset_path: str):
"""把系統的 noVNC 靜態檔（apt: novnc）提供給瀏覽器；缺少時前端會改用 CDN。"""
root = os.path.realpath(DESKTOP_NOVNC_ROOT)
target = os.path.realpath(os.path.join(root, asset_path))
if not (target == root or target.startswith(root + os.sep)) or not os.path.isfile(target):
return JSONResponse({"error": "not found"}, status_code=404)
media = mimetypes.guess_type(target)[0] or "application/octet-stream"
return FileResponse(target, media_type=media)
@app.websocket("/vnc")
async def vnc_proxy(websocket: WebSocket):
"""把 noVNC 的 WebSocket 流量橋接到本機 x11vnc（只監聽 127.0.0.1）。"""
offered = websocket.headers.get("sec-websocket-protocol", "")
protocols = [p.strip() for p in offered.split(",") if p.strip()]
subprotocol = "binary" if "binary" in protocols else (protocols[0] if protocols else None)
await websocket.accept(subprotocol=subprotocol)
if not await desktop_manager.ensure():
try:
await websocket.close(code=1011, reason="desktop unavailable")
except Exception:
pass
return
try:
reader, writer = await asyncio.open_connection("127.0.0.1", DESKTOP_VNC_PORT)
except OSError as exc:
log.warning("[desktop] VNC 連線失敗：%s", exc)
try:
await websocket.close(code=1011, reason="vnc unavailable")
except Exception:
pass
return
async def ws_to_tcp():
try:
while True:
message = await websocket.receive()
if message.get("type") == "websocket.disconnect":
break
data = message.get("bytes")
if data is None:
text = message.get("text")
data = text.encode("utf-8") if text is not None else b""
if data:
writer.write(data)
await writer.drain()
except (WebSocketDisconnect, RuntimeError, OSError):
pass
async def tcp_to_ws():
try:
while True:
data = await reader.read(65536)
if not data:
break
await websocket.send_bytes(data)
except (WebSocketDisconnect, RuntimeError, OSError):
pass
tasks = [asyncio.create_task(ws_to_tcp()), asyncio.create_task(tcp_to_ws())]
try:
_done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for task in pending:
task.cancel()
finally:
for task in tasks:
task.cancel()
try:
writer.close()
await writer.wait_closed()
except Exception:
pass
try:
await websocket.close()
except Exception:
pass
@app.get("/api/sessions")
async def api_sessions():
"""列出目前伺服器上的終端機工作階段（除錯／確認跨裝置是否共用）。"""
now = time.time()
async with SESSIONS_LOCK:
items = [{
"id": s.id,
"tab_id": s.tab_id,
"primary": s.tab_id == PRIMARY_SESSION_ID,
"pid": s.pid,
"cols": s.cols,
"rows": s.rows,
"subscribers": len(s.subscribers),
"created_at": round(s.created_at, 3),
"last_active": round(s.last_active, 3),
"idle_seconds": round(now - s.last_active, 3),
} for s in SESSIONS.values()]
return {"primary": PRIMARY_SESSION_ID, "max_tabs": MAX_TABS,
"tmux": bool(shutil.which("tmux")), "sessions": items}
@app.get("/api/tabs")
async def api_tabs():
"""共用分頁清單：所有裝置都以此為準。"""
return {"tabs": tabs_snapshot(), "max_tabs": MAX_TABS,
"tmux": bool(shutil.which("tmux"))}
@app.post("/api/tabs")
async def api_new_tab():
try:
await create_shared_tab()
except ValueError as exc:
return JSONResponse({"error": str(exc)}, status_code=409)
except OSError as exc:
return JSONResponse({"error": f"無法建立終端機：{exc}"}, status_code=500)
return {"tabs": tabs_snapshot(), "max_tabs": MAX_TABS,
"tmux": bool(shutil.which("tmux"))}
@app.delete("/api/tabs/{tab_id}")
async def api_delete_tab(tab_id: str):
if tab_id not in TABS:
return JSONResponse({"error": "找不到分頁"}, status_code=404)
if not await delete_shared_tab(tab_id):
return JSONResponse({"error": "至少要保留一個分頁"}, status_code=409)
return {"tabs": tabs_snapshot(), "max_tabs": MAX_TABS,
"tmux": bool(shutil.which("tmux"))}
@app.post("/api/oc")
async def api_oc(request: Request):
"""oc 按鈕：對指定分頁送 Ctrl-C×5，再輸入 oshopencode.py --auto --continue 並按 Enter。
直接透過 tmux send-keys 注入工作階段，不依賴瀏覽器焦點或 WebSocket，確保確實執行。
"""
try:
body = await request.json()
except Exception:
body = {}
tab_id = body.get("tab_id") if isinstance(body, dict) else None
if not isinstance(tab_id, str) or tab_id not in TABS:
tab_id = PRIMARY_SESSION_ID if PRIMARY_SESSION_ID in TABS else (TABS[0] if TABS else PRIMARY_SESSION_ID)
if not shutil.which("tmux"):
return JSONResponse({"error": "tmux 不可用，無法送出 oc"}, status_code=500)
if not await asyncio.to_thread(_ensure_tmux_session, tab_id):
return JSONResponse({"error": "無法建立終端機工作階段"}, status_code=500)
name = _tmux_name(tab_id)
try:
for _ in range(5):
await asyncio.to_thread(_tmux, "send-keys", "-t", name, "C-c")
await asyncio.sleep(0.2)
await asyncio.to_thread(_tmux, "send-keys", "-t", name, "-l",
"oshopencode.py --auto --continue")
await asyncio.sleep(0.1)
await asyncio.to_thread(_tmux, "send-keys", "-t", name, "Enter")
except Exception as exc:
return JSONResponse({"error": f"送出 oc 失敗：{exc!r}"}, status_code=500)
log.info("oc 已送出到分頁 %s", tab_id)
return {"ok": True, "tab_id": tab_id,
"message": "已送出 Ctrl-C×5 並執行 oshopencode.py --auto --continue"}
BOOTSTRAP_LOG = "/tmp/webterm-bootstrap.log"
@app.get("/api/bootstrap/status")
async def api_bootstrap_status():
"""回報環境 bootstrap（s5p / 甲）進度，方便測試與除錯。"""
s5p_dir = os.path.join(REPO_DIR, "s5p")
pi_cfg = os.path.join(HOME, ".pi", "agent", "models.json")
log_tail = ""
try:
with open(BOOTSTRAP_LOG, "rb") as f:
try:
f.seek(-4000, os.SEEK_END)
except OSError:
pass
log_tail = f.read().decode("utf-8", "replace")
except OSError:
pass
try:
disk = shutil.disk_usage(HOME)
disk_free_gib = round(disk.free / (1024 ** 3), 2)
except OSError:
disk_free_gib = None
llama_ok = False
try:
async with httpx.AsyncClient(timeout=2) as client:
r = await client.get("http://127.0.0.1:20000/health")
llama_ok = r.status_code == 200
except Exception:
llama_ok = False
# pi 是否真的能執行（不只是檔案存在）；舊版 Node 20 裝到 pi 1.x 會啟動即崩潰。
pi_bin = shutil.which("pi") or os.path.join(HOME, ".local", "bin", "pi")
pi_version = None
try:
proc = await asyncio.create_subprocess_exec(
pi_bin, "--version",
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT,
)
out, _ = await asyncio.wait_for(proc.communicate(), timeout=20)
if proc.returncode == 0:
pi_version = out.decode(errors="replace").strip().splitlines()[0]
except Exception:
pi_version = None
return {
"s5p_dir": s5p_dir,
"s5p_installed": os.path.isfile(os.path.join(s5p_dir, "oshopencode.py")),
"oshopencode_wrapper": shutil.which("oshopencode.py"),
"pi_config": os.path.isfile(pi_cfg),
"pi_version": pi_version,
"node_version": (await _run("node -v"))[1].strip() or None,
"llama_server_20000": llama_ok,
"disk_free_gib": disk_free_gib,
"bootstrap_log": BOOTSTRAP_LOG,
"bootstrap_log_tail": log_tail,
}
# ──────────────────────────── 背景安裝任務 ────────────────────────────
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:])
def reset_opencode_config():
"""移除 OpenCode 使用者設定，令安裝使用內建預設值。"""
xdg_config = os.environ.get("XDG_CONFIG_HOME", os.path.join(HOME, ".config"))
config_files = [
os.path.join(xdg_config, "opencode", "opencode.json"),
os.path.join(xdg_config, "opencode", "opencode.jsonc"),
os.path.join(xdg_config, "opencode", "opencode.json5"),
os.path.join(HOME, ".opencode", "opencode.json"),
os.path.join(HOME, ".opencode", "opencode.jsonc"),
os.path.join(HOME, ".opencode", "opencode.json5"),
]
for path in config_files:
try:
os.remove(path)
log.info("Removed OpenCode config to restore defaults: %s", path)
except FileNotFoundError:
pass
except OSError as exc:
log.warning("Could not remove OpenCode config %s: %s", path, exc)
# ──────────────────────────── 記憶體守衛（memguard）────────────────────────────
# 規則：opencode 及其啟動的子行程合計不得超過 MEMGUARD_TOTAL_GIB，單一行程不得超過
# MEMGUARD_PER_PROC_GIB；超過只砍子行程，opencode 本體永不砍。
# 規則文字放在 Space repo 的 AGENTS.md，啟動時複製到 ~/.config/opencode/AGENTS.md。
MEMGUARD_TOTAL_GIB = int(os.environ.get("MEMGUARD_TOTAL_GIB", "80"))
MEMGUARD_PER_PROC_GIB = int(os.environ.get("MEMGUARD_PER_PROC_GIB", "64"))
MEMGUARD_PIDFILE = "/tmp/.memguard.pid"
MEMGUARD_LOG = "/var/log/memkill.log"
def _copy_if_changed(src, dst):
"""內容不同才覆寫，避免每次啟動都無謂改動 mtime。"""
if not os.path.isfile(src):
log.warning("memguard: 缺少來源檔 %s", src)
return False
try:
with open(src, "rb") as f:
new = f.read()
if os.path.isfile(dst):
with open(dst, "rb") as f:
if f.read() == new:
return False
os.makedirs(os.path.dirname(dst), exist_ok=True)
with open(dst, "wb") as f:
f.write(new)
return True
except OSError as exc:
log.warning("memguard: 無法寫入 %s: %s", dst, exc)
return False
def _spawn_memguard(guard_path):
if not (os.path.isfile(guard_path) and shutil.which("python3")):
log.warning("memguard: 守衛腳本或 python3 不存在，略過啟動")
return
try:
with open(MEMGUARD_PIDFILE) as f:
os.kill(int(f.read().strip()), signal.SIGKILL)
except Exception:
pass
try:
logf = open(MEMGUARD_LOG, "a")
except OSError:
logf = subprocess.DEVNULL
subprocess.Popen(
["python3", guard_path, str(MEMGUARD_TOTAL_GIB), "2", str(MEMGUARD_PER_PROC_GIB)],
stdout=logf, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,
start_new_session=True,
)
log.info("✅ memguard 已啟動（合計上限 %sGiB / 單一行程 %sGiB，只砍子行程）",
MEMGUARD_TOTAL_GIB, MEMGUARD_PER_PROC_GIB)
def install_memguard():
"""安裝記憶體守衛與 OpenCode 記憶體規則；每次 Space 啟動都會重新套用。"""
here = os.path.dirname(os.path.abspath(__file__))
guard_dst = os.path.join(HOME, "memkill.py")
if _copy_if_changed(os.path.join(here, "memguard.py"), guard_dst):
log.info("✅ memguard.py -> %s", guard_dst)
xdg_config = os.environ.get("XDG_CONFIG_HOME", os.path.join(HOME, ".config"))
agents_dst = os.path.join(xdg_config, "opencode", "AGENTS.md")
if _copy_if_changed(os.path.join(here, "AGENTS.md"), agents_dst):
log.info("✅ AGENTS.md -> %s（opencode 每次啟動都會讀這份規則）", agents_dst)
bashrc = os.path.join(HOME, ".bashrc")
try:
old = ""
if os.path.isfile(bashrc):
with open(bashrc) as f:
old = f.read()
if "memkill.py" not in old:
os.makedirs(os.path.dirname(bashrc), exist_ok=True)
with open(bashrc, "a") as f:
f.write(
"\n# opencode memory guard（合計 %sGiB / 單一 %sGiB，只砍子行程）\n"
'pgrep -f "memkill[.]py" >/dev/null || { setsid nohup python3 %s %s 2 %s '
">>%s 2>&1
LXDE 桌面（共用）
正在連線到共用桌面…
"""
def _desktop_port_open() -> bool:
try:
with socket.create_connection(("127.0.0.1", DESKTOP_VNC_PORT), timeout=0.4):
return True
except OSError:
return False
def _desktop_binaries_present() -> bool:
if not (shutil.which("Xvfb") and shutil.which("x11vnc")):
return False
return any(shutil.which(x) for x in ("startlxde", "lxsession", "openbox"))
class DesktopManager:
"""維持唯一一個 Xvfb + LXDE + x11vnc；死了就補、重複就砍。"""
SESSION_NAMES = ("lxsession", "startlxde", "openbox", "lxpanel", "pcmanfm")
TERM_TITLE = DESKTOP_TERM_TITLE
def __init__(self):
self._lock = asyncio.Lock()
self.pids = {"xvfb": None, "session": None, "vnc": None, "term": None, "clip": []}
self.started_at = 0.0
self.last_error = ""
self._logf = None
self._last_install = 0.0
# ── 基本工具 ──────────────────────────────────────────────
def _log(self, message: str, level: int = logging.INFO):
line = f"[desktop] {message}"
log.log(level, "%s", line)
try:
f = self._logfd()
if hasattr(f, "write"):
f.write(time.strftime("%Y-%m-%d %H:%M:%S ") + line + "\n")
f.flush()
except OSError:
pass
def _logfd(self):
if self._logf is None:
try:
self._logf = open(DESKTOP_LOG, "a", buffering=1)
except OSError:
self._logf = subprocess.DEVNULL
return self._logf
@staticmethod
def _pid_alive(pid) -> bool:
if not pid:
return False
try:
pid = int(pid)
except (TypeError, ValueError):
return False
try:
with open(f"/proc/{pid}/stat", "rb") as f:
data = f.read()
except FileNotFoundError:
return False
except OSError:
try:
os.kill(pid, 0)
return True
except (ProcessLookupError, PermissionError, OSError):
return False
end = data.rfind(b")")
state = data[end + 2:end + 3] if end >= 0 else b""
if state in (b"Z", b"X"):
try:
os.waitpid(pid, os.WNOHANG)
except (ChildProcessError, PermissionError, OSError):
pass
return False
try:
os.kill(pid, 0)
return True
except ProcessLookupError:
return False
except PermissionError:
return True
except OSError:
return False
@staticmethod
def _cmdline(pid: int) -> str:
try:
with open(f"/proc/{pid}/cmdline", "rb") as f:
return f.read().replace(b"\0", b" ").decode("utf-8", "replace").strip()
except OSError:
return ""
def _has_display(self, pid: int) -> bool:
wanted = ("DISPLAY=" + DESKTOP_DISPLAY).encode()
try:
with open(f"/proc/{pid}/environ", "rb") as f:
return wanted in f.read().split(b"\0")
except OSError:
return False
def _env(self) -> dict:
env = os.environ.copy()
env["DISPLAY"] = DESKTOP_DISPLAY
env["XDG_SESSION_TYPE"] = "x11"
env["XDG_CURRENT_DESKTOP"] = "LXDE"
env["DESKTOP_SESSION"] = "LXDE"
env.pop("SESSION_MANAGER", None)
runtime = os.path.join("/tmp", f"xdg-runtime-{os.getuid()}")
try:
os.makedirs(runtime, mode=0o700, exist_ok=True)
os.chmod(runtime, 0o700)
except OSError:
pass
env["XDG_RUNTIME_DIR"] = runtime
return env
def _spawn(self, argv, env=None, cwd=None):
self._log("spawn: " + " ".join(argv))
return subprocess.Popen(
argv, env=env or self._env(), cwd=cwd,
stdin=subprocess.DEVNULL, stdout=self._logfd(), stderr=subprocess.STDOUT,
start_new_session=True, close_fds=True,
)
# ── 掃描／清理 ────────────────────────────────────────────
def _scan(self):
xvfb, vnc, session = [], [], []
try:
entries = os.listdir("/proc")
except OSError:
return xvfb, vnc, session
for entry in entries:
if not entry.isdigit():
continue
pid = int(entry)
if pid == os.getpid():
continue
cmd = self._cmdline(pid)
if not cmd:
continue
low = cmd.lower()
if "xvfb" in low and DESKTOP_DISPLAY in cmd:
xvfb.append(pid)
elif "x11vnc" in low and DESKTOP_DISPLAY in cmd:
vnc.append(pid)
elif any(name in low for name in self.SESSION_NAMES) and (
DESKTOP_DISPLAY in cmd or self._has_display(pid)):
session.append(pid)
return xvfb, vnc, session
def _scan_terms(self):
"""只找我們自己開的「共用終端機」視窗（用標題辨識，不影響使用者自己開的終端機）。"""
pids = []
try:
entries = os.listdir("/proc")
except OSError:
return pids
for entry in entries:
if not entry.isdigit():
continue
pid = int(entry)
if pid == os.getpid():
continue
cmd = self._cmdline(pid)
if not cmd or self.TERM_TITLE not in cmd:
continue
if DESKTOP_DISPLAY in cmd or self._has_display(pid):
pids.append(pid)
return pids
def _scan_clip(self):
"""找 autocutsel：把 x11vnc 使用的 CUT_BUFFER0 與 CLIPBOARD/PRIMARY 雙向同步。"""
pids = []
try:
entries = os.listdir("/proc")
except OSError:
return pids
for entry in entries:
if not entry.isdigit():
continue
pid = int(entry)
if pid == os.getpid():
continue
cmd = self._cmdline(pid)
if not cmd or "autocutsel" not in cmd.lower():
continue
if DESKTOP_DISPLAY in cmd or self._has_display(pid):
pids.append(pid)
return pids
def _term_binary(self):
return shutil.which("xterm") or shutil.which("lxterminal")
def _prepare_session_config(self):
"""關掉容器裡必然失敗的 polkit / 螢幕鎖定 autostart，避免出現
「No session for pid …」的 Error 對話框（Xvfb 沒有 logind session）。"""
autostart = os.path.join(HOME, ".config", "autostart")
try:
os.makedirs(autostart, exist_ok=True)
for name in ("lxpolkit", "light-locker"):
path = os.path.join(autostart, name + ".desktop")
content = (
"[Desktop Entry]\n"
"Type=Application\n"
f"Name={name}\n"
"Exec=/bin/true\n"
"Hidden=true\n"
"NoDisplay=true\n"
"X-GNOME-Autostart-enabled=false\n"
)
if not (os.path.isfile(path) and open(path).read() == content):
with open(path, "w") as f:
f.write(content)
except OSError as exc:
self._log(f"無法寫入 autostart 覆寫：{exc}", logging.WARNING)
system_conf = "/etc/xdg/lxsession/LXDE/desktop.conf"
user_conf = os.path.join(HOME, ".config", "lxsession", "LXDE", "desktop.conf")
source = user_conf if os.path.isfile(user_conf) else system_conf
try:
if not os.path.isfile(source):
return
with open(source) as f:
lines = f.read().splitlines()
changed = False
out = []
for line in lines:
if line.startswith("polkit/command=") and line != "polkit/command=":
line = "polkit/command="
changed = True
elif line.startswith("lock_manager/command=") and line != "lock_manager/command=":
line = "lock_manager/command="
changed = True
out.append(line)
if changed or source != user_conf:
os.makedirs(os.path.dirname(user_conf), exist_ok=True)
with open(user_conf, "w") as f:
f.write("\n".join(out) + "\n")
except OSError as exc:
self._log(f"無法寫入 lxsession 設定：{exc}", logging.WARNING)
def _suppress_session_popups(self):
"""保險：如果 lxpolkit / light-locker 還是被拉起來，直接關掉（不影響桌面）。"""
pids = []
try:
entries = os.listdir("/proc")
except OSError:
return
for entry in entries:
if not entry.isdigit():
continue
pid = int(entry)
if pid == os.getpid():
continue
cmd = self._cmdline(pid)
if not cmd:
continue
low = cmd.lower()
if not any(name in low for name in ("lxpolkit", "light-locker")):
continue
if DESKTOP_DISPLAY in cmd or self._has_display(pid):
pids.append(pid)
if pids:
self._log(f"關閉容器內無效的 session 元件：{pids}")
self._kill_pids(pids, signal.SIGKILL)
def _kill_pids(self, pids, sig=signal.SIGTERM):
for pid in pids:
if not self._pid_alive(pid):
continue
try:
pgid = os.getpgid(pid)
if pgid > 1 and pgid != os.getpgid(0):
os.killpg(pgid, sig)
else:
os.kill(pid, sig)
except (ProcessLookupError, PermissionError, OSError):
try:
os.kill(pid, sig)
except OSError:
pass
def _stop_all(self):
xvfb, vnc, session = self._scan()
pids = set(xvfb) | set(vnc) | set(session) | set(self._scan_terms()) | set(self._scan_clip())
for value in self.pids.values():
if isinstance(value, (list, tuple, set)):
pids |= set(value)
elif value:
pids.add(value)
if pids:
self._log(f"停止桌面程序：{sorted(pids)}")
self._kill_pids(pids, signal.SIGTERM)
time.sleep(0.5)
self._kill_pids(pids, signal.SIGKILL)
self.pids = {"xvfb": None, "session": None, "vnc": None, "term": None, "clip": []}
def _clean_display_files(self):
num = DESKTOP_DISPLAY.lstrip(":") or "99"
for path in (f"/tmp/.X{num}-lock", f"/tmp/.X11-unix/X{num}"):
try:
os.remove(path)
except FileNotFoundError:
pass
except OSError as exc:
self._log(f"無法移除 {path}: {exc}", logging.WARNING)
def _display_socket(self) -> str:
num = DESKTOP_DISPLAY.lstrip(":") or "99"
return f"/tmp/.X11-unix/X{num}"
def _healthy(self) -> bool:
required = ["xvfb", "session", "vnc"]
if self._term_binary():
required.append("term")
if shutil.which("autocutsel"):
required.append("clip")
def alive(value):
if isinstance(value, (list, tuple, set)):
return bool(value) and all(self._pid_alive(p) for p in value)
return self._pid_alive(value)
return (
_desktop_port_open()
and all(alive(self.pids[k]) for k in required)
and os.path.exists(self._display_socket())
)
# ── 啟動各元件 ────────────────────────────────────────────
async def _wait_for(self, predicate, timeout: float) -> bool:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
if predicate():
return True
except Exception:
pass
await asyncio.sleep(0.2)
return False
async def _start_xvfb(self) -> bool:
self._log(f"啟動 Xvfb {DESKTOP_DISPLAY}（{DESKTOP_GEOMETRY}）")
try:
proc = self._spawn([
"Xvfb", DESKTOP_DISPLAY, "-screen", "0", DESKTOP_GEOMETRY,
"-nolisten", "tcp", "-ac", "-noreset",
])
except OSError as exc:
self.last_error = f"無法啟動 Xvfb：{exc}"
self._log(self.last_error, logging.ERROR)
return False
self.pids["xvfb"] = proc.pid
ok = await self._wait_for(
lambda: os.path.exists(self._display_socket()) and self._pid_alive(proc.pid), 8)
if not ok:
self.last_error = f"Xvfb 啟動失敗（returncode={proc.poll()}）"
self._log(self.last_error, logging.ERROR)
return False
return True
async def _start_session(self) -> bool:
self._prepare_session_config()
last_error = "找不到 startlxde / lxsession / openbox"
for launcher in ("startlxde", "lxsession", "openbox"):
path = shutil.which(launcher)
if not path:
continue
self._log(f"啟動 LXDE session：{launcher}")
try:
proc = self._spawn([path])
except OSError as exc:
last_error = f"{launcher} 啟動失敗：{exc}"
continue
await asyncio.sleep(1.2)
pid = proc.pid if self._pid_alive(proc.pid) else None
if not pid:
_x, _v, found = self._scan()
pid = found[0] if found else None
if pid:
self.pids["session"] = pid
self._log(f"LXDE session 就緒（{launcher}，pid {pid}）")
await asyncio.sleep(0.5)
await asyncio.to_thread(self._suppress_session_popups)
return True
last_error = f"{launcher} 啟動後立即結束（returncode={proc.poll()}）"
self._log(last_error, logging.WARNING)
self.last_error = last_error
return False
async def _start_vnc(self) -> bool:
self._log(f"啟動 x11vnc（127.0.0.1:{DESKTOP_VNC_PORT}）")
try:
proc = self._spawn([
"x11vnc", "-display", DESKTOP_DISPLAY,
"-rfbport", str(DESKTOP_VNC_PORT), "-localhost",
"-forever", "-shared", "-nopw", "-noxdamage",
"-repeat", "-wait", "10", "-defer", "10", "-no6",
])
except OSError as exc:
self.last_error = f"無法啟動 x11vnc：{exc}"
self._log(self.last_error, logging.ERROR)
return False
self.pids["vnc"] = proc.pid
ok = await self._wait_for(
lambda: _desktop_port_open() and self._pid_alive(proc.pid), 10)
if not ok:
self.last_error = f"x11vnc 啟動失敗（returncode={proc.poll()}）"
self._log(self.last_error, logging.ERROR)
self.pids["vnc"] = None
self._kill_pids([proc.pid], signal.SIGKILL)
return False
return True
async def _start_term(self) -> bool:
"""在桌面上維持一個「共用終端機」視窗；關掉後看門狗會自動再開一個。"""
binary = self._term_binary()
if not binary:
self.pids["term"] = None
return True
name = os.path.basename(binary)
if name == "xterm":
argv = [binary, "-u8", "-fa", "DejaVu Sans Mono", "-fs", "11",
"-geometry", "110x32+50+50",
"-xrm", "XTerm*selectToClipboard: true",
"-T", self.TERM_TITLE, "-e", "bash", "--login"]
else:
argv = [binary, "--title=" + self.TERM_TITLE, "-e", "bash --login"]
cwd = os.path.join(HOME, "app")
self._log(f"啟動共用終端機視窗（{name}）")
try:
proc = self._spawn(argv, env=self._env(), cwd=cwd if os.path.isdir(cwd) else HOME)
except OSError as exc:
self.last_error = f"無法啟動終端機視窗：{exc}"
self._log(self.last_error, logging.ERROR)
return False
await asyncio.sleep(0.8)
if self._pid_alive(proc.pid):
self.pids["term"] = proc.pid
self._log(f"共用終端機視窗就緒（pid {proc.pid}）")
return True
self.pids["term"] = None
self.last_error = f"終端機視窗啟動後立即結束（returncode={proc.poll()}）"
self._log(self.last_error, logging.WARNING)
return False
async def _start_clipboard(self) -> bool:
"""啟動 autocutsel，讓 VNC 的 CUT_BUFFER0 與 X11 的 CLIPBOARD/PRIMARY 同步。"""
binary = shutil.which("autocutsel")
if not binary:
self.pids["clip"] = []
self._log("找不到 autocutsel，X11 剪貼簿同步停用", logging.WARNING)
return True
pids = []
for selection in ("CLIPBOARD", "PRIMARY"):
try:
proc = self._spawn(
[binary, "-selection", selection, "-cutbuffer", "0"],
env=self._env(),
)
pids.append(proc.pid)
except OSError as exc:
self._log(f"autocutsel {selection} 啟動失敗：{exc}", logging.WARNING)
await asyncio.sleep(0.5)
alive = [pid for pid in pids if self._pid_alive(pid)]
self.pids["clip"] = alive
if alive:
self._log("X11 剪貼簿同步就緒（CLIPBOARD + PRIMARY ↔ CUT_BUFFER0）")
else:
self._log("autocutsel 未能啟動，X11 剪貼簿同步停用", logging.WARNING)
return True
async def _full_restart(self) -> bool:
await asyncio.to_thread(self._stop_all)
self._clean_display_files()
if not await self._start_xvfb():
return False
if not await self._start_session():
return False
if not await self._start_vnc():
return False
await self._start_clipboard()
await self._start_term()
await asyncio.to_thread(self._suppress_session_popups)
self.started_at = time.time()
if self._healthy():
self.last_error = ""
self._log("桌面已就緒（Xvfb + LXDE + x11vnc + 共用終端機）")
return self._healthy()
async def _install_binaries(self) -> bool:
now = time.time()
if now - self._last_install < 300:
return False
self._last_install = now
self._log("缺少桌面套件，嘗試安裝：" + DESKTOP_APT_PACKAGES, logging.WARNING)
cmd = ("DEBIAN_FRONTEND=noninteractive apt-get update -y && "
"DEBIAN_FRONTEND=noninteractive apt-get install -y "
"--no-install-recommends " + DESKTOP_APT_PACKAGES)
for prefix in ("", "sudo "):
try:
rc, _out, err = await _run(prefix + cmd)
except Exception as exc:
self._log(f"桌面套件安裝例外：{exc}", logging.WARNING)
continue
if rc == 0 and _desktop_binaries_present():
self._log("桌面套件安裝完成")
return True
self._log(f"桌面套件安裝失敗（prefix={prefix!r}）：{(err or '')[-300:]}", logging.WARNING)
return False
async def ensure(self) -> bool:
async with self._lock:
if self._healthy():
return True
if not _desktop_binaries_present():
ok = await self._install_binaries()
if not ok:
self.last_error = "缺少 Xvfb / x11vnc / LXDE 套件"
return False
self._log("桌面未就緒，開始檢查與修復")
xvfb, vnc, session = await asyncio.to_thread(self._scan)
if len(xvfb) > 1:
self._kill_pids(xvfb[1:], signal.SIGKILL)
xvfb = xvfb[:1]
if len(vnc) > 1:
self._kill_pids(vnc[1:], signal.SIGKILL)
vnc = vnc[:1]
if not xvfb or not self._pid_alive(xvfb[0]):
self._log("找不到 Xvfb，重建整個桌面")
return await self._full_restart()
self.pids["xvfb"] = xvfb[0]
if not os.path.exists(self._display_socket()):
self._log("X11 socket 不存在，重建整個桌面")
return await self._full_restart()
if session:
self.pids["session"] = session[0]
else:
if not await self._start_session():
return False
if vnc and not _desktop_port_open():
self._kill_pids(vnc, signal.SIGKILL)
vnc = []
if vnc:
self.pids["vnc"] = vnc[0]
else:
if not await self._start_vnc():
return False
terms = await asyncio.to_thread(self._scan_terms)
if len(terms) > 1:
self._kill_pids(terms[1:], signal.SIGKILL)
terms = terms[:1]
if terms and self._pid_alive(terms[0]):
self.pids["term"] = terms[0]
elif self._term_binary():
if not await self._start_term():
return False
else:
self.pids["term"] = None
clips = await asyncio.to_thread(self._scan_clip)
if len(clips) > 2:
self._kill_pids(clips[2:], signal.SIGKILL)
clips = clips[:2]
if len(clips) == 2 and all(self._pid_alive(p) for p in clips):
self.pids["clip"] = clips
else:
if clips:
self._kill_pids(clips, signal.SIGKILL)
await self._start_clipboard()
if not self._healthy():
self.last_error = "桌面元件存在但健康檢查未通過"
return False
self.started_at = self.started_at or time.time()
self.last_error = ""
self._log("桌面已修復並就緒")
return True
async def restart(self) -> bool:
async with self._lock:
self._log("收到重新啟動桌面要求")
return await self._full_restart()
def status(self) -> dict:
alive = self._healthy()
return {
"id": DESKTOP_ID,
"title": DESKTOP_TITLE,
"alive": alive,
"display": DESKTOP_DISPLAY,
"geometry": DESKTOP_GEOMETRY,
"vnc_port": DESKTOP_VNC_PORT,
"pids": dict(self.pids),
"started_at": round(self.started_at, 3) if self.started_at else None,
"last_error": self.last_error,
"novnc": os.path.isdir(DESKTOP_NOVNC_ROOT),
}
desktop_manager = DesktopManager()
class PlaywrightManager:
"""背景安裝 Playwright Chromium，完成後在共用桌面建立啟動圖示。"""
def __init__(self):
self._lock = asyncio.Lock()
self._logf = None
self.ready = False
self.installing = False
self.chromium_path = None
self.version = ""
self.last_error = ""
self.installed_at = 0.0
self._last_attempt = 0.0
# ── 日誌 ──────────────────────────────────────────────────
def _log(self, message: str, level: int = logging.INFO):
line = f"[playwright] {message}"
log.log(level, "%s", line)
try:
f = self._logfd()
if hasattr(f, "write"):
f.write(time.strftime("%Y-%m-%d %H:%M:%S ") + line + "\n")
f.flush()
except OSError:
pass
def _logfd(self):
if self._logf is None:
try:
self._logf = open(PW_LOG, "a", buffering=1)
except OSError:
self._logf = subprocess.DEVNULL
return self._logf
# ── 執行工具 ──────────────────────────────────────────────
@staticmethod
def _env():
env = os.environ.copy()
env["PLAYWRIGHT_BROWSERS_PATH"] = PW_BROWSERS
env["DEBIAN_FRONTEND"] = "noninteractive"
return env
async def _exec(self, args, timeout=900):
proc = await asyncio.create_subprocess_exec(
*args, env=self._env(),
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
try:
out, err = await asyncio.wait_for(proc.communicate(), timeout)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
return 124, "", f"逾時 {timeout}s"
return proc.returncode, out.decode(errors="replace"), err.decode(errors="replace")
def _probe_path(self):
code = (
"from playwright.sync_api import sync_playwright\n"
"with sync_playwright() as p:\n"
" print(p.chromium.executable_path)\n"
)
try:
proc = subprocess.run(
[sys.executable, "-c", code], capture_output=True, text=True,
timeout=120, env=self._env(),
)
except Exception as exc:
self.last_error = f"無法取得 Chromium 路徑：{exc}"
return None
lines = [ln.strip() for ln in (proc.stdout or "").splitlines() if ln.strip()]
path = lines[-1] if lines else ""
if proc.returncode != 0 or not path or not os.path.isfile(path):
return None
return path
def _launch_check(self):
code = (
"from playwright.sync_api import sync_playwright\n"
"with sync_playwright() as p:\n"
" b = p.chromium.launch(headless=True, args=['--no-sandbox','--disable-dev-shm-usage'])\n"
" pg = b.new_page()\n"
" pg.set_content('webterm-playwright-ok')\n"
" print(pg.title())\n"
" b.close()\n"
)
try:
proc = subprocess.run(
[sys.executable, "-c", code], capture_output=True, text=True,
timeout=180, env=self._env(),
)
except Exception as exc:
self.last_error = f"Chromium 啟動測試例外：{exc}"
return False
if proc.returncode != 0 or "webterm-playwright-ok" not in (proc.stdout or ""):
detail = ((proc.stderr or "") or (proc.stdout or "")).strip()
self.last_error = f"Chromium 啟動測試失敗：{detail[-500:]}"
return False
return True
def _write_launchers(self):
icon = ""
for name in ("product_logo_48.png", "product_logo_32.png"):
candidate = os.path.join(os.path.dirname(self.chromium_path), name)
if os.path.isfile(candidate):
icon = candidate
break
script = (
"#!/bin/bash\n"
f"export DISPLAY=\"${{DISPLAY:-{DESKTOP_DISPLAY}}}\"\n"
f"exec {shlex.quote(self.chromium_path)} --no-sandbox --disable-dev-shm-usage "
"--no-first-run --no-default-browser-check --start-maximized \"$@\"\n"
)
entry = (
"[Desktop Entry]\n"
"Type=Application\n"
"Version=1.0\n"
f"Name={PW_TITLE}\n"
"Comment=Chromium (Playwright) on the shared LXDE desktop\n"
f"Exec={PW_LAUNCHER} %U\n"
f"Icon={icon or 'web-browser'}\n"
"Terminal=false\n"
"Categories=Network;WebBrowser;\n"
"StartupNotify=true\n"
)
os.makedirs(os.path.dirname(PW_LAUNCHER), exist_ok=True)
with open(PW_LAUNCHER, "w") as f:
f.write(script)
os.chmod(PW_LAUNCHER, 0o755)
for path in (PW_APP_ENTRY, PW_DESKTOP_ENTRY):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
f.write(entry)
os.chmod(path, 0o755)
# ── 安裝 ──────────────────────────────────────────────────
async def _install(self) -> bool:
self.ready = False
self.last_error = ""
self._log("開始安裝 Playwright Chromium…")
rc, _, _ = await self._exec([sys.executable, "-c", "import playwright"], timeout=60)
if rc != 0:
self._log("Python 套件 playwright 不存在，先以 pip 安裝", logging.WARNING)
rc, out, err = await self._exec(
[sys.executable, "-m", "pip", "install", "--no-cache-dir", "playwright"],
timeout=600,
)
if rc != 0:
self.last_error = f"pip install playwright 失敗：{(err or out)[-500:]}"
self._log(self.last_error, logging.WARNING)
return False
rc, out, err = await self._exec(
[sys.executable, "-m", "playwright", "install", "chromium"], timeout=1800)
if rc != 0:
self._log("Chromium 下載失敗，先安裝系統依賴再重試", logging.WARNING)
await self._exec(
[sys.executable, "-m", "playwright", "install-deps", "chromium"], timeout=1800)
rc, out, err = await self._exec(
[sys.executable, "-m", "playwright", "install", "chromium"], timeout=1800)
if rc != 0:
self.last_error = f"playwright install chromium 失敗：{(err or out)[-500:]}"
self._log(self.last_error, logging.WARNING)
return False
path = await asyncio.to_thread(self._probe_path)
if not path:
self.last_error = self.last_error or "找不到 Playwright 的 Chromium 執行檔"
self._log(self.last_error, logging.WARNING)
return False
self.chromium_path = path
self.version = os.path.basename(os.path.dirname(os.path.dirname(path)))
self._log(f"Chromium 已下載：{path}")
if not await asyncio.to_thread(self._launch_check):
self._log(self.last_error + "；嘗試安裝系統依賴", logging.WARNING)
await self._exec(
[sys.executable, "-m", "playwright", "install-deps", "chromium"], timeout=1800)
if not await asyncio.to_thread(self._launch_check):
self._log(self.last_error, logging.WARNING)
return False
try:
await asyncio.to_thread(self._write_launchers)
except Exception as exc:
self.last_error = f"建立桌面啟動圖示失敗：{exc!r}"
self._log(self.last_error, logging.WARNING)
return False
self.ready = True
self.installed_at = time.time()
self.last_error = ""
self._log(f"Playwright Chromium 就緒（{self.version}），桌面圖示：{PW_DESKTOP_ENTRY}")
try:
pcman = shutil.which("pcmanfm")
if pcman and _desktop_port_open():
subprocess.Popen(
[pcman, "--reconfigure"],
env={**os.environ, "DISPLAY": DESKTOP_DISPLAY},
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
)
except Exception:
pass
return True
async def ensure(self, force: bool = False) -> bool:
async with self._lock:
if not force and self.chromium_path and os.path.isfile(self.chromium_path):
self.ready = True
if not os.path.isfile(PW_LAUNCHER):
try:
await asyncio.to_thread(self._write_launchers)
except Exception:
pass
return True
now = time.time()
if not force and now - self._last_attempt < 300:
return False
self._last_attempt = now
self.installing = True
try:
return await self._install()
except Exception as exc:
self.last_error = f"安裝例外：{exc!r}"
self._log(self.last_error, logging.WARNING)
return False
finally:
self.installing = False
def status(self) -> dict:
return {
"ready": self.ready,
"installing": self.installing,
"chromium": self.chromium_path,
"version": self.version,
"browsers_path": PW_BROWSERS,
"launcher": PW_LAUNCHER if os.path.isfile(PW_LAUNCHER) else None,
"desktop_entry": PW_DESKTOP_ENTRY if os.path.isfile(PW_DESKTOP_ENTRY) else None,
"installed_at": round(self.installed_at, 3) if self.installed_at else None,
"last_error": self.last_error,
}
playwright_manager = PlaywrightManager()
@app.get("/api/playwright/status")
async def api_playwright_status():
return playwright_manager.status()
@app.post("/api/playwright/install")
async def api_playwright_install():
global PW_TASK
if PW_TASK is None or PW_TASK.done():
PW_TASK = asyncio.create_task(playwright_manager.ensure(force=True))
return {"ok": True, "started": True, **playwright_manager.status()}
async def playwright_watchdog():
"""背景安裝 Chromium；完成後每 5 分鐘確認執行檔還在。"""
await asyncio.sleep(10)
while True:
ok = False
try:
ok = await playwright_manager.ensure()
except Exception as exc:
log.warning("[playwright] 背景安裝例外（終端機不受影響）：%s", exc)
await asyncio.sleep(300 if ok else 60)
async def desktop_watchdog():
"""每 3 秒確認桌面還在，且永遠只有一份；不影響終端機流程。"""
await asyncio.sleep(1)
while True:
ok = False
try:
ok = await desktop_manager.ensure()
except Exception as exc:
log.warning("[desktop] 看門狗例外（終端機不受影響）：%s", exc)
await asyncio.sleep(3 if ok else 5)
@spaces.GPU(duration=1, queue=False)
def dummy_gpu():
"""空函數，僅用於讓平台識別 GPU 裝飾器（若無 GPU 可忽略）。"""
return
def _ensure_shell_path():
"""把 s5p / opencode / pi 路徑寫進登入 shell 的 rc（必須在建立 tmux 前完成）。"""
s5p_dir = os.path.join(REPO_DIR, "s5p")
line = (f'export PATH="{s5p_dir}:{s5p_dir}/oshopencode:'
f'$HOME/.opencode/bin:$HOME/.local/bin:$PATH"')
for rc in (".bashrc", ".profile", ".bash_profile"):
path = os.path.join(HOME, rc)
try:
old = ""
if os.path.isfile(path):
with open(path) as f:
old = f.read()
if line not in old:
with open(path, "a") as f:
if old and not old.endswith("\n"):
f.write("\n")
f.write(line + "\n")
except OSError as exc:
log.warning("無法更新 %s：%s", path, exc)
@app.on_event("startup")
async def startup_event():
global DESKTOP_WATCHDOG, PW_WATCHDOG
_load_tabs()
_ensure_shell_path()
_write_shell_launcher()
await ensure_tmux()
_reconcile_tmux()
# 固定桌面分頁（Xvfb + LXDE + x11vnc）。獨立背景任務，失敗不影響終端機。
try:
if DESKTOP_WATCHDOG is None or DESKTOP_WATCHDOG.done():
DESKTOP_WATCHDOG = asyncio.create_task(desktop_watchdog())
except Exception as _desktop_err: # pragma: no cover
log.warning("[desktop] 無法啟動看門狗（終端機照常運作）：%r", _desktop_err)
# Playwright Chromium 背景下載；完成後桌面會出現啟動圖示。
try:
if PW_WATCHDOG is None or PW_WATCHDOG.done():
PW_WATCHDOG = asyncio.create_task(playwright_watchdog())
except Exception as _pw_err: # pragma: no cover
log.warning("[playwright] 無法啟動背景安裝（終端機照常運作）：%r", _pw_err)
dummy_gpu()
if SKIP_INSTALL:
log.info("WEBTERM_SKIP_INSTALL=1：跳過 opencode / 編輯器安裝")
return
# 安裝 OpenCode 後清除使用者層級設定，使用 OpenCode 預設值。
await install_opencode()
reset_opencode_config()
# 記憶體守衛 + OpenCode 記憶體規則（必須在 reset 之後）
install_memguard()
#await run_installpi_script()
# 預設環境 bootstrap：安裝 s5p、甲（flint Q4_1 llama-server + pi），背景執行不阻塞啟動。
if os.environ.get("WEBTERM_SKIP_BOOTSTRAP", "") != "1":
asyncio.create_task(run_bootstrap_script())
# 安裝編輯器不影響主要流程，放背景執行
asyncio.create_task(install_editors())
# ──────────────────────────── Terminal Session 管理（tmux）──────────────────────────────
# 每個分頁 = 一個 tmux session（真正共用、長存的 shell）；每個瀏覽器分頁 = 一條獨立的
# tmux client attach。不同裝置可以有不同終端機大小，由 tmux 針對每個 client 重畫，
# 所以不會再出現「手機把電腦畫面擠歪」或「重新整理後亂碼」。
# WebSocket 斷線只會 detach 該 client，tmux session（opencode）繼續跑。
TMUX_SOCKET = "webterm"
TMUX_PREFIX = "wt_"
SESSIONS: dict[str, "TerminalSession"] = {} # 每條 WebSocket 連線一個 attach pty
SESSIONS_LOCK = asyncio.Lock()
ATTACHMENTS: dict[str, set[str]] = {} # tab_id -> {session_id}
LIVE_TABS: set[str] = set() # 已知存在的 tmux session
WINDOW_SIZES: dict[str, tuple[int, int]] = {} # 最近一次附掛的 client 尺寸（僅除錯用）
# 共用分頁清單（順序即顯示順序），持久化到 TABS_FILE。
TABS: list[str] = []
TABS_LOCK = asyncio.Lock()
TAB_CREATE_LOCK = asyncio.Lock()
SHELL_LAUNCHER = os.path.join(HOME, ".webterm_shell.sh")
def _tab_title(tab_id: str) -> str:
if tab_id == PRIMARY_SESSION_ID:
return "終端機 1（共用）"
if tab_id.startswith("tab-") and tab_id[4:].isdigit():
return f"終端機 {int(tab_id[4:])}（共用）"
return f"終端機 {tab_id}（共用）"
def _load_tabs():
global TABS
ids: list[str] = []
try:
with open(TABS_FILE) as f:
data = json.load(f)
if isinstance(data, list):
ids = [str(x) for x in data if isinstance(x, str) and x]
except (OSError, ValueError):
ids = []
if PRIMARY_SESSION_ID not in ids:
ids.insert(0, PRIMARY_SESSION_ID)
# 去除重複、限制上限
seen = set()
TABS = [x for x in ids if not (x in seen or seen.add(x))][:MAX_TABS]
_save_tabs()
def _save_tabs():
try:
tmp = TABS_FILE + ".tmp"
with open(tmp, "w") as f:
json.dump(TABS, f)
os.replace(tmp, TABS_FILE)
except OSError as exc:
log.warning("無法儲存分頁清單 %s: %s", TABS_FILE, exc)
def _new_tab_id() -> str:
n = 2
while f"tab-{n}" in TABS:
n += 1
return f"tab-{n}"
def tabs_snapshot() -> list[dict]:
out = []
for tab_id in TABS:
attached = ATTACHMENTS.get(tab_id, set())
out.append({
"id": tab_id,
"title": _tab_title(tab_id),
"alive": tab_id in LIVE_TABS,
"subscribers": len(attached),
"pid": next((SESSIONS[s].pid for s in attached if s in SESSIONS), None),
"cols": WINDOW_SIZES.get(tab_id, (None, None))[0],
"rows": WINDOW_SIZES.get(tab_id, (None, None))[1],
"buffer_bytes": 0,
})
return out
# ── tmux 輔助函式 ────────────────────────────────────────────────────────────
def _tmux(*args, timeout: float = 5) -> subprocess.CompletedProcess:
return subprocess.run(
["tmux", "-L", TMUX_SOCKET, *args],
capture_output=True, text=True, timeout=timeout,
)
def _tmux_name(tab_id: str) -> str:
return TMUX_PREFIX + tab_id
def _reconcile_tmux():
"""app 重啟後，把仍然存活的 tmux session 接回分頁清單（opencode 不會被重啟）。"""
if not shutil.which("tmux"):
return
try:
r = _tmux("list-sessions", "-F", "#S")
except (OSError, subprocess.SubprocessError):
return
if r.returncode != 0: # 沒有任何 session
return
names = set((r.stdout or "").split())
for name in names:
if not name.startswith(TMUX_PREFIX):
continue
tab_id = name[len(TMUX_PREFIX):]
if tab_id in TABS:
LIVE_TABS.add(tab_id)
else:
try:
_tmux("kill-session", "-t", name)
except (OSError, subprocess.SubprocessError):
pass
def _write_shell_launcher():
"""建立 tmux 內 shell 的啟動腳本（保留 PWD 提示，讓上傳功能知道目前目錄）。"""
s5p_dir = os.path.join(REPO_DIR, "s5p")
content = (
"#!/bin/bash\n"
f"export PATH=\"{s5p_dir}:{s5p_dir}/oshopencode:$HOME/.opencode/bin:$HOME/.local/bin:$PATH\"\n"
"export PROMPT_COMMAND='echo -ne \"\\033]0;PWD:$PWD\\007\"'\n"
"exec bash --login\n"
)
try:
old = ""
if os.path.isfile(SHELL_LAUNCHER):
with open(SHELL_LAUNCHER) as f:
old = f.read()
if old != content:
with open(SHELL_LAUNCHER, "w") as f:
f.write(content)
os.chmod(SHELL_LAUNCHER, 0o755)
except OSError as exc:
log.warning("無法建立 shell launcher %s: %s", SHELL_LAUNCHER, exc)
def _tmux_apply_options(name: str):
"""每次附掛前都套用（舊版留下的 session 也會被修正）。
window-size smallest：tmux 依「所有目前附掛的 client」計算 window 大小，取最小值，
並自動扣掉每台裝置自己的狀態列。這樣任何一台 client 的可視區域都一定放得下
window，不會出現「window 比畫面大」造成的上下錯位、殘影與亂碼。
不再用 resize-window 手動指定尺寸，也不會因為某台裝置按鍵就切換整個 window 的大小。
"""
_tmux("set-option", "-w", "-t", name, "window-size", "smallest")
_tmux("set-option", "-w", "-t", name, "aggressive-resize", "off")
_tmux("set-option", "-t", name, "history-limit", "10000")
def _ensure_tmux_session(tab_id: str) -> bool:
"""確保分頁對應的 tmux session 存在（不存在就建立）。"""
if not shutil.which("tmux"):
return False
name = _tmux_name(tab_id)
working_dir = os.path.join(HOME, "app")
os.makedirs(working_dir, exist_ok=True)
try:
if _tmux("has-session", "-t", name).returncode == 0:
_tmux_apply_options(name)
LIVE_TABS.add(tab_id)
return True
r = _tmux("new-session", "-d", "-s", name, "-c", working_dir,
"-x", "120", "-y", "30", SHELL_LAUNCHER)
if r.returncode != 0:
log.warning("tmux new-session(%s) 失敗: %s", tab_id, (r.stderr or "").strip()[:300])
return False
_tmux_apply_options(name)
LIVE_TABS.add(tab_id)
return True
except (OSError, subprocess.SubprocessError) as exc:
log.warning("tmux 操作失敗（%s）: %s", tab_id, exc)
return False
def _remember_client_size(tab_id: str, cols: int, rows: int):
"""只記錄最近一次附掛的 client 尺寸（除錯用）。實際 window 尺寸由 tmux 依 client 計算。"""
WINDOW_SIZES[tab_id] = (max(2, int(cols)), max(2, int(rows)))
async def ensure_tmux() -> bool:
"""確保 tmux 可用（Space 由 packages.txt 預裝；缺少時嘗試 apt 安裝）。"""
if shutil.which("tmux"):
return True
log.warning("找不到 tmux，嘗試安裝 ...")
for prefix in ("", "sudo "):
rc, _out, err = await _run(f"{prefix}apt-get update -y && {prefix}apt-get install -y tmux")
if rc == 0 and shutil.which("tmux"):
log.info("✅ tmux 安裝成功")
return True
log.error("❌ tmux 安裝失敗，終端機分頁無法使用")
return False
class TerminalSession:
"""一條 WebSocket 連線對應的 tmux client attach（不是共用 shell 本體）。"""
def __init__(self, session_id: str, tab_id: str, pid: int, fd: int,
cols: int = 120, rows: int = 30):
self.id = session_id
self.tab_id = tab_id
self.pid = pid
self.fd = fd
self.cols = cols
self.rows = rows
self.created_at = time.time()
self.last_active = self.created_at
self.subscribers: dict[int, asyncio.Queue] = {}
self._next_sub_id = 0
# 這條 attach 在 WebSocket 訂閱者就緒前先收到的畫面（tmux 一 attach 就會重畫）。
# 只屬於這條連線、尺寸一致，回放不會造成別台裝置的亂碼。
self.buffer = bytearray()
def add_subscriber(self):
self.last_active = time.time()
sid = self._next_sub_id
self._next_sub_id += 1
q: asyncio.Queue = asyncio.Queue()
if self.buffer:
q.put_nowait(bytes(self.buffer))
self.buffer.clear()
self.subscribers[sid] = q
return sid, q
def remove_subscriber(self, sid: int, loop: asyncio.AbstractEventLoop):
self.subscribers.pop(sid, None)
self.last_active = time.time()
def feed(self, data: bytes):
if not self.subscribers:
self.buffer.extend(data)
overflow = len(self.buffer) - 300_000
if overflow > 0:
del self.buffer[:overflow]
return
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_tmux_attach(tab_id: str):
working_dir = os.path.join(HOME, "app")
os.makedirs(working_dir, exist_ok=True)
pid, fd = pty.fork()
if pid == 0:
try:
os.chdir(working_dir)
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}"
os.execvp("tmux", ["tmux", "-L", TMUX_SOCKET,
"attach-session", "-t", _tmux_name(tab_id)])
except Exception:
os._exit(127)
return pid, fd
async def create_attachment(tab_id: str, cols: int = 120, rows: int = 30) -> TerminalSession:
if not await asyncio.to_thread(_ensure_tmux_session, tab_id):
raise OSError("tmux 無法啟動（請確認已安裝 tmux）")
# 不手動調整 tmux window：附掛的 pty 尺寸就是這台裝置的可視區域，
# tmux 會依所有 client 自動計算 window 大小並重畫這台的畫面。
_remember_client_size(tab_id, cols, rows)
pid, fd = _fork_tmux_attach(tab_id)
session_id = uuid.uuid4().hex[:12]
sess = TerminalSession(session_id, tab_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:
# tmux client 結束（detach／session 被關）
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
ATTACHMENTS.setdefault(tab_id, set()).add(session_id)
return sess
async def create_shared_tab(cols: int = 120, rows: int = 30) -> str:
async with TABS_LOCK:
if len(TABS) >= MAX_TABS:
raise ValueError(f"分頁數量已達上限（{MAX_TABS}）")
tab_id = _new_tab_id()
TABS.append(tab_id)
_save_tabs()
if not await asyncio.to_thread(_ensure_tmux_session, tab_id):
async with TABS_LOCK:
if tab_id in TABS:
TABS.remove(tab_id)
_save_tabs()
raise OSError("tmux 無法啟動（請確認已安裝 tmux）")
log.info("已建立共用分頁 %s（共 %d 頁）", tab_id, len(TABS))
return tab_id
async def delete_shared_tab(tab_id: str) -> bool:
"""明確關閉分頁：從清單移除、殺掉 tmux session，所有裝置一起消失。"""
async with TABS_LOCK:
if tab_id not in TABS or len(TABS) <= 1:
return False
TABS.remove(tab_id)
_save_tabs()
LIVE_TABS.discard(tab_id)
WINDOW_SIZES.pop(tab_id, None)
if shutil.which("tmux"):
try:
await asyncio.to_thread(lambda: _tmux("kill-session", "-t", _tmux_name(tab_id)))
except (OSError, subprocess.SubprocessError):
pass
loop = asyncio.get_running_loop()
for sid in list(ATTACHMENTS.get(tab_id, ())):
await kill_session(sid, loop)
ATTACHMENTS.pop(tab_id, None)
log.info("已關閉共用分頁 %s（剩 %d 頁）", tab_id, len(TABS))
return True
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 sess is not None:
attached = ATTACHMENTS.get(sess.tab_id)
if attached is not None:
attached.discard(session_id)
if not attached:
ATTACHMENTS.pop(sess.tab_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("tmux attach %s（分頁 %s）已清除", session_id, sess.tab_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
# 共用分頁：
# * 沒有帶 session_id（新裝置／新瀏覽器）→ 接到清單第一頁
# * 帶清單內的 id → attach 到該分頁的 tmux session（不存在就建立）
# * id 已不在清單（被其他裝置關閉）→ 告知前端收掉這一頁，不再重連
# * session_id=__new__（舊前端按 +）→ 直接開一個新的共用分頁
if not requested_id:
requested_id = TABS[0] if TABS else PRIMARY_SESSION_ID
if requested_id == NEW_SESSION_ID:
try:
tab_id = await create_shared_tab(cols, rows)
except (OSError, ValueError) as e:
await websocket.send_text(json.dumps({
"type": "system",
"data": f"\r\n\x1b[31m[❌ 無法建立新終端機：{e}]\x1b[0m\r\n",
}))
await websocket.close()
return
was_live = False
else:
tab_id = requested_id
if tab_id not in TABS:
try:
await websocket.send_text(json.dumps({"type": "deleted", "id": tab_id}))
except Exception:
pass
await websocket.close()
return
was_live = tab_id in LIVE_TABS
try:
sess = await create_attachment(tab_id, cols, rows)
except OSError as e:
await websocket.send_text(json.dumps({
"type": "system",
"data": f"\r\n\x1b[31m[❌ 無法連接終端機：{e}]\x1b[0m\r\n",
}))
await websocket.close()
return
# 告知前端這次連線對應的分頁 id（前端用它接回同一個 tmux session）
await websocket.send_text(json.dumps({
"type": "session", "id": tab_id, "reattached": was_live,
"primary": tab_id == PRIMARY_SESSION_ID,
}))
if was_live:
try:
await websocket.send_text(json.dumps({
"type": "system",
"data": "\r\n\x1b[32m[✅ 已接回原工作階段]\x1b[0m\r\n",
}))
except Exception:
pass
elif requested_id != NEW_SESSION_ID:
# 分頁還在清單裡，但 tmux session 已結束（exit／app 重啟），已重建
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():
# 增量解碼：tmux 的輸出可能在多位元組字元（中文）中間切斷，逐塊 decode 會變成亂碼。
decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
while True:
chunk = await queue.get()
if chunk is None:
break
text = decoder.decode(chunk)
if not text:
continue
try:
await websocket.send_text(json.dumps({"type": "output", "data": text}))
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":
# 只轉送按鍵。window 大小由 tmux 依所有 client 計算，輸入不會改動大小。
try:
os.write(sess.fd, msg["data"].encode("utf-8"))
except OSError:
break
elif mtype == "resize":
try:
sess.cols, sess.rows = int(msg["cols"]), int(msg["rows"])
_set_winsize(sess.fd, sess.cols, sess.rows) # tmux 收到 SIGWINCH 後自行重算
_remember_client_size(tab_id, sess.cols, sess.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":
# 明確關閉分頁 = 所有裝置一起關閉（分頁清單同步移除）。
# 若只剩最後一頁則不關，只解除這條連線的訂閱。
if not await delete_shared_tab(tab_id):
try:
await websocket.send_text(json.dumps({
"type": "system",
"data": "\r\n\x1b[33m[ℹ️ 至少要保留一個共用分頁，已保留目前工作階段]\x1b[0m\r\n",
}))
except Exception:
pass
break
except (WebSocketDisconnect, RuntimeError):
pass
except Exception:
log.exception("terminal_ws 循環發生未預期錯誤")
finally:
sender.cancel()
# 只結束「這條 WebSocket 的 tmux client attach」，tmux session（opencode）
# 本身保留，其他裝置或重新整理後仍可接回同一個分頁。
sess.remove_subscriber(sub_id, loop)
await kill_session(sess.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")