jam / app.py
qa1145's picture
Update app.py
3d3bcc9 verified
Raw
History Blame Contribute Delete
23.9 kB
#!/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 = """
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Web Terminal</title>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: 'Courier New', monospace;
background: #1a1a1a;
color: #00ff00;
height: 100vh;
display: flex;
flex-direction: column;
}
.header {
background: #2d2d2d;
padding: 10px 20px;
display: flex;
justify-content: space-between;
align-items: center;
border-bottom: 1px solid #444;
}
.header h1 {
font-size: 18px;
color: #fff;
}
.header .status {
color: #00ff00;
font-size: 14px;
}
.tabs {
display: flex;
background: #2d2d2d;
border-bottom: 1px solid #444;
}
.tab {
padding: 10px 20px;
cursor: pointer;
color: #aaa;
border-bottom: 2px solid transparent;
transition: all 0.3s;
}
.tab:hover {
color: #fff;
}
.tab.active {
color: #00ff00;
border-bottom-color: #00ff00;
}
.content {
flex: 1;
display: flex;
flex-direction: column;
overflow: hidden;
}
.tab-content {
display: none;
flex: 1;
flex-direction: column;
padding: 20px;
overflow: hidden;
}
.tab-content.active {
display: flex;
}
.terminal {
flex: 1;
background: #000;
border: 1px solid #333;
border-radius: 5px;
padding: 15px;
overflow-y: auto;
font-size: 14px;
line-height: 1.5;
white-space: pre-wrap;
word-wrap: break-word;
}
.terminal .prompt {
color: #00ff00;
}
.terminal .output {
color: #ccc;
}
.terminal .error {
color: #ff0000;
}
.input-area {
display: flex;
margin-top: 10px;
gap: 10px;
}
.input-area input {
flex: 1;
background: #000;
border: 1px solid #333;
color: #00ff00;
padding: 10px;
font-family: 'Courier New', monospace;
font-size: 14px;
border-radius: 5px;
}
.input-area input:focus {
outline: none;
border-color: #00ff00;
}
.btn {
background: #00ff00;
color: #000;
border: none;
padding: 10px 20px;
cursor: pointer;
font-family: 'Courier New', monospace;
font-size: 14px;
border-radius: 5px;
transition: all 0.3s;
}
.btn:hover {
background: #00cc00;
}
.btn-danger {
background: #ff4444;
}
.btn-danger:hover {
background: #cc0000;
}
.btn-secondary {
background: #444;
color: #fff;
}
.btn-secondary:hover {
background: #555;
}
.info-grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
gap: 10px;
margin-bottom: 20px;
}
.info-card {
background: #2d2d2d;
padding: 15px;
border-radius: 5px;
text-align: center;
}
.info-card .label {
color: #aaa;
font-size: 12px;
}
.info-card .value {
color: #00ff00;
font-size: 20px;
margin-top: 5px;
}
.textarea {
background: #000;
border: 1px solid #333;
color: #00ff00;
padding: 10px;
font-family: 'Courier New', monospace;
font-size: 14px;
border-radius: 5px;
width: 100%;
min-height: 100px;
resize: vertical;
}
.bg-list {
background: #000;
border: 1px solid #333;
border-radius: 5px;
padding: 15px;
max-height: 300px;
overflow-y: auto;
}
.bg-item {
display: flex;
justify-content: space-between;
align-items: center;
padding: 10px;
border-bottom: 1px solid #333;
}
.bg-item:last-child {
border-bottom: none;
}
.bg-item .pid {
color: #00ff00;
}
.bg-item .cmd {
color: #ccc;
flex: 1;
margin: 0 10px;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.log {
background: #000;
border: 1px solid #333;
border-radius: 5px;
padding: 15px;
height: 300px;
overflow-y: auto;
font-size: 12px;
color: #ccc;
white-space: pre-wrap;
word-wrap: break-word;
}
.toast {
position: fixed;
top: 20px;
right: 20px;
background: #333;
color: #fff;
padding: 15px 20px;
border-radius: 5px;
display: none;
z-index: 1000;
animation: slideIn 0.3s;
}
@keyframes slideIn {
from { transform: translateX(100%); }
to { transform: translateX(0); }
}
.loading {
display: inline-block;
width: 20px;
height: 20px;
border: 3px solid #f3f3f3;
border-top: 3px solid #00ff00;
border-radius: 50%;
animation: spin 1s linear infinite;
}
@keyframes spin {
0% { transform: rotate(0deg); }
100% { transform: rotate(360deg); }
}
</style>
</head>
<body>
<div class="header">
<h1>🖥️ Web Terminal</h1>
<span class="status">● 在线</span>
</div>
<div class="tabs">
<div class="tab active" data-tab="terminal">终端</div>
<div class="tab" data-tab="background">后台进程</div>
<div class="tab" data-tab="system">系统信息</div>
<div class="tab" data-tab="zeabur">Zeabur</div>
</div>
<div class="content">
<!-- 终端标签页 -->
<div class="tab-content active" id="tab-terminal">
<div class="terminal" id="terminal-output">
<span class="prompt">$ </span>欢迎使用 Web Terminal<br>
<span class="prompt">$ </span>输入命令并按回车执行<br>
<span class="prompt">$ </span>使用 "help" 查看可用命令<br><br>
</div>
<div class="input-area">
<input type="text" id="command-input" placeholder="输入命令..." autocomplete="off">
<button class="btn" onclick="executeCommand()">执行</button>
<button class="btn btn-secondary" onclick="clearTerminal()">清空</button>
</div>
</div>
<!-- 后台进程标签页 -->
<div class="tab-content" id="tab-background">
<div class="info-grid">
<div class="info-card">
<div class="label">运行中进程</div>
<div class="value" id="process-count">0</div>
</div>
<div class="info-card">
<div class="label">系统负载</div>
<div class="value" id="system-load">-</div>
</div>
</div>
<div class="input-area">
<input type="text" id="bg-command" placeholder="输入后台命令..." autocomplete="off" class="textarea">
<button class="btn" onclick="startBackground()">启动</button>
<button class="btn btn-danger" onclick="stopAllBackground()">停止全部</button>
</div>
<div class="bg-list" id="bg-list">
<div style="color: #aaa; text-align: center; padding: 20px;">暂无后台进程</div>
</div>
</div>
<!-- 系统信息标签页 -->
<div class="tab-content" id="tab-system">
<div class="info-grid">
<div class="info-card">
<div class="label">主机名</div>
<div class="value" id="hostname">-</div>
</div>
<div class="info-card">
<div class="label">用户</div>
<div class="value" id="username">-</div>
</div>
<div class="info-card">
<div class="label">当前目录</div>
<div class="value" id="cwd">-</div>
</div>
<div class="info-card">
<div class="label">内存</div>
<div class="value" id="memory">-</div>
</div>
</div>
<button class="btn" onclick="refreshSystemInfo()">刷新系统信息</button>
<div class="log" id="system-log">点击按钮获取系统信息...</div>
</div>
<!-- Zeabur 标签页 -->
<div class="tab-content" id="tab-zeabur">
<div class="input-area">
<input type="password" id="zeabur-token" placeholder="输入 Zeabur Token..." autocomplete="off">
<button class="btn" onclick="installZeabur()">安装 Zeabur</button>
</div>
<div class="log" id="zeabur-log">输入 Token 并点击安装...</div>
</div>
</div>
<div class="toast" id="toast"></div>
<script>
// Tab 切换
document.querySelectorAll('.tab').forEach(tab => {
tab.addEventListener('click', () => {
document.querySelectorAll('.tab').forEach(t => t.classList.remove('active'));
document.querySelectorAll('.tab-content').forEach(c => c.classList.remove('active'));
tab.classList.add('active');
document.getElementById('tab-' + tab.dataset.tab).classList.add('active');
});
});
// 终端功能
function executeCommand() {
const input = document.getElementById('command-input');
const command = input.value.trim();
if (!command) return;
const terminal = document.getElementById('terminal-output');
terminal.innerHTML += `<span class="prompt">$ </span>${escapeHtml(command)}<br>`;
terminal.innerHTML += `<span class="loading"></span>`;
input.value = '';
fetch('/api/execute', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({command: command})
})
.then(response => response.json())
.then(data => {
terminal.innerHTML = terminal.innerHTML.replace('<span class="loading"></span>', '');
const outputClass = data.success ? 'output' : 'error';
terminal.innerHTML += `<span class="${outputClass}">${escapeHtml(data.output)}</span><br><br>`;
terminal.scrollTop = terminal.scrollHeight;
})
.catch(error => {
terminal.innerHTML = terminal.innerHTML.replace('<span class="loading"></span>', '');
terminal.innerHTML += `<span class="error">执行错误: ${escapeHtml(error.message)}</span><br><br>`;
});
}
function clearTerminal() {
document.getElementById('terminal-output').innerHTML = '<span class="prompt">$ </span>终端已清空<br>';
}
// 回车执行
document.getElementById('command-input').addEventListener('keypress', (e) => {
if (e.key === 'Enter') executeCommand();
});
// 后台进程
function startBackground() {
const command = document.getElementById('bg-command').value.trim();
if (!command) return;
fetch('/api/background/start', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({command: command})
})
.then(response => response.json())
.then(data => {
showToast(data.output);
document.getElementById('bg-command').value = '';
refreshBackgroundList();
});
}
function stopAllBackground() {
fetch('/api/background/stop', {method: 'POST'})
.then(response => response.json())
.then(data => {
showToast(data.output);
refreshBackgroundList();
});
}
function stopBackground(pid) {
fetch('/api/background/stop', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({pid: pid})
})
.then(response => response.json())
.then(data => {
showToast(data.output);
refreshBackgroundList();
});
}
function refreshBackgroundList() {
fetch('/api/background/list')
.then(response => response.json())
.then(data => {
const list = document.getElementById('bg-list');
document.getElementById('process-count').textContent = data.processes.length;
if (data.processes.length === 0) {
list.innerHTML = '<div style="color: #aaa; text-align: center; padding: 20px;">暂无后台进程</div>';
return;
}
list.innerHTML = data.processes.map(p => `
<div class="bg-item">
<span class="pid">PID: ${p.pid}</span>
<span class="cmd">${escapeHtml(p.command)}</span>
<button class="btn btn-danger" onclick="stopBackground(${p.pid})">停止</button>
</div>
`).join('');
});
}
// 系统信息
function refreshSystemInfo() {
fetch('/api/system/info')
.then(response => response.json())
.then(data => {
document.getElementById('hostname').textContent = data.hostname;
document.getElementById('username').textContent = data.username;
document.getElementById('cwd').textContent = data.cwd;
document.getElementById('memory').textContent = data.memory;
document.getElementById('system-log').textContent = data.output;
});
}
// Zeabur 安装
function installZeabur() {
const token = document.getElementById('zeabur-token').value.trim();
if (!token) {
showToast('请输入 Zeabur Token');
return;
}
const log = document.getElementById('zeabur-log');
log.innerHTML = '正在安装 Zeabur Mesh...<br>';
fetch('/api/zeabur/install', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({token: token})
})
.then(response => response.json())
.then(data => {
log.innerHTML = data.output.replace(/\\n/g, '<br>');
});
}
// 工具函数
function escapeHtml(text) {
const div = document.createElement('div');
div.textContent = text;
return div.innerHTML;
}
function showToast(message) {
const toast = document.getElementById('toast');
toast.textContent = message;
toast.style.display = 'block';
setTimeout(() => { toast.style.display = 'none'; }, 3000);
}
// 初始化
refreshBackgroundList();
setInterval(refreshBackgroundList, 5000);
</script>
</body>
</html>
"""
# 路由
@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)