x01.weiqi.15: AI 对弈

美丽宝磊 · 发表于 2026-4-21 22:06:00

x01.weiqi

本文档是 x01.weiqi 围棋对弈平台的核心技术参考资料，详细解释系统架构、核心模块实现、关键算法和前后端交互协议。

项目地址: x01.weiqi

系统架构概览
核心模块详解
- 2.1 main.py - FastAPI主应用
- 2.2 core/connect.py - 连接管理器
- 2.3 core/state.py - 游戏状态引擎
- 2.4 core/goai.py - AI引擎封装
- 2.5 core/auth.py - 认证与VIP系统
- 2.6 routers/ - 路由模块
- 2.7 static/script.js - 前端核心逻辑
WebSocket通信协议
Katago引擎集成
关键算法实现
部署指南
修复记录

1. 系统架构概览

1.1 系统定位

x01.weiqi 是一个基于 Web 的围棋对弈平台，集成了 KataGo AI 引擎（职业级水平），提供从入门到专业级的 AI 对手。系统采用前后端分离架构，支持实时对弈、棋谱管理、VIP会员等功能。

1.2 技术栈

层级	技术选型	说明
后端框架	Python 3.12 + FastAPI	高性能异步Web框架，支持WebSocket
实时通信	WebSocket	双向实时通信，支持断线重连
数据库	SQLite	轻量级嵌入式数据库，适合中小规模
AI引擎	KataGo v1.16.0	职业级围棋AI，EigenAVX2优化版本
AI框架	Katago	KataGo的Python封装，提供多种AI策略
前端	原生 JavaScript + Canvas	无框架依赖，高性能棋盘渲染
支付	微信支付V3	Native扫码支付

1.3 项目目录结构

x01.weiqi/
├── main.py                    # FastAPI主应用（WebSocket端点、打谱API、在线用户API）
├── tool.py                    # 运维工具脚本（密钥生成、棋谱清理）
├── core/                      # 核心后端模块
│   ├── state.py               # GameState类 - 围棋游戏核心逻辑（~3534行）
│   ├── connect.py             # ConnectionManager类 - WebSocket连接管理（~473行）
│   ├── auth.py                # 认证、用户管理、VIP、数据库操作（~829行）
│   ├── goai.py                # GoAI类 - KataGo AI引擎集成
│   ├── ai.py                  # AI策略系统（15+种策略注册表）
│   ├── game.py                # Game/BaseGame类（棋局树、落子、提子）
│   ├── game_node.py           # GameNode类（SGF节点、分析数据）
│   ├── engine.py              # KataGoEngine引擎通信（子进程+JSON协议）
│   ├── katabase.py            # KatagoBase基类（配置加载、Player类）
│   ├── sgf_parser.py          # SGF解析器（Move类、SGF类）
│   ├── constants.py           # 全局常量定义
│   ├── utils.py               # 工具函数
│   ├── wechat_pay.py          # 微信支付V3 API封装
│   ├── email_config.py        # 邮件发送配置
│   └── bj_time.py             # 北京时间工具
├── routers/                   # 路由模块（从main.py拆分）
│   ├── __init__.py            # 路由注册汇总
│   ├── deps.py                # 依赖注入（get_current_user, require_admin, check_vip_feature）
│   ├── auth.py                # 认证路由（注册/登录/改密码/重置密码）
│   ├── vip.py                 # VIP路由（订单/支付/设置）
│   ├── admin.py               # 管理路由（用户/棋谱管理）
│   ├── game.py                # 游戏路由（棋谱查看）
│   └── doc.py                 # 文档路由（Markdown管理）
├── static/                    # 前端静态资源
│   ├── index.html             # 主HTML入口
│   ├── script.js              # 前端核心JS（~4034行）
│   ├── style.css              # 样式（~1654行）
│   ├── template-loader.js     # 模板加载器（异步加载视图HTML）
│   ├── marked.min.js          # Markdown解析库
│   ├── highlight.min.js       # 代码高亮库
│   └── views/                 # HTML视图模板
│       ├── menu.html          # 菜单栏
│       ├── home.html          # 主页视图
│       ├── game.html          # 围棋对弈视图
│       ├── help.html          # 帮助视图
│       ├── doc.html           # 文档视图
│       ├── admin.html         # 管理视图
│       └── modals.html        # 模态框（密码找回、VIP购买等）
├── data/                      # 数据与配置
│   ├── config.json            # Katago配置（引擎路径、AI策略参数）
│   ├── analysis_config.cfg    # KataGo分析配置
│   └── *.bin.gz               # KataGo神经网络模型
├── requirements.txt           # Python依赖
└── Makefile                   # 构建脚本

1.4 架构图

┌─────────────────────────────────────────────────────────────────┐
│                        前端层 (Browser)                          │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │  UI渲染层    │  │  事件处理层  │  │  通信层      │          │
│  │  (Canvas)    │  │  (Events)    │  │  (WebSocket) │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘
                              ↕ WebSocket (JSON协议)
┌─────────────────────────────────────────────────────────────────┐
│                      后端层 (FastAPI)                            │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │  路由层      │  │  连接管理器  │  │  游戏状态    │          │
│  │  (routers/)  │  │  (connect.py)│  │  (GameState) │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘
                              ↕ API调用
┌─────────────────────────────────────────────────────────────────┐
│                      AI引擎层 (GoAI)                             │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  KataGo引擎 (Katago) → 策略选择 → 落子生成              │  │
│  └──────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘
                              ↕ 子进程通信 (stdin/stdout JSON)
┌─────────────────────────────────────────────────────────────────┐
│                    KataGo引擎进程                                │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │  MCTS搜索    │  │  神经网络    │  │  分析输出    │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘

2. 核心模块详解

2.1 main.py - FastAPI主应用

2.1.1 文件概述

文件路径: main.py
代码行数: ~889行
主要职责: FastAPI应用入口、WebSocket端点、打谱模式API、在线用户API、积分计算

核心组件:

应用生命周期管理（lifespan）
路由注册（从routers/模块导入）
WebSocket端点（游戏通信）
WebSocket消息处理（handle_websocket_message）
游戏结束处理（handle_game_over）
积分计算（calculate_score_change）
打谱模式API

2.1.2 应用生命周期管理

# main.py:31-49

async def session_cleanup_task():
    """定期清理过期会话"""
    while True:
        await asyncio.sleep(300)  # 5分钟
        if manager:
            manager.cleanup_expired_sessions()

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    global manager
    manager = ConnectionManager()
    cleanup_task = asyncio.create_task(session_cleanup_task())
    logging.info("[LIFESPAN] 应用启动")
    yield
    cleanup_task.cancel()
    logging.info("[LIFESPAN] 应用关闭")

