Clawdbot管理平台开发：Vue3前端与Python后端联调指南-开发者社区

Clawdbot管理平台开发：Vue3前端与Python后端联调指南

1. 引言

在现代Web应用开发中，前后端分离架构已成为主流模式。本文将带您从零开始，使用Vue3和Python构建Clawdbot的Web管理界面，涵盖从基础环境搭建到高级功能实现的完整流程。

通过本教程，您将掌握：

Vue3前端与Python后端的通信机制
如何封装axios实现高效API调用
JWT鉴权在前端的安全实现
WebSocket实时日志推送技术

2. 环境准备与项目初始化

2.1 前端环境搭建

首先确保已安装Node.js（建议版本16+），然后创建Vue3项目：

npm init vue@latest clawdbot-admin cd clawdbot-admin npm install

安装必要依赖：

npm install axios vue-router pinia socket.io-client

2.2 后端环境准备

Python环境（建议3.8+）和FastAPI框架安装：

pip install fastapi uvicorn python-jose[cryptography] passlib python-multipart

3. 前端核心架构实现

3.1 axios封装与API管理

在src/utils目录下创建api.js：

import axios from 'axios' const apiClient = axios.create({ baseURL: 'http://localhost:8000/api', timeout: 10000, headers: { 'Content-Type': 'application/json' } }) // 请求拦截器 apiClient.interceptors.request.use(config => { const token = localStorage.getItem('access_token') if (token) { config.headers.Authorization = `Bearer ${token}` } return config }) // 响应拦截器 apiClient.interceptors.response.use( response => response.data, error => { if (error.response?.status === 401) { // 处理token过期 router.push('/login') } return Promise.reject(error) } ) export default apiClient

3.2 JWT鉴权实现

创建认证Store（使用Pinia）：

// stores/auth.js import { defineStore } from 'pinia' import api from '@/utils/api' export const useAuthStore = defineStore('auth', { state: () => ({ user: null, isAuthenticated: false }), actions: { async login(credentials) { try { const response = await api.post('/auth/login', credentials) localStorage.setItem('access_token', response.access_token) this.isAuthenticated = true this.user = response.user return true } catch (error) { console.error('Login failed:', error) return false } }, logout() { localStorage.removeItem('access_token') this.isAuthenticated = false this.user = null } } })

4. 后端接口开发

4.1 FastAPI基础配置

# main.py from fastapi import FastAPI, Depends, HTTPException from fastapi.security import OAuth2PasswordBearer app = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login") @app.get("/api/protected") async def protected_route(token: str = Depends(oauth2_scheme)): return {"message": "This is a protected route"}

4.2 JWT认证实现

# auth.py from datetime import datetime, timedelta from jose import JWTError, jwt from passlib.context import CryptContext SECRET_KEY = "your-secret-key" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def create_access_token(data: dict): to_encode = data.copy() expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) def verify_token(token: str): try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) return payload except JWTError: return None

5. WebSocket实时日志推送

5.1 后端WebSocket实现

# websocket.py from fastapi import WebSocket class ConnectionManager: def __init__(self): self.active_connections = [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: str): for connection in self.active_connections: await connection.send_text(message) manager = ConnectionManager() @app.websocket("/ws/logs") async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: data = await websocket.receive_text() # 处理接收到的消息 except WebSocketDisconnect: manager.disconnect(websocket)

5.2 前端WebSocket连接

// src/utils/socket.js import { io } from 'socket.io-client' const socket = io('ws://localhost:8000', { path: '/ws/logs', transports: ['websocket'] }) socket.on('connect', () => { console.log('WebSocket connected') }) socket.on('log', (data) => { // 处理日志数据 console.log('New log:', data) }) export default socket

6. 前后端联调实战

6.1 跨域问题解决

在后端添加CORS中间件：

from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:8080"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )

6.2 接口测试示例

前端调用示例：

// 获取任务列表 async function fetchTasks() { try { const response = await api.get('/tasks') console.log('Tasks:', response) } catch (error) { console.error('Failed to fetch tasks:', error) } }

后端对应接口：

@app.get("/api/tasks") async def get_tasks(token: str = Depends(oauth2_scheme)): # 验证token payload = verify_token(token) if not payload: raise HTTPException(status_code=401, detail="Invalid token") # 返回模拟数据 return [ {"id": 1, "name": "数据采集任务1", "status": "running"}, {"id": 2, "name": "数据分析任务2", "status": "completed"} ]

7. 项目部署与优化

7.1 前端构建

npm run build

7.2 后端生产环境启动

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

7.3 Nginx配置示例

server { listen 80; server_name yourdomain.com; location / { root /path/to/clawdbot-admin/dist; try_files $uri $uri/ /index.html; } location /api { proxy_pass http://localhost:8000; proxy_set_header Host $host; } location /ws/ { proxy_pass http://localhost:8000; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; } }