#!/usr/bin/env python3 """ Hugging Face Spaces Docker 网页终端 监听 7860 端口 """ import os import subprocess import threading import time import signal import sys import json from flask import Flask, render_template_string, request, jsonify from typing import Optional app = Flask(__name__) class WebTerminal: def __init__(self): self.processes = {} self.lock = threading.Lock() def execute_command(self, command: str, timeout: int = 30) -> dict: """执行命令并返回输出""" if not command.strip(): return {"success": False, "output": "请输入命令"} try: # 使用 subprocess 执行命令 result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=timeout, cwd="/workspace" if os.path.exists("/workspace") else "/" ) # 组合输出 output = "" if result.stdout: output += result.stdout if result.stderr: output += f"\n[STDERR]\n{result.stderr}" if result.returncode != 0: output += f"\n[退出码: {result.returncode}]" return { "success": result.returncode == 0, "output": output if output else "(无输出)" } except subprocess.TimeoutExpired: return {"success": False, "output": f"命令执行超时（{timeout}秒）"} except Exception as e: return {"success": False, "output": f"执行错误: {str(e)}"} def start_background_process(self, command: str) -> dict: """启动后台进程""" try: process = subprocess.Popen( command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd="/workspace" if os.path.exists("/workspace") else "/" ) process_id = str(process.pid) self.processes[process_id] = process return { "success": True, "output": f"✅ 后台进程已启动 (PID: {process.pid})", "pid": process.pid } except Exception as e: return {"success": False, "output": f"❌ 启动失败: {str(e)}"} def stop_background_process(self, pid: Optional[int] = None) -> dict: """停止后台进程""" if pid: try: process = self.processes.get(str(pid)) if process and process.poll() is None: process.terminate() del self.processes[str(pid)] return {"success": True, "output": f"✅ 进程 {pid} 已停止"} return {"success": False, "output": f"进程 {pid} 不存在或已停止"} except Exception as e: return {"success": False, "output": f"停止失败: {str(e)}"} # 停止所有进程 stopped = [] for pid, process in list(self.processes.items()): if process.poll() is None: process.terminate() stopped.append(pid) del self.processes[pid] return {"success": True, "output": f"✅ 已停止 {len(stopped)} 个进程" if stopped else "没有运行中的进程"} def get_system_info(self) -> str: """获取系统信息""" commands = [ ("系统信息", "uname -a"), ("用户", "whoami"), ("当前目录", "pwd"), ("磁盘空间", "df -h"), ("内存", "free -h"), ("运行进程", "ps aux | head -20"), ("网络端口", "ss -tlnp"), ("环境变量", "env | head -20") ] results = [] for title, cmd in commands: result = self.execute_command(cmd, timeout=10) results.append(f"📌 {title}\n$ {cmd}\n{result['output']}\n{'='*50}") return "\n".join(results) # 创建终端实例 terminal = WebTerminal() # HTML 模板 HTML_TEMPLATE = """ Web Terminal

🖥️ Web Terminal

● 在线
终端
后台进程
系统信息
Zeabur
$ 欢迎使用 Web Terminal
$ 输入命令并按回车执行
$ 使用 "help" 查看可用命令

运行中进程
0
系统负载
-
暂无后台进程
主机名
-
用户
-
当前目录
-
内存
-
点击按钮获取系统信息...
输入 Token 并点击安装...
""" # 路由 @app.route('/') def index(): return render_template_string(HTML_TEMPLATE) @app.route('/api/execute', methods=['POST']) def api_execute(): data = request.json command = data.get('command', '') timeout = data.get('timeout', 30) result = terminal.execute_command(command, timeout) return jsonify(result) @app.route('/api/background/start', methods=['POST']) def api_background_start(): data = request.json command = data.get('command', '') result = terminal.start_background_process(command) return jsonify(result) @app.route('/api/background/stop', methods=['POST']) def api_background_stop(): data = request.json or {} pid = data.get('pid') result = terminal.stop_background_process(pid) return jsonify(result) @app.route('/api/background/list', methods=['GET']) def api_background_list(): processes = [] for pid, process in list(terminal.processes.items()): if process.poll() is None: processes.append({ 'pid': pid, 'command': process.args if hasattr(process, 'args') else 'unknown' }) return jsonify({'processes': processes}) @app.route('/api/system/info', methods=['GET']) def api_system_info(): import platform import socket info = { 'hostname': socket.gethostname(), 'username': os.getenv('USER', 'unknown'), 'cwd': os.getcwd(), 'memory': subprocess.run(['free', '-h'], capture_output=True, text=True).stdout.split('\\n')[1].split()[1] if os.path.exists('/usr/bin/free') else 'N/A', 'output': terminal.get_system_info() } return jsonify(info) @app.route('/api/zeabur/install', methods=['POST']) def api_zeabur_install(): data = request.json token = data.get('token', '') if not token: return jsonify({'success': False, 'output': '请输入 Token'}) command = f"curl -fsSL 'https://api.zeabur.com/mesh-server/install.sh?token={token}' | sudo bash" result = terminal.execute_command(command, timeout=120) return jsonify(result) # 健康检查 @app.route('/health') def health(): return jsonify({'status': 'ok', 'port': 7860}) # 启动 if __name__ == '__main__': print(f"🚀 Web Terminal 启动中...") print(f" 端口: 7860") print(f" 地址: http://0.0.0.0:7860") # 清理退出 def cleanup(signum, frame): print("\n🛑 正在关闭...") terminal.stop_background_process() sys.exit(0) signal.signal(signal.SIGINT, cleanup) signal.signal(signal.SIGTERM, cleanup) app.run(host='0.0.0.0', port=7860, debug=False)