app = FastAPI(lifespan=lifespan)
app.mount("/static", StaticFiles(directory="static"), name="static")

# 注册路由
app.include_router(auth_router)
app.include_router(vip_router)
app.include_router(admin_router)
app.include_router(game_router)
app.include_router(doc_router)

代码解释:

session_cleanup_task(): 异步后台任务，每5分钟清理一次过期的独立游戏会话（打谱模式）
lifespan(): FastAPI应用生命周期上下文管理器
- 启动时：创建全局 ConnectionManager 单例，启动会话清理后台任务
- 关闭时：取消后台任务
路由注册: 从 routers/ 模块导入5个路由器并注册到应用

设计模式: 生命周期管理模式 + 路由模块化

2.1.3 WebSocket端点

# main.py:109-169

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: str, token: str = None):
    """WebSocket连接处理"""
    # 验证token
    if not token:
        await websocket.close(code=1008, reason="Missing authentication token")
        return

    username = decode_token(token)
    if not username:
        await websocket.close(code=1008, reason="Invalid authentication token")
        return

    color_str = None
    try:
        success = await manager.join_room(room_id, websocket, username)
        if not success:
            return

        # lobby房间只保持连接
        if room_id == "lobby":
            while True:
                msg = await websocket.receive_json()
                if msg.get("type") == "ping":
                    await websocket.send_json({"type": "pong"})
            return

        # 接收颜色确认
        data = await websocket.receive_json()
        if data.get("type") == "color_ack":
            color_str = str(data["color"])
            # 存储用户的颜色信息
            ...

        game = manager.games[room_id]
        # 发送当前状态
        await websocket.send_json({
            "type": "state_update",
            "state": game_state_to_frontend(game)
        })

        # 消息循环
        while True:
            data = await websocket.receive_json()
            await handle_websocket_message(room_id, username, data, websocket)
    except WebSocketDisconnect:
        pass
    finally:
        if color_str:
            manager.leave_room(room_id, color_str, username)

代码解释:

WebSocket端点处理流程：

验证token: 检查JWT令牌是否有效
加入房间: 调用 manager.join_room()
lobby处理: lobby房间只保持连接，响应ping
颜色确认: 接收 color_ack 消息，存储颜色信息
发送状态: 发送当前游戏状态
消息循环: 委托给 handle_websocket_message() 处理
断线清理: finally块中调用 manager.leave_room()

2.1.4 WebSocket消息处理

# main.py:236-622

async def handle_websocket_message(room_id: str, username: str, data: dict, websocket: WebSocket):
    """处理WebSocket消息"""
    msg_type = data.get("type")
    game = manager.games.get(room_id)
    if not game:
        return

    if msg_type == "move":
        # 落子处理：验证轮次 -> place_stone() -> 广播 -> 检查game_over或AI轮次
        ...
    elif msg_type == "pass":
        # Pass处理：pass_turn() -> 广播 -> 检查game_over或AI轮次
        ...
    elif msg_type == "analyze":
        # 点目：VIP权限检查 -> analyze_territory() -> 返回analysis_result
        ...
    elif msg_type == "undo":
        # 悔棋：undo_move() -> 广播state_update
        ...
    elif msg_type == "resign":
        # 认输：设置game_over/winner/final_score -> 保存棋谱(>=50手) -> 广播 -> 清理
        ...
    elif msg_type == "end_game":
        # 结束对局：点目计算 -> AI对弈直接结束 / 人类对弈(>=170手直接结束 / <170手需对方确认)
        ...
    elif msg_type == "end_game_confirm":
        # 确认结束：end_game_manually() -> 保存棋谱 -> 计算积分 -> 广播 -> 清理
        ...

消息类型说明:

消息类型	处理逻辑
`move`	验证轮次 → place_stone() → 广播state_update → 检查game_over或AI轮次
`pass`	pass_turn() → 广播 → 检查game_over或AI轮次
`analyze`	VIP权限检查 → analyze_territory() → 返回analysis_result
`undo`	undo_move() → 广播state_update
`resign`	设置game_over → 保存棋谱(>=50手) → 广播game_over → 清理
`end_game`	点目计算 → AI直接结束 / 人类(>=170手直接结束 / <170手需确认)
`end_game_confirm`	确认结束 → 保存棋谱 → 计算积分 → 广播 → 清理

2.1.5 积分计算

# main.py:171-233

def calculate_score_change(winner_username: str, loser_username: str):
    """计算积分变化
    
    规则：
    - 同段位：胜方+1，负方-1
    - 不同段位：
      - 高段位胜：不加积分
      - 高段位负：-1
      - 低段位胜：+1
      - 低段位负：不减积分
    - AI固定为9段，不计算积分
    - 10段积分到100后：胜不加，负减1
    """

2.1.6 打谱模式API

方法	路径	职责
POST	`/api/import_sgf`	独立SGF导入（创建会话）
POST	`/api/export_sgf_session`	基于会话ID导出SGF
POST	`/api/replay_next`	打谱下一步
POST	`/api/replay_prev`	打谱上一步
POST	`/api/analyze`	独立模式点目
POST	`/api/analyze_board`	基于棋盘数据的点目分析

2.1.7 其他API

方法	路径	职责
GET	`/`	主页
GET	`/api/online-users`	获取在线用户列表
GET	`/api/generate-room`	生成新房间号
GET	`/api/user-status`	获取用户对局状态

2.2 core/connect.py - 连接管理器

2.2.1 文件概述

文件路径: core/connect.py
代码行数: ~473行
主要职责: WebSocket连接管理、房间管理、断线重连、AI落子调度、状态序列化

核心组件:

game_state_to_frontend() - 游戏状态序列化函数
ConnectionManager - 连接管理器类

2.2.2 game_state_to_frontend函数

# core/connect.py:18-61

def game_state_to_frontend(game: GameState) -> dict:
    """将游戏状态转换为前端格式
    
    转换规则：
    - 棋盘转置：board[r][c] → board_transposed[c][r]
    - 劫点转换：{row, col} → [col, row]
    - 最后一手：{row, col, stone} → [col, row, stone]
    - 地盘棋盘：同棋盘转置规则
    """
    return {
        "board": board_transposed,
        "turn": turn,
        "game_over": game.game_over,
        "winner": game.winner,
        "final_score": game.final_score,
        "ko": ko,
        "black_captured": game.black_captured,
        "white_captured": game.white_captured,
        "last_move": last_move,
        "territory_board": territory_board,
        "moves_count": game.current_move_number,
    }

