Create app.py

app.py

@@ -0,0 +1,233 @@
+import os
+import pty
+import signal
+import struct
+import fcntl
+import termios
+import asyncio
+import json
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from fastapi.responses import HTMLResponse
+import uvicorn
+
+app = FastAPI()
+
+# ──────────────────────────── Frontend ────────────────────────────
+HTML_PAGE = r"""<!DOCTYPE html>
+<html lang="en">
+<head>
+<meta charset="utf-8"/>
+<meta name="viewport" content="width=device-width,initial-scale=1"/>
+<title>Web Terminal</title>
+<link rel="stylesheet"
+      href="https://cdn.jsdelivr.net/npm/@xterm/xterm@5.5.0/css/xterm.min.css"/>
+<style>
+*{margin:0;padding:0;box-sizing:border-box}
+html,body{height:100%;overflow:hidden;background:#1e1e1e;color:#ccc;
+          font-family:'Segoe UI',system-ui,sans-serif}
+#tab-bar{display:flex;align-items:flex-end;background:#252526;
+         padding:4px 6px 0;gap:2px;border-bottom:1px solid #3c3c3c;
+         min-height:38px;flex-shrink:0;overflow-x:auto}
+.tab{background:#2d2d2d;color:#969696;padding:6px 14px 5px;
+     border-radius:6px 6px 0 0;cursor:pointer;font-size:13px;
+     display:flex;align-items:center;gap:8px;user-select:none;
+     white-space:nowrap;border:1px solid transparent;border-bottom:none}
+.tab:hover{color:#ddd}
+.tab.active{background:#1e1e1e;color:#fff;border-color:#3c3c3c}
+.tab .x{color:#666;font-size:15px;line-height:1;padding:0 2px}
+.tab .x:hover{color:#f55}
+#add-btn{background:none;border:1px dashed #555;color:#777;
+         padding:3px 12px;border-radius:4px;cursor:pointer;
+         font-size:18px;margin-bottom:4px;margin-left:4px}
+#add-btn:hover{color:#fff;border-color:#aaa}
+#term-box{flex:1;overflow:hidden;padding:2px}
+.tw{width:100%;height:100%;display:none}
+.tw.active{display:block}
+body{display:flex;flex-direction:column}
+</style>
+</head>
+<body>
+<div id="tab-bar"><button id="add-btn" title="New Terminal (Ctrl+Shift+T)">+</button></div>
+<div id="term-box"></div>
+
+<script src="https://cdn.jsdelivr.net/npm/@xterm/xterm@5.5.0/lib/xterm.min.js"></script>
+<script src="https://cdn.jsdelivr.net/npm/@xterm/addon-fit@0.10.0/lib/addon-fit.min.js"></script>
+<script src="https://cdn.jsdelivr.net/npm/@xterm/addon-web-links@0.11.0/lib/addon-web-links.min.js"></script>
+<script>
+const bar=document.getElementById('tab-bar'),
+      box=document.getElementById('term-box'),
+      addBtn=document.getElementById('add-btn');
+let tabs=[],activeId=null,cnt=0;
+
+function wsUrl(){
+  const p=location.protocol==='https:'?'wss:':'ws:';
+  return `${p}//${location.host}/ws`;
+}
+
+function createTab(){
+  const id=++cnt;
+  /* tab button */
+  const el=document.createElement('div');
+  el.className='tab';el.dataset.id=id;
+  el.innerHTML=`<span>Term ${id}</span><span class="x">&times;</span>`;
+  bar.insertBefore(el,addBtn);
+  /* wrapper */
+  const wrap=document.createElement('div');
+  wrap.className='tw';wrap.dataset.id=id;
+  box.appendChild(wrap);
+  /* xterm */
+  const term=new Terminal({
+    cursorBlink:true,fontSize:14,
+    fontFamily:"'Cascadia Code','Fira Code',Consolas,monospace",
+    allowProposedApi:true,
+    theme:{background:'#1e1e1e',foreground:'#d4d4d4',cursor:'#aeafad',
+           selectionBackground:'#264f78'}
+  });
+  const fit=new FitAddon.FitAddon();
+  term.loadAddon(fit);
+  term.loadAddon(new WebLinksAddon.WebLinksAddon());
+  term.open(wrap);
+  fit.fit();
+  /* websocket */
+  const ws=new WebSocket(wsUrl());
+  ws.onopen=()=>ws.send(JSON.stringify({type:'resize',cols:term.cols,rows:term.rows}));
+  ws.onmessage=e=>{
+    try{const m=JSON.parse(e.data);if(m.type==='output')term.write(m.data);}
+    catch{term.write(e.data);}
+  };
+  ws.onclose=()=>term.write('\r\n\x1b[31m[Connection closed]\x1b[0m\r\n');
+  term.onData(d=>{if(ws.readyState===1)ws.send(JSON.stringify({type:'input',data:d}))});
+  term.onResize(({cols,rows})=>{
+    if(ws.readyState===1)ws.send(JSON.stringify({type:'resize',cols,rows}));
+  });
+  /* events */
+  el.onclick=e=>{
+    if(e.target.classList.contains('x'))closeTab(id);
+    else activate(id);
+  };
+  const obj={id,el,wrap,term,fit,ws};
+  tabs.push(obj);
+  activate(id);
+  return obj;
+}
+
+function activate(id){
+  activeId=id;
+  tabs.forEach(t=>{
+    const on=t.id===id;
+    t.el.classList.toggle('active',on);
+    t.wrap.classList.toggle('active',on);
+    if(on)setTimeout(()=>{t.fit.fit();t.term.focus()},30);
+  });
+}
+
+function closeTab(id){
+  const i=tabs.findIndex(t=>t.id===id);
+  if(i<0)return;
+  const t=tabs[i];
+  t.ws.close();t.term.dispose();t.el.remove();t.wrap.remove();
+  tabs.splice(i,1);
+  if(!tabs.length)return createTab();
+  if(activeId===id)activate(tabs[Math.min(i,tabs.length-1)].id);
+}
+
+addBtn.onclick=createTab;
+window.addEventListener('resize',()=>tabs.forEach(t=>{
+  if(t.wrap.classList.contains('active'))t.fit.fit();
+}));
+document.addEventListener('keydown',e=>{
+  if(e.ctrlKey&&e.shiftKey&&e.key==='T'){e.preventDefault();createTab();}
+  if(e.ctrlKey&&e.shiftKey&&e.key==='W'){e.preventDefault();if(activeId)closeTab(activeId);}
+});
+createTab();
+</script>
+</body>
+</html>"""
+
+# ──────────────────────────── Routes ──────────────────────────────
+
+@app.get("/")
+async def index():
+    return HTMLResponse(HTML_PAGE)
+
+
+@app.websocket("/ws")
+async def terminal_ws(websocket: WebSocket):
+    await websocket.accept()
+
+    # ── fork a bash process inside a pty ──
+    pid, fd = pty.fork()
+    if pid == 0:                       # child
+        os.environ["TERM"] = "xterm-256color"
+        os.environ["COLORTERM"] = "truecolor"
+        os.execvp("bash", ["bash", "--login"])
+
+    # ── parent ──
+    loop = asyncio.get_running_loop()
+
+    def set_winsize(cols: int, rows: int):
+        fcntl.ioctl(fd, termios.TIOCSWINSZ,
+                    struct.pack("HHHH", rows, cols, 0, 0))
+
+    set_winsize(120, 30)
+
+    # non-blocking master fd
+    fl = fcntl.fcntl(fd, fcntl.F_GETFL)
+    fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
+
+    queue: asyncio.Queue = asyncio.Queue()
+
+    def _on_read():
+        try:
+            data = os.read(fd, 65536)
+            queue.put_nowait(data if data else None)
+        except (BlockingIOError, OSError):
+            queue.put_nowait(None)
+
+    loop.add_reader(fd, _on_read)
+
+    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()
+            msg = json.loads(raw)
+            if msg["type"] == "input":
+                os.write(fd, msg["data"].encode("utf-8"))
+            elif msg["type"] == "resize":
+                set_winsize(msg["cols"], msg["rows"])
+    except (WebSocketDisconnect, Exception):
+        pass
+    finally:
+        loop.remove_reader(fd)
+        sender.cancel()
+        for sig in (signal.SIGTERM, signal.SIGKILL):
+            try:
+                os.kill(pid, sig)
+                os.waitpid(pid, 0)
+                break
+            except Exception:
+                pass
+        try:
+            os.close(fd)
+        except Exception:
+            pass
+
+
+# ──────────────────────────── Entry ───────────────────────────────
+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")