| import os, pty, subprocess, select, requests, eventlet |
| eventlet.monkey_patch() |
| from flask import Flask, render_template_string |
| from flask_socketio import SocketIO |
|
|
| app = Flask(__name__) |
| socketio = SocketIO(app, cors_allowed_origins="*", async_mode='eventlet') |
| fd = None |
|
|
| |
| html_code = """ |
| <!DOCTYPE html> |
| <html> |
| <head> |
| <link rel="stylesheet" href="https://cdn.jsdelivr.net" /> |
| <script src="https://cdn.jsdelivr.net"></script> |
| <script src="https://cdnjs.cloudflare.com"></script> |
| </head> |
| <body style="background: #000; margin: 0;"><div id="terminal" style="height: 100vh;"></div> |
| <script> |
| const term = new Terminal({ cursorBlink: true, theme: { background: '#1e1e1e' } }); |
| const socket = io({ transports: ['websocket', 'polling'] }); |
| term.open(document.getElementById('terminal')); |
| term.onData(data => socket.emit("pty-input", { input: data })); |
| socket.on("pty-output", data => term.write(data.output)); |
| </script> |
| </body></html>""" |
|
|
| def suggest_fix(error_msg): |
| api_key = os.getenv("NVIDIA_API_KEY") |
| if not api_key: return |
| try: |
| r = requests.post("https://integrate.api.nvidia.com", |
| headers={"Authorization": f"Bearer {api_key}"}, |
| json={"model": "moonshotai/kimi-k2.5", "messages": [{"role": "user", "content": f"Ubuntu error: '{error_msg}'. Give a 1-line fix."}]}) |
| fix = r.json()['choices'][0]['message']['content'] |
| socketio.emit("pty-output", {"output": f"\\r\\n\\033[1;33mAI Fix: {fix}\\033[0m\\r\\n"}) |
| except: pass |
|
|
| def read_pty(): |
| global fd |
| while True: |
| socketio.sleep(0.01) |
| if fd and select.select([fd], [], [], 0)[0]: |
| out = os.read(fd, 1024).decode(errors='ignore') |
| socketio.emit("pty-output", {"output": out}) |
| if "not found" in out.lower(): socketio.start_background_task(suggest_fix, out) |
|
|
| @socketio.on("pty-input") |
| def on_input(data): |
| if fd: os.write(fd, data["input"].encode()) |
|
|
| @app.route("/") |
| def index(): return render_template_string(html_code) |
|
|
| if __name__ == "__main__": |
| child_pid, fd = pty.fork() |
| if child_pid == 0: |
| os.environ["TERM"] = "xterm-256color" |
| subprocess.run(["bash"]) |
| else: |
| socketio.start_background_task(read_pty) |
| socketio.run(app, host="0.0.0.0", port=7860, log_output=True) |
|
|