关键点: 前端使用 (x, y) 坐标系（x=列, y=行），后端使用 (row, col) 坐标系，需要转置。

2.2.3 ConnectionManager类

# core/connect.py:64-102

class ConnectionManager:
    """WebSocket连接管理器"""
    
    # 类常量
    MAX_AI_ROOM_USERS = 1       # AI房间最大用户数
    SESSION_TIMEOUT = 1800      # 会话过期时间（30分钟）
    RECONNECT_TIMEOUT = 60      # 断线重连超时（60秒）
    MOVE_TIMEOUT = 60           # 落子超时（60秒）
    MAX_TIMEOUT_COUNT = 3       # 最大超时次数
    MAX_REPLAY_ANALYSIS = 3     # 打谱点目最大次数

    def __init__(self):
        # 房间管理
        self.rooms: Dict[str, Dict[str, WebSocket]] = {}       # {room_id: {color_str: WebSocket}}
        self.games: Dict[str, GameState] = {}                   # {room_id: GameState}
        self.ai_tasks: Dict[str, asyncio.Task] = {}             # {room_id: asyncio.Task}
        self.room_counter = 0                                   # 房间号计数器
        
        # 用户管理
        self.online_users: Dict[str, WebSocket] = {}            # {username: WebSocket}
        self.user_game_status: Dict[str, dict] = {}             # {username: {color, room_id, opponent}}
        self.invitations: Dict[str, dict] = {}                  # {invite_id: {from, to, room_id, status}}
        
        # AI房间管理
        self.ai_room_users: Dict[str, list] = {}                # {room_id: [username1, ...]}
        self.room_users: Dict[str, list] = {}                   # {room_id: [username1, username2]}
        
        # 会话管理
        self.standalone_games: Dict[str, dict] = {}             # {session_id: {game, created_at, last_accessed}}
        self.daily_ai_usage: Dict[str, dict] = {}               # {username: {date, count}}
        self.disconnected_users: Dict[str, dict] = {}           # {username: {room_id, color, disconnect_time}}
        self.move_timers: Dict[str, dict] = {}                  # {room_id: {turn_start_time, current_player, timeout_count}}
        self.replay_analysis_counts: Dict[str, dict] = {}       # {session_id: {count}}
        self.timer_tasks: Dict[str, asyncio.Task] = {}          # {room_id: asyncio.Task}

数据结构详解:

属性	类型	用途	示例
`rooms`	`Dict[str, Dict[str, WebSocket]]`	房间连接池，按颜色索引WebSocket	`{"room1": {"1": ws1, "2": ws2}}`
`games`	`Dict[str, GameState]`	每个房间的游戏状态实例	`{"room1": GameState实例}`
`ai_tasks`	`Dict[str, asyncio.Task]`	AI落子异步任务，支持取消	`{"room1": Task对象}`
`online_users`	`Dict[str, WebSocket]`	在线用户列表，用于大厅显示和邀请	`{"user1": ws1}`
`user_game_status`	`Dict[str, dict]`	用户对局状态，用于断线重连	`{"user1": {"room_id": "room1", "color": 1}}`
`disconnected_users`	`Dict[str, dict]`	断线缓冲区，60秒重连窗口	`{"user1": {"room_id": "room1", "disconnect_time": datetime}}`
`move_timers`	`Dict[str, dict]`	落子计时器，三次超时判负	`{"room1": {"timeout_count": {1: 0, 2: 0}}}`
`standalone_games`	`Dict[str, dict]`	独立游戏实例（打谱模式）	`{"session1": {"game": GameState}}`
`daily_ai_usage`	`Dict[str, dict]`	非VIP用户今日AI对弈次数	`{"user1": {"date": "2026-04-23", "count": 2}}`

常量配置:

常量	值	说明
`MAX_AI_ROOM_USERS`	1	AI房间最大用户数（避免卡顿）
`SESSION_TIMEOUT`	1800	会话过期时间（30分钟）
`RECONNECT_TIMEOUT`	60	断线重连超时时间（60秒）
`MOVE_TIMEOUT`	60	落子超时时间（60秒）
`MAX_TIMEOUT_COUNT`	3	最大超时次数（3次判负）
`MAX_REPLAY_ANALYSIS`	3	打谱点目最大次数

2.2.4 核心方法

join_room() - 加入房间核心逻辑:

# core/connect.py:190-351

async def join_room(self, room_id: str, websocket: WebSocket, username: str):
    """加入房间核心逻辑
    
    流程：
    1. 重连检测
    2. AI房间限制检查（VIP权限、用户已在其他AI房间、总用户数限制）
    3. 房间初始化
    4. lobby房间处理
    5. 重连处理（恢复颜色、发送状态、通知对手）
    6. 正常加入（分配颜色、判断游戏开始）
    """

leave_room() - 离开房间:

# core/connect.py:353-383

def leave_room(self, room_id: str, color_str: str, username: str):
    """离开房间
    
    流程：
    1. 从房间连接中移除
    2. 从AI房间用户列表中移除
    3. 清理用户游戏状态
    4. AI房间：清理游戏、房间、取消AI任务
    """

_schedule_ai_move() / _ai_move_task() - AI落子调度:

# core/connect.py:385-460

def _schedule_ai_move(self, room_id: str):
    """调度AI落子任务：取消旧任务，创建新异步任务"""

async def _ai_move_task(self, room_id: str):
    """AI落子异步任务
    
    流程：
    1. 等待0.3秒（给前端渲染时间）
    2. 检查游戏状态和是否轮到AI
    3. 执行AI落子（game.make_ai_move()）
    4. 广播状态更新
    5. 检查游戏是否结束（保存棋谱、清理资源）
    6. 如果AI连续落子，递归调度
    """

其他方法:

方法	职责
`cleanup_expired_sessions()`	清理过期独立游戏会话
`broadcast(room_id, message)`	广播消息到房间
`broadcast_user_list_update()`	广播用户列表更新
`get_online_users()`	获取在线用户列表
`get_online_users_with_status()`	获取带状态的在线用户列表
`generate_room_id()`	生成房间ID
`is_user_playing(username)`	检查用户是否在对局中
`cancel_ai_task(room_id)`	取消AI任务
`rejoin_users_to_online(room_id)`	将对局用户重新加入大厅

设计模式: 单例模式 + 数据管理中心模式

2.3 core/state.py - 游戏状态引擎

2.3.1 文件概述

