前端antd，后端fastapi，实现运行系统指令，并打印运行日志

一、需求场景与技术栈价值

1. 核心需求

   • 前端通过表单提交系统命令（如 ls -l、docker ps）
   • 后端安全执行命令并实时返回日志流
   • 前端动态渲染日志，支持自动滚动与状态监控

2. 技术栈优势

   • Ant Design：提供健壮的 Table、List 组件展示日志，Modal 实现交互式命令窗口
   • FastAPI：异步处理命令执行（asyncio.subprocess），支持 WebSocket 实时推送日志流
   • 安全性：命令白名单校验 + 用户权限隔离（参考企业级设计）

二、前后端实现详解

前端（Ant Design）关键代码

import { Button, List, Input, Modal } from 'antd';
import { useWebSocket } from 'react-use-websocket';

const CommandExecutor = () => {
  const [logs, setLogs] = useState([]);
  const [command, setCommand] = useState('');
  
  // WebSocket 连接：实时接收日志
  const { sendMessage, lastMessage } = useWebSocket('ws://localhost:8000/ws/log', {
    onMessage: (e) => {
      setLogs(prev => [...prev, { id: Date.now(), text: e.data }]);
    }
  });

  // 提交命令到后端
  const executeCommand = () => {
    if (command.trim()) {
      sendMessage(JSON.stringify({ cmd: command }));
      setCommand('');
    }
  };

  return (
    <>
      <Input.Search 
        placeholder="输入系统命令（例：ls -l）" 
        enterButton="执行" 
        value={command}
        onChange={(e) => setCommand(e.target.value)}
        onSearch={executeCommand}
      />
      <List
        dataSource={logs}
        renderItem={log => <List.Item>{log.text}</List.Item>}
        style={{ maxHeight: '400px', overflowY: 'auto' }}
      />
    </>
  );
};

后端（FastAPI）关键代码

from fastapi import FastAPI, WebSocket
import asyncio

app = FastAPI()

# 命令白名单（防止非法操作）
ALLOWED_COMMANDS = ["ls", "docker ps", "git status"]

@app.websocket("/ws/log")
async def websocket_exec(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_json()
        cmd = data.get("cmd")
        
        # 校验命令安全性
        if not any(cmd.startswith(allowed) for allowed in ALLOWED_COMMANDS):
            await websocket.send_text("错误：禁止执行的命令")
            continue

        # 异步执行命令并实时返回输出
        process = await asyncio.create_subprocess_shell(
            cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

        # 流式读取日志
        while not process.stdout.at_eof:
            line = await process.stdout.readline()
            await websocket.send_text(f"STDOUT: {line.decode().strip()}")

        # 错误流处理
        err = await process.stderr.read()
        if err:
            await websocket.send_text(f"STDERR: {err.decode()}")

三、安全与性能优化

1. 命令安全层

   • 白名单机制：仅允许预定义命令（如 ALLOWED_COMMANDS）
   • 参数过滤：使用 shlex.split() 避免命令注入攻击
   • 权限控制：JWT 鉴权绑定用户与可执行命令范围

2. 性能提升方向

   • 日志分块传输：大日志拆分为多消息包，避免 WebSocket 阻塞
   • 前端虚拟滚动：Ant Design 的 List 开启 virtual 属性支持万级日志渲染
   • 后端异步队列：Celery 后台执行长耗时命令，通过 WebSocket 推送进度

四、扩展功能实践

场景：文件上传后自动解压并日志监控

# FastAPI 文件上传接口（结合命令执行）
@app.post("/upload-and-extract")
async def upload_file(file: UploadFile):
    save_path = f"./uploads/{file.filename}"
    with open(save_path, "wb") as f:
        f.write(await file.read())
    
    # 执行解压命令并记录日志
    proc = await asyncio.create_subprocess_shell(
        f"tar -xzvf {save_path}",
        stdout=asyncio.subprocess.PIPE
    )
    return {"status": "success", "task_id": proc.pid}

前端通过 轮询接口 或 WebSocket 获取解压日志流，使用 Ant Design 的 Progress 组件展示进度。