前后端分离重构

[DoctorReid]

ZenlessZoneZero-OneDragon C/S架构重构设计文档

1. 项目概述

1.1 背景

当前项目使用PySide6作为GUI前端，业务逻辑与UI紧密耦合。为了提高系统的可扩展性、可维护性，并支持多种客户端形态，需要重构为Client/Server架构。

1.2 目标

• 将业务逻辑与UI完全分离，构建独立的后端服务
• 支持Web、Electron等多种客户端形态
• 实现实时日志推送和状态同步
• 支持游戏窗口透明浮层
• 提供完善的并发控制机制

1.3 关键需求

1. WebSocket实时通信：服务端推送日志和运行状态
2. Electron支持：打包成桌面应用，支持页面刷新保持状态
3. 透明浮层：在游戏窗口上显示透明浮层
4. 并发控制：API层面的并发限制，避免资源冲突

2. 架构设计

2.1 整体架构

┌─────────────────────────────────────────────────────────────┐
│                         客户端层                              │
├──────────────┬──────────────┬──────────────┬────────────────┤
│  Web Client  │   Electron   │  Mobile App  │  CLI Client    │
│  (Vue/React) │   Desktop    │   (Future)   │    (Future)    │
└──────┬───────┴──────┬───────┴──────┬───────┴────────┬───────┘
       │              │              │                │
       └──────────────┴──────────────┴────────────────┘
                            │
                    ┌───────▼───────┐
                    │   API Gateway  │
                    │   (FastAPI)    │
                    └───────┬───────┘
                            │
        ┌───────────────────┼───────────────────┐
        │                   │                   │
┌───────▼───────┐   ┌──────▼──────┐   ┌───────▼───────┐
│  RESTful API  │   │  WebSocket  │   │  Static Files │
│   Endpoints   │   │   Handler   │   │    Server     │
└───────┬───────┘   └──────┬──────┘   └───────────────┘
        │                  │
        └──────────┬───────┘
                   │
           ┌───────▼───────┐
           │  Service Layer │
           │   (Business)   │
           └───────┬───────┘
                   │
    ┌──────────────┼──────────────┐
    │              │              │
┌───▼───┐   ┌─────▼─────┐   ┌───▼───┐
│Context│   │Task Queue │   │Logger │
│Manager│   │ Manager   │   │System │
└───────┘   └───────────┘   └───────┘

2.2 技术栈

后端 (Python)

• 框架: FastAPI 0.104+
• 异步运行时: asyncio + uvicorn
• WebSocket: FastAPI WebSocket
• 任务队列: asyncio.Queue + Custom TaskManager
• 日志系统: Python logging + Custom WebSocketHandler
• 数据验证: Pydantic
• 并发控制: asyncio.Semaphore

前端 (Web)

• 框架: Vue 3 / React 18
• 状态管理: Pinia / Redux Toolkit
• UI组件库: Ant Design / Element Plus
• HTTP客户端: Axios
• WebSocket: native WebSocket API / Socket.io-client
• 构建工具: Vite

桌面客户端 (Electron/Tauri/WPF/Avalonia)

• 框架: Electron 27+
• 进程通信: IPC (Inter-Process Communication)
• 透明窗口: Electron transparent window
• 打包工具: electron-builder

3. 详细设计

3.1 后端服务设计

3.1.1 API结构

src/zzz_od/api/
├── main.py                 # FastAPI应用入口
├── deps.py                 # 依赖注入
├── security.py             # 认证授权
├── models.py               # Pydantic模型
├── routers/                # API路由
│   ├── auth.py            # 认证相关
│   ├── tasks.py           # 任务管理
│   ├── config.py          # 配置管理
│   ├── game.py            # 游戏控制
│   └── overlay.py         # 浮层控制
├── services/               # 业务服务
│   ├── task_manager.py    # 任务管理器
│   ├── log_service.py     # 日志服务
│   └── game_service.py    # 游戏服务
├── ws/                     # WebSocket
│   ├── manager.py         # 连接管理
│   ├── handlers.py        # 消息处理
│   └── events.py          # 事件定义
└── middleware/             # 中间件
    ├── cors.py            # CORS配置
    ├── rate_limit.py      # 速率限制
    └── error_handler.py   # 错误处理

3.1.2 WebSocket日志推送实现

# src/zzz_od/api/services/log_service.py
import logging
import asyncio
from typing import Dict, Set
from datetime import datetime