文件路径: core/state.py
代码行数: ~3534行
主要职责: 围棋游戏核心逻辑、落子规则、提子算法、终局计分、AI集成、SGF导入导出、打谱模式

核心类:

GameState: 游戏状态主类

2.3.2 GameState类结构

# core/state.py

class GameState:
    def __init__(self, board_size=19, enable_analysis=False):
        # ========== 棋盘状态 ==========
        self.board_size = board_size
        self.black_turn = True          # 当前回合（True=黑方）
        self.board = []                 # 棋盘状态：0=空, 1=黑, 2=白
        self.move_numbers = []          # 落子序号（用于显示步数）
        self.last_move = None           # 最后一手 {row, col, stone} 或 {stone, pass}
        self.ko_point = None            # 劫点 {row, col}
        
        # ========== 提子统计 ==========
        self.black_captured = 0         # 黑方提子数
        self.white_captured = 0         # 白方提子数
        self.current_move_number = 0    # 当前手数
        
        # ========== 地盘分析 ==========
        self.territory_board = None     # 地盘归属：0=中立, 1=黑, 2=白
        self.show_territory = False     # 是否显示地盘
        
        # ========== 历史记录（支持悔棋）==========
        self.move_history = []          # 落子历史
        self.board_history = []         # 棋盘快照历史
        self.ko_history = []            # 劫点历史
        self.captured_history = []      # 提子历史
        
        # ========== AI相关 ==========
        self.ai_enabled = False         # AI是否启用
        self.ai_playing = False         # AI是否正在思考
        self.ai_color = 2               # AI执子颜色（1=黑, 2=白）
        self.goai = None                # GoAI实例（延迟创建）
        
        # ========== SGF节点树 ==========
        self.root_node = GameNode(parent=None)
        self.current_node = self.root_node
        
        # ========== 终局结果 ==========
        self.game_over = False
        self.winner = None              # 1=黑胜, 2=白胜
        self.final_score = None         # 如 "B+12.5"
        
        # ========== KataGo引擎 ==========
        self.katago = None
        self.katago_game = None
        self.enable_analysis = enable_analysis

2.3.3 主要方法分组

功能组	方法
初始化/重置	`initialize_board()`, `save_game_state()`
落子/提子	`place_stone(row, col)`, `pass_turn()`, `undo_move()`, `is_valid_move()`, `_get_captured_stones()`, `_check_ko()`
终局计分	`check_game_over()`, `calculate_final_score()`, `end_game_manually()`, `_fallback_score()`, `analyze_territory()`
Benson算法	`_benson_alive_groups()`, `_find_alive_groups()`, `identify_alive_groups()`
死子识别	`_find_dead_stones_enhanced()`, `_is_group_dead()`, `_check_influence_strength()`, `_is_big_eye_alive()`, `_is_special_dead_eye()`, `_is_true_eye()`, `_check_opponent_also_dead()`
辅助判断	`_analyze_capture_race()`, `_calc_escape_potential()`, `_count_eye_potential()`, `_calc_liberty_enclosure()`, `_calc_opponent_strength_ratio()`, `_is_point_in_opponent_territory()`
KataGo集成	`_init_analysis_engine()`, `_ensure_analysis_engine()`, `release_analysis_engine()`, `_sync_katago_game()`, `_query_katago_ownership()`, `_find_dead_stones_by_katago()`, `_calculate_score_by_katago()`
点目分析	`calculate_territory_smart()`, `_calculate_territory_heuristic()`, `_calculate_territory_endgame_style()`, `_calculate_territory_heuristic_with_verification()`
AI	`enable_ai()`, `disable_ai()`, `is_ai_turn()`, `get_ai_move()`, `make_ai_move()`
SGF	`export_sgf()`, `import_sgf()`
打谱	`enter_replay_mode()`, `replay_next()`, `replay_prev()`, `exit_replay_mode()`
其他	`cleanup()`, `clear_territory()`, `update_status()`, `analyze_position()`, `get_game_phase()`

2.3.4 落子合法性检查

# core/state.py

def is_valid_move(self, row, col, stone):
    """验证落子合法性
    
    检查步骤：
    1. 边界检查
    2. 空位检查
    3. 劫争检查
    4. 模拟落子，检查是否有气或能提子
    """

检查步骤:

边界检查: 确保坐标在棋盘范围内（0-18）
空位检查: 确保目标位置为空
劫争检查: 确保不是劫点（禁止立即回提）
自杀检查: 模拟落子，检查是否有气或能提子

2.3.5 提子算法

# core/state.py

def _get_captured_stones(self, row, col, stone):
    """获取落子后能提掉的对方棋子
    
    算法：
    1. 检查四个方向的相邻对方棋子
    2. 获取连通块
    3. 检查该连通块是否有气
    4. 如果无气，加入提子列表
    """

2.3.6 连通块与气计算

# core/state.py

def _get_group(self, row, col):
    """获取指定位置的连通块（使用DFS）"""

def _count_group_liberties(self, row, col):
    """计算指定位置连通块的气数（DFS遍历连通块，统计周围空位数量）"""

2.3.7 劫争检测

# core/state.py

def _check_ko(self, row, col, stone, captured_stones):
    """检测是否形成劫争
    
    劫争条件：
    1. 只提掉一颗子
    2. 对方立即回提也能提掉一颗子（且是刚才落下的子）
    """

2.3.8 落子执行

# core/state.py

def place_stone(self, row, col):
    """执行落子
    
    流程：
    1. 检查游戏是否结束
    2. 验证落子合法性
    3. 放置棋子
    4. 提子
    5. 检查劫争
    6. 保存历史状态
    7. 切换回合
    8. 检查终局
    """

2.4 core/goai.py - AI引擎封装

2.4.1 文件概述

文件路径: core/goai.py
主要职责: KataGo AI引擎封装、AI落子决策、策略选择

核心类:

GoAI: AI引擎主类

2.4.2 GoAI类结构

# core/goai.py

class GoAI:
    def __init__(self, game_state):
        """AI引擎初始化"""
        self.game_state = game_state
        self.board_size = game_state.board_size
        self.katago = CallableKatagoWrapper()
        self.katago.start()
        self.last_move_count = 0  # 缓存，避免重复重放

2.4.3 AI落子主流程

# core/goai.py

def get_best_move(self):
    """获取AI推荐落子
    
    流程：
    1. 检查提子机会（>=3子直接提）
    2. 获取所有合法落子
    3. 调用KataGo引擎分析
    4. 返回推荐落子
    """

性能优化: 如果能提掉>=3子，直接提子，无需引擎分析。

2.4.4 KataGo引擎调用

# core/goai.py

def _ai_move(self, legal_moves, stone):
    """高难度AI落子：使用KataGo引擎
    
    流程：
    1. 重置Katago游戏
    2. 重放历史记录（禁用分析以提高性能）
    3. 触发当前节点分析
    4. 使用策略生成落子
    5. 处理特殊情况（人类pass后AI不应pass）
    6. 验证合法性
    7. 降级：随机落子
    """

2.5 core/auth.py - 认证与VIP系统

2.5.1 文件概述

文件路径: core/auth.py
代码行数: ~829行
主要职责: 用户认证、JWT令牌、VIP会员、数据库操作

2.5.2 数据库表结构

表名	字段	说明
`users`	id, username, hashed_password, score, rank, role, email, reset_token, reset_token_expires, is_vip, vip_expire_at, created_at	用户表
`games`	id, black_player, white_player, winner, final_score, sgf_content, end_reason, created_at	棋谱表
`vip_orders`	id, username, plan_type, amount, status, trade_no, transaction_id, created_at, paid_at	VIP订单表
`vip_settings`	key, value, description	VIP功能开关表
`doc_markdown_files`	id, filename, content, uploaded_by, approved, created_at	文档表

2.5.3 VIP套餐配置

# core/auth.py

VIP_PLANS = {
    "monthly": {"name": "月度VIP", "price": 3000, "days": 30},   # 30元/月
    "yearly": {"name": "年度VIP", "price": 28800, "days": 365}   # 288元/年
}

FREE_USER_DAILY_AI_LIMIT = 3  # 非VIP用户每日AI对弈限制

2.5.4 主要函数

函数	职责
`init_db()`	初始化数据库表
`verify_password()`, `get_password_hash()`	密码验证和哈希
`get_user()`, `create_user()`, `delete_user()`	用户CRUD
`update_user_score()`, `get_user_rank_info()`	积分和段位
`save_game()`, `get_user_games()`, `get_game_sgf()`	棋谱存储和查询
`create_access_token()`, `decode_token()`	JWT令牌
`is_admin()`, `is_vip_or_admin()`	权限检查
`is_user_vip()`, `get_user_vip_info()`, `activate_vip()`	VIP状态管理
`create_vip_order()`, `complete_vip_order()`, `get_vip_order_by_trade_no()`	VIP订单
`get_vip_settings()`, `get_vip_setting()`, `set_vip_setting()`	VIP功能开关
`upload_doc_markdown()`, `get_doc_markdown_files()`, `get_doc_markdown_content()`, `delete_doc_markdown()`, `approve_doc_markdown()`	文档Markdown管理

2.5.5 JWT认证

# core/auth.py

def create_access_token(data: dict, expires_delta: timedelta = None):
    """创建JWT令牌（默认24小时过期）"""

def decode_token(token: str):
    """解码JWT令牌，返回用户名或None"""

JWT结构:

{
  "sub": "username",
  "exp": 1234567890
}

2.6 routers/ - 路由模块

2.6.1 概述

路由模块从原 main.py 中拆分，按功能域组织为5个独立模块，每个模块使用 APIRouter。

2.6.2 routers/deps.py - 依赖注入

# routers/deps.py

def get_current_user(token: str) -> str:
    """获取当前用户，验证token有效性"""

def require_admin(token: str) -> str:
    """要求管理员权限"""

def check_vip_feature(token: str, feature_key: str) -> None:
    """检查VIP功能权限"""

设计模式: 依赖注入模式，统一token验证和权限检查逻辑，避免各路由重复代码。

2.6.3 routers/auth.py - 认证路由

前缀: /api
标签: auth

方法	路径	职责
POST	`/register`	用户注册
POST	`/login`	用户登录
POST	`/change-password`	修改密码
POST	`/update-email`	更新邮箱
GET	`/user-email`	获取用户邮箱
POST	`/request-reset-password`	请求重置密码（发送邮件）

2.6.4 routers/vip.py - VIP路由

前缀: /api/vip
标签: vip

方法	路径	职责
GET	`/info`	获取VIP信息
POST	`/create-order`	创建VIP购买订单
GET	`/order-status`	查询订单支付状态
GET	`/orders`	获取用户订单列表
POST	`/notify`	微信支付回调通知
POST	`/activate-test`	测试用：直接激活VIP
GET	`/settings`	获取VIP功能开关（管理员）
POST	`/settings`	更新VIP功能开关（管理员）
GET	`/settings/public`	获取VIP功能开关公开状态

2.6.5 routers/admin.py - 管理路由

前缀: /api/admin
标签: admin

方法	路径	职责
GET	`/users`	获取所有用户列表
PUT	`/users/{username}/role`	修改用户角色
DELETE	`/users/{username}`	删除用户
GET	`/games`	获取所有棋谱列表
DELETE	`/games/{id}`	删除棋谱
POST	`/init`	初始化管理员账户

2.6.6 routers/game.py - 游戏路由

前缀: /api
标签: game

方法	路径	职责
GET	`/my-games`	获取用户历史棋谱
GET	`/user-games/{username}`	获取指定用户的棋谱
GET	`/game-sgf/{id}`	获取棋谱SGF内容

2.6.7 routers/doc.py - 文档路由

前缀: /api/doc
标签: doc

方法	路径	职责
POST	`/markdown/upload`	上传Markdown文件
GET	`/markdown/files`	获取Markdown文件列表
GET	`/markdown/{id}`	获取单个Markdown文件内容
DELETE	`/markdown/{id}`	删除Markdown文件（管理员）
POST	`/markdown/{id}/approve`	审核Markdown文件（管理员）
POST	`/markdown/{id}/unapprove`	取消审核（管理员）

2.7 static/script.js - 前端核心逻辑

2.7.1 文件概述

文件路径: static/script.js
代码行数: ~4034行
主要职责: 前端核心逻辑、棋盘渲染、WebSocket通信、UI交互

核心功能:

棋盘Canvas渲染
WebSocket通信
用户交互（落子、pass、认输等）
国际化（i18n）
模态框管理
视图切换

2.7.2 国际化配置

// static/script.js

let currentLang = localStorage.getItem('lang') || 'zh';

const i18n = {
    menu_home: { zh: '主页', en: 'Home' },
    menu_game: { zh: '围棋', en: 'Go' },
    // ... 更多翻译
};

function t(key, args = {}) {
    """翻译函数，支持模板参数替换"""
    let text = i18n[key]?.[currentLang] || key;
    for (const [k, v] of Object.entries(args)) {
        text = text.replace(`{${k}}`, v);
    }
    return text;
}