class WebSocketLogHandler(logging.Handler):
    """WebSocket日志处理器"""
    
    def __init__(self, ws_manager):
        super().__init__()
        self.ws_manager = ws_manager
        self.task_logs: Dict[str, list] = {}  # 任务ID -> 日志列表
        
    def emit(self, record):
        """发送日志到WebSocket"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "level": record.levelname,
            "message": self.format(record),
            "module": record.module,
            "funcName": record.funcName,
            "lineno": record.lineno
        }
        
        # 获取当前任务ID（从上下文中）
        task_id = getattr(record, 'task_id', None)
        
        if task_id:
            # 存储日志到任务
            if task_id not in self.task_logs:
                self.task_logs[task_id] = []
            self.task_logs[task_id].append(log_entry)
            
            # 推送到WebSocket
            asyncio.create_task(
                self.ws_manager.broadcast_to_task(
                    task_id, 
                    {"type": "log", "data": log_entry}
                )
            )
        
        # 全局日志广播
        asyncio.create_task(
            self.ws_manager.broadcast_global(
                {"type": "log", "data": log_entry}
            )
        )

3.1.3 任务队列与并发控制

# src/zzz_od/api/services/task_manager.py
import asyncio
from typing import Dict, Optional, Any
from dataclasses import dataclass
from enum import Enum
import uuid

class TaskPriority(Enum):
    LOW = 0
    NORMAL = 1
    HIGH = 2
    URGENT = 3

@dataclass
class Task:
    id: str
    type: str
    payload: Dict[str, Any]
    priority: TaskPriority = TaskPriority.NORMAL
    status: str = "pending"
    result: Optional[Any] = None
    error: Optional[str] = None

class TaskManager:
    """任务管理器，支持队列和并发控制"""
    
    def __init__(self, max_concurrent: int = 3):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.task_queue = asyncio.PriorityQueue()
        self.running_tasks: Dict[str, Task] = {}
        self.task_results: Dict[str, Task] = {}
        
    async def submit_task(self, task_type: str, payload: Dict, 
                          priority: TaskPriority = TaskPriority.NORMAL) -> str:
        """提交任务到队列"""
        task_id = str(uuid.uuid4())
        task = Task(id=task_id, type=task_type, payload=payload, priority=priority)
        
        # 优先级队列（优先级值越大越优先）
        await self.task_queue.put((
            -priority.value,  # 负数使得值越大优先级越高
            task_id,
            task
        ))
        
        return task_id
    
    async def process_tasks(self):
        """处理任务队列"""
        while True:
            try:
                # 从队列获取任务
                _, task_id, task = await self.task_queue.get()
                
                # 并发控制
                async with self.semaphore:
                    self.running_tasks[task_id] = task
                    task.status = "running"
                    
                    try:
                        # 执行任务
                        result = await self._execute_task(task)
                        task.result = result
                        task.status = "completed"
                    except Exception as e:
                        task.error = str(e)
                        task.status = "failed"
                    finally:
                        del self.running_tasks[task_id]
                        self.task_results[task_id] = task
                        
            except asyncio.CancelledError:
                break
            except Exception as e:
                logging.error(f"Task processing error: {e}")
    
    async def cancel_task(self, task_id: str) -> bool:
        """取消任务"""
        if task_id in self.running_tasks:
            # 实现任务取消逻辑
            task = self.running_tasks[task_id]
            task.status = "cancelled"
            return True
        return False

3.2 前端设计

3.2.1 项目结构

frontend/
├── src/
│   ├── api/               # API客户端
│   │   ├── client.ts     # HTTP客户端
│   │   ├── websocket.ts  # WebSocket客户端
│   │   └── types.ts      # 类型定义
│   ├── stores/           # 状态管理
│   │   ├── task.ts       # 任务状态
│   │   ├── config.ts     # 配置状态
│   │   └── log.ts        # 日志状态
│   ├── components/       # 通用组件
│   │   ├── TaskList.vue  # 任务列表
│   │   ├── LogViewer.vue # 日志查看器
│   │   └── ConfigPanel.vue # 配置面板
│   ├── views/            # 页面组件
│   │   ├── Dashboard.vue # 仪表盘
│   │   ├── Tasks.vue     # 任务管理
│   │   └── Settings.vue  # 设置页面
│   └── utils/            # 工具函数
│       ├── storage.ts    # 本地存储
│       └── reconnect.ts  # 重连逻辑
├── electron/             # Electron相关
│   ├── main.ts          # 主进程
│   ├── preload.ts       # 预加载脚本
│   └── overlay.ts       # 透明浮层窗口
└── package.json

3.2.2 WebSocket连接管理

// src/api/websocket.ts
export class WebSocketManager {
    private ws: WebSocket | null = null;
    private reconnectTimer: number | null = null;
    private reconnectAttempts = 0;
    private maxReconnectAttempts = 10;
    private reconnectDelay = 1000;
    
    constructor(private url: string) {
        this.connect();
    }
    
    private connect(): void {
        this.ws = new WebSocket(this.url);
        
        this.ws.onopen = () => {
            console.log('WebSocket connected');
            this.reconnectAttempts = 0;
            this.reconnectDelay = 1000;
            
            // 恢复订阅
            this.restoreSubscriptions();
        };
        
        this.ws.onmessage = (event) => {
            const message = JSON.parse(event.data);
            this.handleMessage(message);
        };
        
        this.ws.onclose = () => {
            console.log('WebSocket disconnected');
            this.attemptReconnect();
        };
        
        this.ws.onerror = (error) => {
            console.error('WebSocket error:', error);
        };
    }
    
    private attemptReconnect(): void {
        if (this.reconnectAttempts >= this.maxReconnectAttempts) {
            console.error('Max reconnection attempts reached');
            return;
        }
        
        this.reconnectAttempts++;
        this.reconnectTimer = window.setTimeout(() => {
            console.log(`Reconnecting... (attempt ${this.reconnectAttempts})`);
            this.connect();
        }, this.reconnectDelay);
        
        // 指数退避
        this.reconnectDelay = Math.min(this.reconnectDelay * 2, 30000);
    }
    
    private restoreSubscriptions(): void {
        // 从本地存储恢复订阅状态
        const subscriptions = localStorage.getItem('ws_subscriptions');
        if (subscriptions) {
            const subs = JSON.parse(subscriptions);
            subs.forEach((sub: any) => {
                this.subscribe(sub.channel, sub.params);
            });
        }
    }
    
    public subscribe(channel: string, params?: any): void {
        if (this.ws?.readyState === WebSocket.OPEN) {
            this.ws.send(JSON.stringify({
                type: 'subscribe',
                channel,
                params
            }));
            
            // 保存订阅到本地存储
            this.saveSubscription(channel, params);
        }
    }
    
    private saveSubscription(channel: string, params?: any): void {
        const subscriptions = JSON.parse(
            localStorage.getItem('ws_subscriptions') || '[]'
        );
        subscriptions.push({ channel, params });
        localStorage.setItem('ws_subscriptions', JSON.stringify(subscriptions));
    }
}

3.3 Electron透明浮层实现

3.3.1 主进程配置

// electron/main.ts
import { app, BrowserWindow, ipcMain, screen } from 'electron';

let mainWindow: BrowserWindow | null = null;
let overlayWindow: BrowserWindow | null = null;

function createOverlayWindow(): void {
    const display = screen.getPrimaryDisplay();
    const { width, height } = display.workAreaSize;
    
    overlayWindow = new BrowserWindow({
        width: 400,
        height: 200,
        x: width - 420,
        y: 20,
        frame: false,
        transparent: true,
        alwaysOnTop: true,
        skipTaskbar: true,
        resizable: false,
        webPreferences: {
            nodeIntegration: false,
            contextIsolation: true,
            preload: path.join(__dirname, 'preload.js')
        }
    });
    
    // 设置窗口点击穿透（Windows）
    overlayWindow.setIgnoreMouseEvents(true, { forward: true });
    
    // 加载浮层页面
    overlayWindow.loadFile('overlay.html');
    
    // 处理鼠标事件
    ipcMain.on('overlay-mouse-event', (event, data) => {
        if (data.type === 'enter') {
            overlayWindow?.setIgnoreMouseEvents(false);
        } else if (data.type === 'leave') {
            overlayWindow?.setIgnoreMouseEvents(true, { forward: true });
        }
    });
}

// IPC通信处理
ipcMain.on('show-overlay', (event, data) => {
    if (!overlayWindow) {
        createOverlayWindow();
    }
    overlayWindow?.webContents.send('update-overlay', data);
    overlayWindow?.show();
});

ipcMain.on('hide-overlay', () => {
    overlayWindow?.hide();
});

ipcMain.on('update-overlay-position', (event, position) => {
    overlayWindow?.setPosition(position.x, position.y);
});

3.3.2 浮层渲染进程

<!-- overlay.html -->
<!DOCTYPE html>
<html>
<head>
    <style>
        body {
            margin: 0;
            padding: 10px;
            background: rgba(0, 0, 0, 0.7);
            color: white;
            font-family: 'Microsoft YaHei', sans-serif;
            border-radius: 8px;
            backdrop-filter: blur(10px);
            -webkit-app-region: drag;
        }
        
        .content {
            -webkit-app-region: no-drag;
        }
        
        .status-bar {
            display: flex;
            justify-content: space-between;
            margin-bottom: 10px;
        }
        
        .progress {
            height: 20px;
            background: rgba(255, 255, 255, 0.2);
            border-radius: 10px;
            overflow: hidden;
        }
        
        .progress-bar {
            height: 100%;
            background: linear-gradient(90deg, #4CAF50, #8BC34A);
            transition: width 0.3s ease;
        }
        
        .info-grid {
            display: grid;
            grid-template-columns: 1fr 1fr;
            gap: 10px;
            margin-top: 10px;
        }
        
        .info-item {
            display: flex;
            align-items: center;
        }
        
        .info-label {
            font-size: 12px;
            opacity: 0.8;
            margin-right: 5px;
        }
        
        .info-value {
            font-size: 14px;
            font-weight: bold;
        }
    </style>
</head>
<body>
    <div class="content">
        <div class="status-bar">
            <span id="task-name">空闲</span>
            <span id="task-status">--</span>
        </div>
        <div class="progress">
            <div class="progress-bar" id="progress-bar" style="width: 0%"></div>
        </div>
        <div class="info-grid">
            <div class="info-item">
                <span class="info-label">耗时:</span>
                <span class="info-value" id="elapsed-time">00:00</span>
            </div>
            <div class="info-item">
                <span class="info-label">剩余:</span>
                <span class="info-value" id="remaining-time">--:--</span>
            </div>
            <div class="info-item">
                <span class="info-label">成功:</span>
                <span class="info-value" id="success-count">0</span>
            </div>
            <div class="info-item">
                <span class="info-label">失败:</span>
                <span class="info-value" id="failed-count">0</span>
            </div>
        </div>
    </div>
    
    <script>
        const { ipcRenderer } = require('electron');
        
        // 接收更新
        ipcRenderer.on('update-overlay', (event, data) => {
            document.getElementById('task-name').textContent = data.taskName || '空闲';
            document.getElementById('task-status').textContent = data.status || '--';
            document.getElementById('progress-bar').style.width = `${data.progress || 0}%`;
            document.getElementById('elapsed-time').textContent = data.elapsedTime || '00:00';
            document.getElementById('remaining-time').textContent = data.remainingTime || '--:--';
            document.getElementById('success-count').textContent = data.successCount || '0';
            document.getElementById('failed-count').textContent = data.failedCount || '0';
        });
        
        // 鼠标事件处理
        document.addEventListener('mouseenter', () => {
            ipcRenderer.send('overlay-mouse-event', { type: 'enter' });
        });
        
        document.addEventListener('mouseleave', () => {
            ipcRenderer.send('overlay-mouse-event', { type: 'leave' });
        });
    </script>
</body>
</html>

4. 核心功能实现

4.1 状态持久化

// frontend/src/utils/storage.ts
export class StateManager {
    private readonly STORAGE_KEY = 'app_state';
    
    saveState(state: AppState): void {
        const serialized = JSON.stringify({
            ...state,
            timestamp: Date.now()
        });
        localStorage.setItem(this.STORAGE_KEY, serialized);
        
        // 同时保存到SessionStorage以防页面刷新
        sessionStorage.setItem(this.STORAGE_KEY, serialized);
    }
    
    loadState(): AppState | null {
        const stored = localStorage.getItem(this.STORAGE_KEY) || 
                      sessionStorage.getItem(this.STORAGE_KEY);
        
        if (!stored) return null;
        
        try {
            const state = JSON.parse(stored);
            
            // 检查状态是否过期（24小时）
            if (Date.now() - state.timestamp > 24 * 60 * 60 * 1000) {
                this.clearState();
                return null;
            }
            
            return state;
        } catch (e) {
            console.error('Failed to parse stored state:', e);
            return null;
        }
    }
    
    clearState(): void {
        localStorage.removeItem(this.STORAGE_KEY);
        sessionStorage.removeItem(this.STORAGE_KEY);
    }
}

4.2 并发控制实现

# src/zzz_od/api/middleware/rate_limit.py
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
import time
from collections import defaultdict
from typing import Dict, Tuple

class RateLimiter:
    """API速率限制器"""
    
    def __init__(self, requests_per_minute: int = 60):
        self.requests_per_minute = requests_per_minute
        self.requests: Dict[str, list] = defaultdict(list)
        
    async def __call__(self, request: Request, call_next):
        client_id = request.client.host
        now = time.time()
        
        # 清理过期的请求记录
        self.requests[client_id] = [
            req_time for req_time in self.requests[client_id]
            if now - req_time < 60
        ]
        
        # 检查速率限制
        if len(self.requests[client_id]) >= self.requests_per_minute:
            return JSONResponse(
                status_code=429,
                content={
                    "error": {
                        "code": "RATE_LIMITED",
                        "message": "Too many requests",
                        "retry_after": 60
                    }
                }
            )
        
        # 记录请求
        self.requests[client_id].append(now)
        
        # 继续处理请求
        response = await call_next(request)
        return response

# 任务级别的并发控制
class TaskConcurrencyLimiter:
    """任务并发限制器"""
    
    def __init__(self):
        self.limits = {
            "battle": 1,      # 战斗任务同时只能有1个
            "patrol": 2,      # 巡逻任务最多2个
            "collection": 5,  # 收集任务最多5个
            "default": 3      # 默认最多3个
        }
        self.current_tasks: Dict[str, int] = defaultdict(int)
        
    def can_execute(self, task_type: str) -> Tuple[bool, str]:
        """检查任务是否可以执行"""
        limit = self.limits.get(task_type, self.limits["default"])
        current = self.current_tasks[task_type]
        
        if current >= limit:
            return False, f"任务类型 {task_type} 已达到并发限制 ({limit})"
        
        return True, ""
    
    def acquire(self, task_type: str) -> bool:
        """获取任务执行权限"""
        can_exec, _ = self.can_execute(task_type)
        if can_exec:
            self.current_tasks[task_type] += 1
            return True
        return False
    
    def release(self, task_type: str):
        """释放任务执行权限"""
        if self.current_tasks[task_type] > 0:
            self.current_tasks[task_type] -= 1

5. 用户故事 (User Stories)

5.1 核心用户故事

ID	角色	故事	验收标准
US001	用户	作为用户，我希望通过Web界面控制游戏辅助功能，以便在任何设备上使用	1. 可以通过浏览器访问 2. 支持移动端响应式 3. 功能完整性与桌面版一致
US002	用户	作为用户，我希望实时看到运行日志和状态，以便了解执行进度	1. 日志实时推送延迟<1秒 2. 支持日志级别过滤 3. 支持日志搜索和导出
US003	用户	作为用户，我希望在游戏窗口上看到透明浮层提示，以便获得实时辅助信息	1. 浮层始终在游戏窗口之上 2. 透明度可调节 3. 可拖动位置
US004	用户	作为用户，我希望刷新页面后保持当前任务状态，以便不丢失进度	1. 刷新后任务继续运行 2. 状态自动恢复 3. WebSocket自动重连
US005	用户	作为用户，我希望能够同时运行多个不冲突的任务，以提高效率	1. 支持任务队列 2. 自动检测任务冲突 3. 提供优先级设置
US006	开发者	作为开发者，我希望有清晰的API文档，以便二次开发	1. OpenAPI文档自动生成 2. 提供SDK和示例代码 3. 支持Postman导入
US007	运维	作为运维人员，我希望有完善的日志和监控，以便排查问题	1. 结构化日志输出 2. 性能指标采集 3. 错误报警机制

5.2 扩展用户故事

ID	角色	故事	验收标准
US008	用户	作为用户，我希望能够创建和管理多个配置方案	1. 支持配置导入导出 2. 配置版本管理 3. 配置共享功能
US009	用户	作为用户，我希望能够设置任务执行计划	1. 支持定时任务 2. 支持循环任务 3. 支持条件触发
US010	用户	作为用户，我希望有任务执行历史记录	1. 记录保存30天 2. 支持统计分析 3. 支持导出报表

6. 功能需求 (Feature Requirements)

6.1 后端功能需求

功能模块	需求描述	优先级	备注
认证授权	- API Key认证 - JWT Token支持 - 权限控制	P0	必须
RESTful API	- 完整的CRUD接口 - 统一的错误处理 - 请求验证	P0	必须
WebSocket	- 实时日志推送 - 状态更新推送 - 心跳保活	P0	必须
任务管理	- 任务队列 - 优先级调度 - 任务取消	P0	必须
并发控制	- API速率限制 - 任务并发限制 - 资源锁机制	P0	必须
日志系统	- 分级日志 - 日志持久化 - 日志检索	P1	重要
配置管理	- 配置热加载 - 配置验证 - 配置版本化	P1	重要
监控告警	- 性能监控 - 错误追踪 - 告警通知	P2	可选

6.2 前端功能需求

功能模块	需求描述	优先级	备注
任务管理界面	- 任务列表展示 - 任务创建/编辑 - 任务状态监控	P0	必须
实时日志	- 日志实时展示 - 日志过滤搜索 - 日志导出	P0	必须
配置管理界面	- 配置编辑 - 配置导入导出 - 配置预设	P0	必须
状态持久化	- 本地存储 - 状态恢复 - 数据同步	P0	必须
响应式设计	- 移动端适配 - 自适应布局 - 触摸优化	P1	重要
主题切换	- 明暗主题 - 自定义主题 - 主题持久化	P2	可选
国际化	- 多语言支持 - 语言切换 - 自动检测	P2	可选

6.3 Electron功能需求

功能模块	需求描述	优先级	备注
透明浮层	- 游戏窗口检测 - 浮层定位 - 透明度调节	P0	必须
系统托盘	- 最小化到托盘 - 托盘菜单 - 消息通知	P1	重要
快捷键	- 全局快捷键 - 快捷键自定义 - 快捷键冲突检测	P1	重要
自动更新	- 版本检测 - 增量更新 - 回滚机制	P1	重要
本地文件	- 文件读写 - 目录选择 - 文件监控	P1	重要
性能优化	- 内存优化 - 启动优化 - 渲染优化	P2	可选

7. 系统流程图

7.1 任务执行流程

graph TD
    A[用户提交任务] --> B{认证检查}
    B -->|失败| C[返回401]
    B -->|成功| D{并发检查}
    D -->|超限| E[加入队列]
    D -->|通过| F[创建任务]
    E --> G[等待调度]
    G --> F
    F --> H[返回任务ID]
    H --> I[执行任务]
    I --> J{执行状态}
    J -->|进行中| K[推送日志]
    K --> J
    J -->|成功| L[更新状态]
    J -->|失败| M[记录错误]
    L --> N[推送结果]
    M --> N
    N --> O[清理资源]

Loading

7.2 WebSocket通信流程

sequenceDiagram
    participant C as Client
    participant S as Server
    participant T as Task
    participant L as Logger
    
    C->>S: 建立WebSocket连接
    S->>C: 连接确认
    C->>S: 订阅任务channel
    S->>C: 订阅成功
    
    C->>S: 启动任务(HTTP)
    S->>T: 创建任务
    T->>S: 返回任务ID
    S->>C: 返回任务ID
    
    T->>L: 输出日志
    L->>S: 日志事件
    S->>C: 推送日志(WS)
    
    T->>S: 状态更新
    S->>C: 推送状态(WS)
    
    Note over C,S: 网络断开
    C->>C: 检测断开
    C->>S: 重新连接
    S->>C: 连接确认
    C->>S: 恢复订阅
    S->>C: 推送缓存的消息

Loading

7.3 透明浮层交互流程

graph LR
    A[主窗口] --> B[检测游戏窗口]
    B --> C{游戏运行?}
    C -->|是| D[创建浮层窗口]
    C -->|否| E[隐藏浮层]
    D --> F[获取游戏窗口位置]
    F --> G[计算浮层位置]
    G --> H[显示浮层]
    H --> I[接收任务状态]
    I --> J[更新浮层内容]
    J --> K{鼠标悬停?}
    K -->|是| L[启用交互]
    K -->|否| M[禁用交互]
    L --> N[处理用户操作]
    M --> I

Loading

8. 数据模型设计

8.1 核心数据模型

# src/zzz_od/api/models.py
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from datetime import datetime
from enum import Enum

class TaskStatus(str, Enum):
    PENDING = "pending"
    QUEUED = "queued"
    RUNNING = "running"
    PAUSED = "paused"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

class TaskType(str, Enum):
    BATTLE = "battle"
    PATROL = "patrol"
    COLLECTION = "collection"
    HOLLOW = "hollow"
    COFFEE = "coffee"
    CUSTOM = "custom"

class TaskPriority(int, Enum):
    LOW = 0
    NORMAL = 1
    HIGH = 2
    URGENT = 3

class TaskCreateRequest(BaseModel):
    type: TaskType
    name: str
    config: Dict[str, Any]
    priority: TaskPriority = TaskPriority.NORMAL
    scheduled_at: Optional[datetime] = None
    
class TaskResponse(BaseModel):
    id: str
    type: TaskType
    name: str
    status: TaskStatus
    priority: TaskPriority
    progress: float = Field(ge=0, le=100)
    created_at: datetime
    started_at: Optional[datetime]
    completed_at: Optional[datetime]
    result: Optional[Dict[str, Any]]
    error: Optional[str]
    logs_count: int = 0
    
class LogEntry(BaseModel):
    timestamp: datetime
    level: str
    message: str
    module: Optional[str]
    task_id: Optional[str]
    extra: Optional[Dict[str, Any]]

class WebSocketMessage(BaseModel):
    type: str
    channel: str
    data: Any
    timestamp: datetime = Field(default_factory=datetime.now)

class ConfigSchema(BaseModel):
    """配置模型基类"""
    version: str = "1.0.0"
    updated_at: datetime = Field(default_factory=datetime.now)
    
class GameConfig(ConfigSchema):
    """游戏配置"""
    resolution: tuple[int, int] = (1920, 1080)
    fps_limit: int = 60
    graphics_quality: str = "high"
    
class BattleConfig(ConfigSchema):
    """战斗配置"""
    auto_dodge: bool = True
    skill_priority: List[str] = []
    team_composition: List[str] = []

9. 开发计划

9.1 里程碑规划

阶段1：基础架构 (第1-2周)

目标：完成后端基础框架和前端项目搭建

任务清单：

• FastAPI基础框架搭建（已完成）
• WebSocket基础实现（已完成）
• 运行注册表实现（已完成）
• 完善日志推送机制
• 实现任务队列和并发控制
• 前端项目初始化（Vue/React）
• 基础组件开发
• API客户端封装

交付物：

• 可运行的后端服务
• 前端基础框架
• 基础API文档

阶段2：核心功能 (第3-4周)

目标：完成所有业务API和前端主要页面

任务清单：

• 完成所有业务API开发
  • 任务管理API
  • 配置管理API
  • 游戏控制API
• 前端页面开发
  • 仪表盘页面
  • 任务管理页面
  • 配置管理页面
  • 日志查看页面
• WebSocket实时通信完善
• 状态管理和持久化

交付物：

• 完整的API接口
• 功能完整的Web界面
• 集成测试用例

阶段3：Electron集成 (第5周)

目标：实现桌面客户端和透明浮层

任务清单：

• Electron项目搭建
• 主进程和渲染进程开发
• 透明浮层窗口实现
• 系统托盘功能
• 全局快捷键
• 自动更新机制

交付物：

• Electron桌面客户端
• 透明浮层功能
• 安装包构建脚本

阶段4：优化完善 (第6周)

目标：性能优化和文档完善

任务清单：

• 性能优化
  • 后端性能优化
  • 前端渲染优化
  • 内存使用优化
• 错误处理完善
• 单元测试完善
• 文档编写
  • API文档
  • 部署文档
  • 用户手册

交付物：

• 优化后的系统
• 完整的文档
• 部署脚本

9.2 风险评估

风险项	可能性	影响	缓解措施
WebSocket连接不稳定	中	高	实现重连机制和消息缓存
并发控制复杂度高	高	中	使用成熟的任务队列方案
透明浮层兼容性问题	中	中	提供备选方案（独立窗口）
性能瓶颈	中	高	提前进行性能测试和优化
前后端接口不匹配	低	中	使用OpenAPI规范和Mock数据

10. 技术决策记录

10.1 为什么选择FastAPI？

决策：使用FastAPI作为后端框架

原因：

1. 原生支持异步，性能优异
2. 自动生成OpenAPI文档
3. 内置数据验证（Pydantic）
4. WebSocket原生支持
5. 社区活跃，生态完善

替代方案：

• Flask：需要额外配置异步支持
• Django：过于重量级
• Tornado：文档和生态较差

10.2 为什么需要Electron？

决策：使用Electron实现桌面客户端

原因：

1. 支持透明窗口和always-on-top
2. 跨平台支持
3. 可以复用Web前端代码
4. 成熟的自动更新机制
5. 丰富的系统API访问

替代方案：

• Tauri：生态不够成熟
• Qt WebEngine：需要C++开发
• 纯Web：无法实现透明浮层

10.3 状态管理方案选择

决策：前端使用Pinia（Vue）或Redux Toolkit（React）

原因：

1. 官方推荐方案
2. TypeScript支持好
3. DevTools支持
4. 中间件生态丰富
5. 时间旅行调试

11. 测试策略

11.1 测试类型

测试类型	覆盖范围	工具	目标覆盖率
单元测试	业务逻辑、工具函数	pytest, Jest	80%
集成测试	API接口、数据库操作	pytest, TestClient	70%
E2E测试	用户流程、关键路径	Playwright	核心流程100%
性能测试	API响应时间、并发处理	Locust	N/A
安全测试	认证授权、输入验证	OWASP ZAP	N/A

11.2 测试用例示例

# tests/api/test_task_manager.py
import pytest
from fastapi.testclient import TestClient
from src.zzz_od.api.main import app

client = TestClient(app)

class TestTaskAPI:
    def test_create_task(self):
        """测试创建任务"""
        response = client.post("/api/v1/tasks", json={
            "type": "battle",
            "name": "自动战斗",
            "config": {"mode": "auto"},
            "priority": 1
        })
        assert response.status_code == 201
        data = response.json()
        assert "id" in data
        assert data["status"] == "pending"
    
    def test_get_task_status(self):
        """测试获取任务状态"""
        # 创建任务
        create_response = client.post("/api/v1/tasks", json={
            "type": "patrol",
            "name": "世界巡逻",
            "config": {},
        })
        task_id = create_response.json()["id"]
        
        # 获取状态
        response = client.get(f"/api/v1/tasks/{task_id}")
        assert response.status_code == 200
        data = response.json()
        assert data["id"] == task_id
        assert data["status"] in ["pending", "queued", "running"]
    
    @pytest.mark.asyncio
    async def test_websocket_connection(self):
        """测试WebSocket连接"""
        from fastapi.testclient import TestClient
        
        with client.websocket_connect("/ws/v1/tasks") as websocket:
            # 发送订阅消息
            websocket.send_json({
                "type": "subscribe",
                "channel": "task_status"
            })
            
            # 接收确认
            data = websocket.receive_json()
            assert data["type"] == "subscribed"
            assert data["channel"] == "task_status"

12. 部署方案

12.1 开发环境

# 后端启动
cd backend
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install -r requirements.txt
uvicorn src.zzz_od.api.main:app --reload --host 0.0.0.0 --port 8000

# 前端启动
cd frontend
npm install
npm run dev

# Electron开发
npm run electron:dev

12.2 生产部署

# docker-compose.yml
version: '3.8'

services:
  backend:
    build: ./backend
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db/dbname
      - REDIS_URL=redis://redis:6379
      - API_KEY=${API_KEY}
    volumes:
      - ./data:/app/data
    depends_on:
      - db
      - redis
      
  frontend:
    build: ./frontend
    ports:
      - "80:80"
    depends_on:
      - backend
      
  db:
    image: postgres:15
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=dbname
    volumes:
      - postgres_data:/var/lib/postgresql/data
      
  redis:
    image: redis:7
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  redis_data:

13. 现有实现总结

基于代码分析，项目当前已经完成的部分：

13.1 后端已实现

1. 基础框架

   • FastAPI应用框架 ✅
   • 生命周期管理（lifespan） ✅
   • CORS配置 ✅
   • 异常处理 ✅
   • 可选API Key认证 ✅

2. 核心功能

   • 运行注册表（RunRegistry） ✅
   • WebSocket基础实现 ✅
   • 事件桥接机制 ✅
   • 上下文管理（ZContext） ✅

3. API路由（部分实现）

   • home（首页相关） ✅
   • accounts（账户管理） ✅
   • onedragon（一条龙） ✅
   • settings（设置） ✅
   • resources（资源） ✅
   • world_patrol（世界巡逻） ✅

13.2 待实现功能

1. 后端待完成

   • 完善的日志推送机制
   • 任务队列实现
   • 并发控制机制
   • 状态持久化
   • 更多业务API

2. 前端待开发

   • Web前端框架搭建
   • 页面组件开发
   • 状态管理
   • WebSocket客户端

3. Electron待实现

   • 项目初始化
   • 透明浮层窗口
   • 系统集成功能

14. 下一步行动计划

立即执行（本周）

1. 完善后端日志系统

   • 实现WebSocketLogHandler
   • 集成到现有日志系统
   • 添加日志过滤和搜索API

2. 实现任务队列

   • 基于asyncio.Queue实现
   • 添加优先级调度
   • 实现并发控制

3. 前端项目初始化

   • 选择Vue 3 + Vite
   • 搭建基础项目结构
   • 实现WebSocket客户端

短期计划（2周内）

1. 完成所有业务API
2. 开发前端核心页面
3. 实现状态管理和持久化

中期计划（1个月内）

1. Electron集成
2. 透明浮层实现
3. 性能优化和测试

15. 结论

本次C/S架构重构将大幅提升系统的可扩展性和用户体验。通过分离前后端，不仅可以支持多种客户端形态，还能更好地进行团队协作开发。虽然需要投入一定的开发资源，但长期来看将带来显著的收益。

关键成功因素：

1. 保持API设计的一致性和完整性
2. 确保WebSocket连接的稳定性
3. 优化性能，特别是并发处理
4. 提供完善的文档和测试
5. 循序渐进，避免一次性改动过大

通过遵循本设计文档，项目能够顺利完成从单体应用到C/S架构的转型。

[DoctorReid]

FastAPI 接口总览与设计原则

目标：在不改变现有业务逻辑的前提下，为应用增加一个稳定的 HTTP/WS（后续可选 gRPC）边界，便于替换 GUI、集成到 MCP 或任意前端。

• 最小改动：不改动 application/domain 流程，仅新增 API 适配层与运行时任务管理。
• 新入口：保留原 python src/zzz_od/gui/app.py；服务入口新增 uvicorn zzz_od.api.main:app。
• 协议：REST（查询/变更）+ WebSocket（长任务进度）。
• 安全：默认本机监听，支持可选 X-Api-Key；CORS 默认允许本地前端。
• 版本：基础路径 /api/v1，WS /ws/v1。
• 错误：统一错误体 { error: { code, message, details? } }。
• 长任务：POST 启动返回 runId；GET 查询状态；WS 订阅事件；支持取消。

模块覆盖：

• 仪表盘、账户管理、设置、资源下载
• 战斗助手（自动战斗、躲避、操作调试、模板生成）
• 一条龙（队伍、体力计划、恶名狩猎、咖啡、石英防线）
• 零号空洞/迷失之地
• 游戏助手（委托、网络生存、鼠标灵敏度、预设队伍）
• 锄大地（路线/大地图/运行/记录）
• 开发工具（截图助手、大地图工具、图标编辑）

文档：

• 合同与事件：docs/fastapi/contract.md
• 完整接口清单：docs/fastapi/endpoints.md
• 与现有代码映射：docs/fastapi/mapping.md

运行方式（Windows PowerShell）

开发环境建议在仓库根目录执行，并将内容根 src 加入 PYTHONPATH：

Set-Location C:\Users\Anonymous\IdeaProjects\ZenlessZoneZero-OneDragon
$env:PYTHONPATH = (Resolve-Path .\src).Path
python -m pip install -U uvicorn[standard]
python -m uvicorn zzz_od.api.main:app --reload --host 127.0.0.1 --port 8000

说明：

• PYTHONPATH 指向 src 用于导入 zzz_od、one_dragon 等包。
• 生产环境可外置 Uvicorn/Gunicorn 并按需关闭 /docs、启用 API Key（环境变量 OD_API_KEY）。

[DoctorReid]

#1363

[DoctorReid]

@A-nony-mous @Qian23333 可以看得出AI写了很多的废话，两份md有用的内容可能不到20%，而且还跟pr里具体代码不符

[DoctorReid]

我建议是尽量不在设计文档里写具体代码，文档里应该重点说明各个模块的职责、关系，处理不同流程的流程图。这样就算是你们两个互相看，也比较能比较快理解对方的大概方案。

好的 prd我会重新整一份

这个是不是还没有？

[DoctorReid]

我建议是尽量不在设计文档里写具体代码，文档里应该重点说明各个模块的职责、关系，处理不同流程的流程图。这样就算是你们两个互相看，也比较能比较快理解对方的大概方案。

好的 prd我会重新整一份

关键几个组件我从pr上大概看懂了，按review内容补充一下几个核心组件想负责的内容。

周末出差了没自己电脑用，明晚再来看看

[A-nony-mous]

好的 prd我会重新整一份

这个是不是还没有？

#1372 #1374

[DoctorReid]

好的 prd我会重新整一份

这个是不是还没有？

#1372 #1374

我觉得核心内容还是太简洁了，那几个流程图

[A-nony-mous]

好的 prd 我会重新整一份

这个是不是还没有？

#1372 #1374

我觉得核心内容还是太简洁了，那几个流程图

#1382 更新了一些图标