2.7.3 WebSocket连接

// static/script.js

function connectWebSocket(roomId) {
    const token = localStorage.getItem('token');
    const wsUrl = `${wsProtocol}://${window.location.host}/ws/${roomId}?token=${token}`;
    ws = new WebSocket(wsUrl);
    
    ws.onmessage = function(event) {
        const data = JSON.parse(event.data);
        handleMessage(data);
    };
}

function handleMessage(data) {
    switch(data.type) {
        case 'assign_color': ...  break;
        case 'state_update': ...  break;
        case 'game_over':    ...  break;
        case 'move_timeout': ...  break;
        // ... 更多消息类型
    }
}

2.7.4 棋盘渲染

// static/script.js

function renderBoard() {
    // 清空画布
    // 绘制棋盘线（19x19）
    // 绘制星位（9个星位点）
    // 绘制棋子（径向渐变，更真实）
    // 绘制最后一手标记
    // 绘制步数（如果启用）
}

function drawStone(ctx, row, col, stone) {
    // 绘制阴影（偏移2像素）
    // 黑子：深灰到黑色渐变
    // 白子：白色到浅灰渐变
}

2.7.5 模板加载

// static/template-loader.js

// 异步加载7个视图HTML模板，注入DOM插槽，触发templatesLoaded事件

3. WebSocket通信协议

3.1 连接流程

客户端                              服务端
   │                                  │
   │──── WebSocket连接请求 ──────────→│
   │     /ws/{room_id}?token=xxx      │
   │                                  │
   │←─── 验证token，调用join_room() ──│
   │                                  │
   │←───── assign_color ─────────────│  分配颜色
   │     {type, color, username}      │
   │                                  │
   │──── color_ack ─────────────────→│  确认颜色
   │     {type, color}                │
   │                                  │
   │←───── state_update ─────────────│  发送当前状态
   │     {type, state}                │
   │                                  │
   │        进入消息循环              │

3.2 客户端消息类型

消息类型	数据格式	功能
`color_ack`	`{type, color}`	确认颜色分配
`move`	`{type, x, y, color}`	落子
`pass`	`{type, color}`	跳过
`resign`	`{type, color}`	认输
`undo`	`{type}`	AI对弈悔棋
`undo_request`	`{type}`	请求悔棋（人对弈）
`undo_accept`	`{type}`	同意悔棋
`undo_reject`	`{type}`	拒绝悔棋
`analyze`	`{type}`	点目分析
`end_game`	`{type}`	请求结束对局
`end_game_confirm`	`{type}`	确认结束对局
`end_game_reject`	`{type}`	拒绝结束对局
`export_sgf`	`{type}`	导出SGF
`import_sgf`	`{type, sgf_content}`	导入SGF
`enter_replay`	`{type}`	进入打谱模式
`replay_next`	`{type}`	打谱下一步
`replay_prev`	`{type}`	打谱上一步
`exit_replay`	`{type}`	退出打谱模式
`ping`	`{type}`	心跳

3.3 服务端消息类型

消息类型	数据格式	功能
`assign_color`	`{type, color, username}`	分配颜色
`waiting`	`{type, message}`	等待对手
`game_start`	`{type, state}`	游戏开始
`state_update`	`{type, state}`	状态更新
`game_over`	`{type, message, winner, score, state}`	游戏结束
`error`	`{type, message}`	错误消息
`analysis_result`	`{type, data}`	分析结果
`move_timeout`	`{type, player, timeout_count, max_timeout, message}`	落子超时
`timeout_loss`	`{type, winner, final_score, message, state}`	三次超时判负
`opponent_left`	`{type, message}`	对手离开
`opponent_reconnected`	`{type, message}`	对手重连
`opponent_disconnected_temporary`	`{type, message, timeout}`	对手暂时断线
`opponent_disconnected`	`{type, message, winner, final_score, state}`	对手断线认输
`invitation`	`{type, invite_id, from, to, room_id}`	邀请通知
`invitation_accepted`	`{type, invite_id, to, room_id}`	邀请被接受
`invitation_rejected`	`{type, invite_id, to}`	邀请被拒绝
`user_list_update`	`{type, users}`	用户列表更新
`end_game_request`	`{type, from, message, result, winner, final_score}`	结束对局请求
`end_game_rejected`	`{type, message}`	结束对局被拒绝
`undo_request`	`{type, from}`	悔棋请求
`undo_accepted`	`{type, state}`	悔棋被接受
`undo_rejected`	`{type, message}`	悔棋被拒绝
`sgf_export_result`	`{type, data}`	SGF导出结果
`sgf_import_result`	`{type, data}`	SGF导入结果
`replay_result`	`{type, data}`	打谱结果
`lobby_joined`	`{type, message}`	大厅加入成功
`pong`	`{type}`	心跳回复

4. Katago引擎集成

4.1 Katago架构

GoAI (goai.py)
    │
    ├── CallableKatagoWrapper (KatagoBase)
    │       │
    │       ├── KataGoEngine (engine.py)
    │       │       │
    │       │       ├── katago_process (子进程)
    │       │       ├── queries (查询队列)
    │       │       └── 3个守护线程
    │       │               ├── _analysis_read_thread (读取stdout)
    │       │               ├── _read_stderr_thread (读取stderr)
    │       │               └── _write_stdin_thread (写入stdin)
    │       │
    │       └── Game (game.py)
    │               │
    │               ├── root (GameNode)
    │               ├── current_node (GameNode)
    │               └── engines (Dict[player, KataGoEngine])
    │
    └── STRATEGY_REGISTRY (ai.py)
            │
            ├── ai:default (DefaultStrategy)
            ├── ai:handicap (HandicapStrategy)
            ├── ai:human (HumanStyleStrategy)
            └── ... (15+种策略)

4.2 KataGoEngine通信协议

# core/engine.py

class KataGoEngine(BaseEngine):
    def start(self):
        """启动KataGo子进程"""
        cmd = [
            self.katago_path, "analysis",
            "-model", self.model_path,
            "-config", self.config_path,
            "-analysis-threads", str(self.num_threads),
        ]
        self.katago_process = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        # 启动3个守护线程
        threading.Thread(target=self._analysis_read_thread, daemon=True).start()
        threading.Thread(target=self._read_stderr_thread, daemon=True).start()
        threading.Thread(target=self._write_stdin_thread, daemon=True).start()

    def request_analysis(self, node, callback, ...):
        """请求分析"""
        query = {
            "id": str(uuid.uuid4()),
            "rules": "chinese", "komi": 7.5,
            "boardXSize": 19, "boardYSize": 19,
            "moves": [...], "maxVisits": 200,
            "includePolicy": True, "includeOwnership": True,
        }
        self.send_query(query, callback)

KataGo查询示例:

{
  "id": "query-123",
  "rules": "chinese",
  "komi": 7.5,
  "boardXSize": 19,
  "boardYSize": 19,
  "moves": [["B", "Q16"], ["W", "D4"], ["B", "Q3"]],
  "maxVisits": 200,
  "includePolicy": true,
  "includeOwnership": true
}

KataGo响应示例:

{
  "id": "query-123",
  "turnNumber": 3,
  "rootInfo": {"winrate": 0.52, "scoreLead": 1.3, "visits": 200},
  "moveInfos": [
    {"move": "Q4", "winrate": 0.55, "scoreLead": 2.1, "visits": 150},
    {"move": "D16", "winrate": 0.53, "scoreLead": 1.8, "visits": 30}
  ],
  "policy": [0.01, 0.02, ...],
  "ownership": [0.1, -0.2, ...]
}

4.3 AI策略系统

# core/ai.py

STRATEGY_REGISTRY = {}

def register_strategy(strategy_name):
    """策略注册装饰器"""
    def decorator(strategy_class):
        STRATEGY_REGISTRY[strategy_name] = strategy_class
        return strategy_class
    return decorator

@register_strategy("ai:default")
class DefaultStrategy(AIStrategy):
    """默认策略：选择引擎推荐的首选"""
    def generate_move(self):
        self.wait_for_analysis()
        moves = self.cn.candidate_moves
        if moves:
            return Move.from_gtp(moves[0]["move"], self.cn.next_player), "Best move"
        return Move.pass_move(), "Pass"

内置策略列表:

策略名	说明	适用场景
`ai:default`	默认策略	专业对弈
`ai:handicap`	让子策略	让子局
`ai:jigo`	和棋策略	目标分数0.5
`ai:scoreloss`	分损策略	教学训练
`ai:p:weighted`	加权策略	可调难度
`ai:p:pick`	选取策略	变化探索
`ai:human`	人类风格	教学演示
`ai:p:local`	局部策略	局部训练
`ai:p:tenuki`	脱先策略	战术训练

5. 关键算法实现

5.1 终局计分算法（Benson算法）

# core/state.py

def calculate_final_score(self):
    """终局精确点目
    
    算法：
    1. 识别活棋（Benson算法：2+真眼=无条件活）
    2. 标记死子并移除
    3. 洪水填充计算空点归属
    4. 中国规则计分：子空皆地
    """

算法流程:

识别活棋: 使用Benson算法（2+真眼=无条件活）
标记死子: 不在活棋组中的棋子标记为死子
洪水填充: 计算空点归属（只与一种颜色相邻的空点属于该颜色）
中国规则计分: 子空皆地 + 贴目（7.5目）

5.2 中盘形势分析

# core/state.py

def calculate_territory_smart(self):
    """中盘形势分析
    
    算法：
    - 终局阶段：使用严格死子识别
    - 布局/中盘：使用影响力方法（8格范围，非线性衰减）
    """

影响力方法:

扩散范围: 8格
衰减函数: 1.0 / (1.0 + dist)（非线性）
阈值: 0.3（影响力>0.3为黑，<-0.3为白）

5.3 KataGo增强死子识别

# core/state.py

def _find_dead_stones_by_katago(self):
    """使用KataGo的ownership数据识别死子
    
    优势：比纯规则方法更准确，特别是在复杂局面中
    """

def _calculate_score_by_katago(self):
    """使用KataGo的ownership数据计算得分"""

6. 部署指南

6.1 服务器配置

配置项	推荐值	说明
操作系统	Ubuntu 22.04/20.04	LTS版本
CPU	4核+	AI计算需求
内存	8GB+	KataGo内存占用
网络	弹性公网IP	公网访问
安全组	22, 80, 443	SSH, HTTP, HTTPS

6.2 部署架构

┌─────────────┐
│   Nginx     │  (反向代理, HTTPS, 静态文件)
│   :443      │
└─────────────┘
       ↓
┌─────────────┐
│  Gunicorn   │  (进程管理, 多worker)
│   :unix     │
└─────────────┘
       ↓
┌─────────────┐
│  Uvicorn    │  (ASGI服务器, 异步处理)
│  FastAPI    │
└─────────────┘
       ↓
┌─────────────┐
│   KataGo    │  (AI引擎进程)
│   Engine    │
└─────────────┘

6.3 Gunicorn配置

#!/bin/bash
# gunicorn_start.sh

NAME="x01.weiqi"
DIR=/opt/x01.weiqi
WORKERS=3
BIND=unix:/opt/x01.weiqi/app.sock
WORKER_CLASS=uvicorn.workers.UvicornWorker

cd $DIR
source .venv/bin/activate
export PYTHONPATH=$DIR:$PYTHONPATH

exec .venv/bin/gunicorn main:app \
  --name $NAME \
  --workers $WORKERS \
  --bind=$BIND \
  --worker-class $WORKER_CLASS \
  --timeout 1200

6.4 Nginx配置

server {
    listen 80;
    server_name your-domain.com;

    location /static/ {
        alias /opt/x01.weiqi/static/;
    }

    location / {
        proxy_pass http://unix:/opt/x01.weiqi/app.sock;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_read_timeout 3600s;
        proxy_send_timeout 3600s;
    }
}

6.5 环境变量配置

# .env

# JWT配置
JWT_SECRET_KEY=your-secret-key
JWT_ALGORITHM=HS256
JWT_EXPIRATION_HOURS=24

# SMTP配置
SMTP_SERVER=smtp.qq.com
SMTP_PORT=587
SMTP_USERNAME=your-email@qq.com
SMTP_PASSWORD=your-smtp-password

# 微信支付配置
WECHAT_PAY_MCHID=1230000109
WECHAT_PAY_APPID=wx1234567890abcdef
WECHAT_PAY_API_KEY=your32charapiv3key...
WECHAT_PAY_SERIAL_NO=444F4864EA9B34415...
WECHAT_PAY_PRIVATE_KEY_PATH=/opt/x01.weiqi/apiclient_key.pem
WECHAT_PAY_NOTIFY_URL=https://your-domain.com/api/vip/notify

7. 修复记录

7.1 2025-04-25 断线认输逻辑修复

问题描述: 玩家刷新页面断线后，对手仍处于等待落子状态，未正确显示终局信息

修复方案: 修改前端 opponent_disconnected 消息处理逻辑

先调用 handleStateUpdate(data.state) 更新游戏状态
显示终局棋盘和胜利消息
3秒后清空UI并返回大厅

7.2 2026-04-25 断线重连机制重构（重大修改）

核心问题: 原有逻辑将断线（包括刷新）直接等同于认输，这是严重的逻辑错误

修复方案: 实现断线缓冲机制，区分"暂时断线"和"断线认输"

核心改进点:

场景	修改前	修改后
用户刷新页面	立即判定认输	60秒内可重连恢复对局
网络短暂波动	对局结束	60秒内重连继续对局
真正断线	立即结束	60秒后判定认输
对手体验	立即看到胜利	先看到"等待重连"，超时后看到胜利

7.3 路由模块化重构

重构内容: 将原 main.py（~2500行）中的HTTP API路由拆分为独立模块

新增文件:

routers/__init__.py - 路由注册汇总
routers/deps.py - 依赖注入（get_current_user, require_admin, check_vip_feature）
routers/auth.py - 认证路由
routers/vip.py - VIP路由
routers/admin.py - 管理路由
routers/game.py - 游戏路由
routers/doc.py - 文档路由

新增模块:

core/connect.py - ConnectionManager类和game_state_to_frontend()函数从main.py拆出
core/email_config.py - 邮件发送配置从auth.py拆出

重构效果:

main.py 从 ~2500行减至 ~889行
路由按功能域组织，职责清晰
依赖注入模式统一权限检查逻辑
ConnectionManager独立模块，连接管理与路由逻辑分离

附录

A. 关键文件清单

文件	职责	大小
main.py	FastAPI主应用（WebSocket、打谱API）	~889行
core/state.py	游戏状态引擎	~3534行
core/connect.py	连接管理器	~473行
core/auth.py	认证与VIP	~829行
core/goai.py	AI引擎封装	~190行
core/ai.py	AI策略系统	~1500行
core/engine.py	引擎接口	~500行
core/game.py	棋局树管理	~600行
core/game_node.py	SGF节点	~400行
core/katabase.py	Katago基类	~300行
core/sgf_parser.py	SGF解析器	~400行
core/wechat_pay.py	微信支付	~200行
core/email_config.py	邮件配置	~69行
core/utils.py	工具函数	~78行
core/bj_time.py	北京时间	~19行
routers/auth.py	认证路由	~108行
routers/vip.py	VIP路由	~172行
routers/admin.py	管理路由	~99行
routers/game.py	游戏路由	~63行
routers/doc.py	文档路由	~86行
routers/deps.py	依赖注入	~27行
static/script.js	前端逻辑	~4034行
static/style.css	样式	~1654行

B. API接口清单

接口	方法	模块	说明
/ws/	WebSocket	main.py	游戏连接
/api/register	POST	routers/auth.py	用户注册
/api/login	POST	routers/auth.py	用户登录
/api/change-password	POST	routers/auth.py	修改密码
/api/update-email	POST	routers/auth.py	更新邮箱
/api/request-reset-password	POST	routers/auth.py	重置密码
/api/user-email	GET	routers/auth.py	获取邮箱
/api/vip/info	GET	routers/vip.py	VIP信息
/api/vip/create-order	POST	routers/vip.py	VIP购买
/api/vip/order-status	GET	routers/vip.py	订单状态
/api/vip/orders	GET	routers/vip.py	订单列表
/api/vip/notify	POST	routers/vip.py	微信回调
/api/vip/settings	GET/POST	routers/vip.py	VIP设置
/api/admin/users	GET	routers/admin.py	用户管理
/api/admin/games	GET	routers/admin.py	棋谱管理
/api/my-games	GET	routers/game.py	我的棋谱
/api/user-games/	GET	routers/game.py	用户棋谱
/api/game-sgf/	GET	routers/game.py	棋谱SGF
/api/doc/markdown/upload	POST	routers/doc.py	上传文档
/api/doc/markdown/files	GET	routers/doc.py	文档列表
/api/online-users	GET	main.py	在线用户
/api/generate-room	GET	main.py	生成房间
/api/user-status	GET	main.py	用户状态
/api/import_sgf	POST	main.py	SGF导入
/api/replay_next	POST	main.py	打谱下一步
/api/replay_prev	POST	main.py	打谱上一步
/api/analyze	POST	main.py	点目分析
/api/analyze_board	POST	main.py	棋盘点目

C. 模块依赖关系

main.py
  ├── core.state.GameState
  ├── core.connect.ConnectionManager, game_state_to_frontend
  ├── core.auth (decode_token, is_vip_or_admin, get_vip_setting, ...)
  └── routers (auth_router, vip_router, admin_router, game_router, doc_router)

core/connect.py
  ├── core.state.GameState
  ├── core.auth (is_vip_or_admin, get_vip_setting, get_user_rank_info, ...)
  └── core.bj_time

core/state.py
  ├── core.bj_time
  ├── core.goai.GoAI
  ├── core.game_node.GameNode
  ├── core.sgf_parser.Move, SGF
  ├── core.katabase.KatagoBase
  ├── core.engine.KataGoEngine
  └── core.game.Game

core/goai.py
  ├── core.sgf_parser.Move
  ├── core.katabase.KatagoBase
  ├── core.ai (AI_DEFAULT, STRATEGY_REGISTRY)
  └── core.constants

core/auth.py
  └── core.bj_time

routers/*
  ├── core.auth (各认证/数据库函数)
  ├── core.wechat_pay (VIP路由)
  └── routers.deps (get_current_user, require_admin, check_vip_feature)

D. 性能指标

指标	简单AI	中等AI	困难AI (KataGo)
响应时间	<10ms	<50ms	0.3-2s
CPU占用	极低	低	中-高
内存占用	<1MB	<1MB	100-500MB
准确度	随机	中等	职业级

总结

x01.weiqi AI对弈围棋系统是一个功能完整、架构清晰、性能优良的围棋对弈平台。系统核心特点：

架构清晰: 前后端分离，路由模块化，职责明确
AI强大: 集成KataGo职业级引擎，支持多难度
功能完整: 对弈、棋谱、VIP、支付等完整功能
性能优化: 异步调度、缓存优化、增量更新
稳定可靠: 完善的异常处理和降级机制
易于部署: 标准化部署流程，systemd管理
可扩展: 支持自定义策略、多引擎、配置化
模块化: 路由按功能域拆分，依赖注入统一权限检查

来源：https://www.cnblogs.com/china_x01/p/19903950