一、项目概述

1.1 项目简介

项目名称: x01.weiqi (围棋对弈系统)
技术栈: Python 3.8+ + FastAPI + WebSocket + KataGo AI引擎
主要功能: 在线围棋对弈平台，支持人机对弈、人人对弈、棋谱管理、VIP会员系统等

1.2 核心特性

• 实时对弈: 基于 WebSocket 的低延迟实时通信
• AI引擎: 集成 KataGo 引擎，提供职业级围棋AI
• 精确点目: 基于 Benson 算法的终局死子识别
• 多策略AI: 16种不同风格的AI策略
• VIP系统: 完整的会员体系和微信支付集成
• 国际化: 支持中英文双语界面

1.3 项目统计

类别	文件数量	总行数	总大小
根目录 Python 文件	2	126	3.4 KB
core/ Python 文件	15	10,989	418 KB
routers/ Python 文件	8	1,643	64 KB
static/ 前端文件	6	7,425	393 KB
static/views/ 模板	7	525	31 KB
data/ 配置文件	3	-	25 KB
总计	41	约 20,100	约 931 KB

---

二、项目架构

2.1 目录结构

	x01.weiqi/
	├── main.py # FastAPI应用入口（57行）
	├── tool.py # 开发运维工具（69行）
	├── requirements.txt # Python依赖清单
	├── Makefile # 构建部署命令
	│
	├── core/ # 核心业务逻辑模块
	│ ├── state.py # 游戏状态管理（3559行）⭐核心
	│ ├── ai.py # AI策略系统（2193行）⭐核心
	│ ├── auth.py # 用户认证授权（834行）
	│ ├── connect.py # WebSocket连接管理（862行）
	│ ├── game.py # 游戏规则实现（800行）⭐核心
	│ ├── engine.py # KataGo引擎接口（455行）
	│ ├── game_node.py # 游戏树节点（452行）
	│ ├── sgf_parser.py # SGF棋谱解析（713行）
	│ ├── katabase.py # KataGo配置管理（233行）
	│ ├── goai.py # AI封装接口（173行）
	│ ├── wechat_pay.py # 微信支付集成（273行）
	│ ├── constants.py # 全局常量定义（280行）
	│ ├── utils.py # 工具函数（76行）
	│ ├── email_config.py # 邮件服务配置（68行）
	│ └── bj_time.py # 北京时间工具（18行）
	│
	├── routers/ # API路由模块
	│ ├── ws.py # WebSocket路由（755行）⭐核心
	│ ├── game.py # 游戏相关路由（362行）
	│ ├── vip.py # VIP会员路由（171行）
	│ ├── doc.py # 文档管理路由（105行）
	│ ├── auth.py # 认证相关路由（107行）
	│ ├── admin.py # 管理员路由（98行）
	│ ├── deps.py # 路由依赖项（36行）
	│ └── __init__.py # 路由模块导出（9行）
	│
	├── static/ # 静态资源
	│ ├── script.js # 前端主逻辑（4371行）⭐核心
	│ ├── style.css # 样式表（1668行）
	│ ├── index.html # 主页HTML（36行）
	│ ├── template-loader.js # 模板加载器（69行）
	│ ├── highlight.min.js # 代码高亮库（121KB）
	│ └── marked.min.js # Markdown解析库（39KB）
	│
	├── static/views/ # 页面视图模板
	│ ├── modals.html # 模态框模板（111行）
	│ ├── help.html # 帮助页面（97行）
	│ ├── home.html # 主页视图（92行）
	│ ├── admin.html # 管理页面（73行）
	│ ├── game.html # 对弈页面（68行）
	│ ├── doc.html # 文档页面（62行）
	│ └── menu.html # 菜单栏（22行）
	│
	├── data/ # 数据和配置文件
	│ ├── config.json # 主配置文件（251行）
	│ ├── analysis_config.cfg # KataGo分析配置
	│ └── contribute_config.cfg # KataGo贡献配置
	│
	├── users.db # SQLite用户数据库
	└── .env # 环境变量配置

2.2 技术架构图

	┌─────────────────────────────────────────────────────────────┐
	│ 前端层 (static/) │
	│ index.html + script.js + style.css + views/ │
	│ - WebSocket客户端 │
	│ - Canvas棋盘渲染 │
	│ - 国际化支持(i18n) │
	└────────────────────┬────────────────────────────────────────┘
	│ WebSocket + HTTP REST API
	┌────────────────────┴────────────────────────────────────────┐
	│ 应用层 (main.py) │
	│ FastAPI应用 + 生命周期管理 + 路由注册 │
	│ - lifespan: 应用启动/关闭管理 │
	│ - session_cleanup_task: 会话清理后台任务 │
	└────────────────────┬────────────────────────────────────────┘
	│
	┌────────────┼────────────┬────────────┐
	│ │ │ │
	┌───────┴────┐ ┌────┴─────┐ ┌────┴─────┐ ┌────┴─────┐
	│ routers/ │ │ core/ │ │ data/ │ │ static/ │
	│ API路由 │ │ 核心逻辑 │ │ AI引擎 │ │ 静态资源 │
	│ │ │ │ │ │ │ │
	│ - ws.py │ │- state.py│ │- KataGo │ │- JS/CSS │
	│ - auth.py │ │- game.py │ │ 引擎 │ │- HTML │
	│ - game.py │ │- ai.py │ │- 配置 │ │- 模板 │
	│ - vip.py │ │- engine │ │ │ │ │
	│ - doc.py │ │- connect │ │ │ │ │
	└────────────┘ └──────────┘ └──────────┘ └──────────┘

2.3 数据流图

	用户操作 → WebSocket消息 → routers/ws.py → core/connect.py
	↓
	core/state.py
	↓
	core/game.py
	↓
	core/engine.py → KataGo引擎
	↓
	状态更新 → WebSocket广播 → 前端渲染

---

三、核心模块详解

3.1 main.py - 应用入口

文件位置: main.py
文件大小: 1542 字节 (57 行)
功能: FastAPI应用初始化和生命周期管理

3.1.1 核心代码结构

	# 1. 后台任务：定期清理过期会话
	async def session_cleanup_task(manager):
	"""每300秒清理一次过期会话"""
	while True:
	await asyncio.sleep(300)
	if manager:
	manager.cleanup_expired_sessions()
	# 2. 应用生命周期管理
	@asynccontextmanager
	async def lifespan(app: FastAPI):
	"""启动时创建ConnectionManager，关闭时取消清理任务"""
	manager = ConnectionManager()
	init_manager(manager)
	cleanup_task = asyncio.create_task(session_cleanup_task(manager))
	logging.info("[LIFESPAN] 应用启动")
	yield
	cleanup_task.cancel()
	logging.info("[LIFESPAN] 应用关闭")
	# 3. 创建应用并注册路由
	app = FastAPI(lifespan=lifespan)
	app.mount("/static", StaticFiles(directory="static"), name="static")
	# 注册路由
	app.include_router(auth_router) # 认证路由
	app.include_router(vip_router) # VIP路由
	app.include_router(admin_router) # 管理员路由
	app.include_router(game_router) # 游戏路由
	app.include_router(doc_router) # 文档路由
	app.include_router(ws_router) # WebSocket路由

3.1.2 关键特性

• 生命周期管理: 使用FastAPI的lifespan上下文管理器管理应用生命周期
• 后台任务: 定期清理过期会话（默认1800秒超时）
• 静态文件: 挂载到 /static 路径
• 路由注册: 集成所有路由模块
• 日志配置: 设置INFO级别日志

3.1.3 启动方式

	# 开发模式
	python main.py
	# 生产模式
	uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
	# 或使用gunicorn
	gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

---

3.2 tool.py - 开发运维工具

文件位置: tool.py
文件大小: 1864 字节 (69 行)
功能: 开发运维辅助脚本

3.2.1 核心功能

	# 1. 生成安全密钥
	def gen_secretkey():
	"""生成32字节的URL安全密钥"""
	return secrets.token_urlsafe(32)
	# 2. 删除短棋谱
	def delete_short_games(min_moves: int = 50):
	"""
	删除少于指定手数的棋谱
	Args:
	min_moves: 最小手数，默认50手
	Returns:
	deleted_count: 删除的棋谱数量
	"""
	# 计算手数：sgf_content.count(';') - 1
	# 删除少于min_moves手的棋谱
	# 3. 清理临时文件
	def clean():
	"""清理指定目录和文件"""
	del_dirs = ['__pycache__', '.vscode', '.codeartsdoer', '.pytest_cache']
	# 递归删除这些目录
	# 4. 设置用户角色
	set_user_role(username='admin', role='admin')

3.2.2 使用场景

• 初始化: 生成JWT密钥、创建管理员账户
• 维护: 清理临时文件、删除无效棋谱
• 管理: 修改用户角色、权限管理

---

3.3 core/state.py - 游戏状态管理 ⭐核心

文件位置: core/state.py
文件大小: 156590 字节 (3559 行)
功能: 围棋游戏核心逻辑，游戏状态管理，集成KataGo分析引擎

3.3.1 GameState类 - 核心属性

	class GameState:
	# 棋盘状态
	board: List[List[int]] # 0=空，1=黑，2=白
	move_numbers: List[List[int]] # 每个位置的落子手数
	black_turn: bool # 当前是否黑方行棋
	ko_point: Optional[Tuple[int, int]] # 劫点位置
	# 提子统计
	black_captured: int # 黑方提子数
	white_captured: int # 白方提子数
	# 历史记录
	move_history: List[Dict] # 历史记录（支持悔棋）
	# KataGo集成
	katago_game: Optional[Game] # KataGo游戏对象
	territory_board: List[List[int]] # 领地标记（用于点目显示）
	# 游戏模式
	mode: str # 'play', 'analyze', 'selfplay'
	ai_enabled: bool # 是否启用AI
	ai_color: int # AI执子颜色

3.3.2 基础落子规则

is_valid_move(row, col, stone) - 判断落子是否合法

	def is_valid_move(self, row: int, col: int, stone: int) -> bool:
	"""
	判断落子是否合法
	检查步骤：
	1. 位置是否在棋盘内
	2. 位置是否为空
	3. 是否为劫点
	4. 临时落子，检查是否能提子或己方有气
	"""
	# 边界检查
	if not (0 <= row < self.size and 0 <= col < self.size):
	return False
	# 空位检查
	if self.board[row][col] != 0:
	return False
	# 劫点检查
	if self.ko_point == (row, col):
	return False
	# 临时落子，检查合法性
	# ...（详细实现见源码）

place_stone(row, col) - 执行落子

	def place_stone(self, row: int, col: int) -> bool:
	"""
	执行落子
	流程：
	1. 验证落子合法性
	2. 更新棋盘状态
	3. 执行提子
	4. 检查劫争
	5. 保存历史记录
	6. 检查游戏是否结束
	"""
	# 验证
	if not self.is_valid_move(row, col, stone):
	return False
	# 落子
	self.board[row][col] = stone
	self.move_numbers[row][col] = self.move_count
	# 提子
	captured = self._capture_opponent(row, col, stone)
	# 劫争检查
	self._check_ko(row, col, stone, captured)
	# 保存历史
	self.save_game_state()
	return True

3.3.3 终局精确点目（核心算法）

calculate_final_score() - 终局计分

	def calculate_final_score(self) -> Dict:
	"""
	终局计分
	算法流程：
	1. 优先使用KataGo计分（准确率最高）
	- 调用 _calculate_score_by_katago()
	- 使用KataGo的ownership数据进行计分
	- 如果成功，直接返回结果
	2. 回退到启发式算法
	- 使用 _find_dead_stones_enhanced() 识别死子
	- 移除死子后，用洪水填充计算空点归属
	- 使用中国规则（数子法）计分：子空皆地
	返回：
	{
	'black_score': float, # 黑方得分
	'white_score': float, # 白方得分
	'winner': str, # 'B' 或 'W'
	'margin': float, # 胜差
	'method': str # 'katago' 或 'heuristic'
	}
	"""

3.3.4 Benson算法 - 无条件活棋识别

_benson_alive_groups(board, color) - Benson算法核心

	def _benson_alive_groups(self, board: List[List[int]], color: int) -> List[Set]:
	"""
	Benson算法：识别无条件活棋
	算法原理：
	Benson算法通过迭代收缩"健康连通分量"来识别无条件活棋，
	比"2+真眼"更精确。
	算法步骤：
	1. 找到所有同色连通块
	2. 预计算所有空点的连通区域
	3. 判断每个空区域是否为"健康眼"（不接触对方棋子）
	4. 迭代收缩：
	- 计算每个候选块的"私有健康眼"数量
	- 私有眼 ≥ 2 的块保留
	- 其他块移除
	- 重复直到收敛
	5. 收敛后剩余的块即为无条件活棋
	优势：
	- 能识别大眼活棋（如直四、曲四、板六）
	- 能识别借劲活棋
	- 比简单的"2+真眼"判断更精确
	返回：无条件活棋的棋块列表
	"""

3.3.5 增强版死子识别

_find_dead_stones_enhanced(board, strict, use_katago) - 死子识别

	def _find_dead_stones_enhanced(
	self,
	board: List[List[int]],
	strict: bool = True,
	use_katago: bool = True
	) -> Tuple[Set[Tuple], Set[Tuple]]:
	"""
	增强版死子识别
	算法流程：
	1. 优先使用KataGo（如果允许）
	- 调用 _find_dead_stones_by_katago()
	- 使用KataGo的ownership数据判断死活
	- ownership > 0.5: 黑方领地
	- ownership < -0.5: 白方领地
	2. 启发式算法：
	- 用Benson算法识别无条件活棋
	- 收集所有非活棋的棋块
	- 合并共气棋块（共气棋子应作为整体判断）
	- 对每个合并后的棋块进行死活判断
	死活判断因素：
	- 气数：气越少越可能死
	- 被包围程度：被对方棋子包围的比例
	- 眼位潜力：能否做出两眼
	- 周围对方棋子的强度
	- 逃逸路线：是否有向开阔区域延伸的可能
	- 气点封闭度：气点是否被对方包围
	- 对杀比气：双方互相包围时比较气数
	- 大眼活棋：大眼区域是否为活形
	返回：(黑方死子集合, 白方死子集合)
	"""

3.3.6 对杀比气分析

_analyze_capture_race(group, stone, board) - 对杀分析

	def _analyze_capture_race(
	self,
	group: Set[Tuple[int, int]],
	stone: int,
	board: List[List[int]]
	) -> str:
	"""
	对杀比气分析
	算法：
	1. 找到与本块相邻的所有对方棋块
	2. 检查对方是否也在对杀中（依赖共享气）
	3. 计算双方的有效气数
	4. 比较气数决定胜负
	返回值：
	- 'win': 对杀胜，活棋
	- 'lose': 对杀败，死棋
	- 'none': 不在对杀中
	"""

3.3.7 大眼活棋判断

_is_big_eye_alive(group, stone, board) - 大眼活棋判断

	def _is_big_eye_alive(
	self,
	group: Set[Tuple[int, int]],
	stone: int,
	board: List[List[int]]
	) -> bool:
	"""
	大眼活棋判断
	死形列表：
	- 直三/曲三：被对方先手包围时可点杀
	- 方块四：2x2正方形
	- 斗笠四：4点对角形
	- 刀把五：5点T形
	- 梅花六：6点十字形
	- 盘角曲四：劫活（按死处理）
	- 角部板六：劫活（按死处理）
	活形列表：
	- 直四：4点在一条线上
	- 曲四：4点弯折形（非角上）
	- 板六：2x3矩形（非角部）
	- 更大区域（≥7点）
	返回：True=活棋，False=死棋
	"""

3.3.8 历史记录与悔棋

	def save_game_state(self):
	"""
	保存游戏状态
	保存内容：
	- 棋盘状态
	- 手数标记
	- 劫点
	- 提子数
	最多保存350步历史
	"""
	state = {
	'board': [row[:] for row in self.board],
	'move_numbers': [row[:] for row in self.move_numbers],
	'ko_point': self.ko_point,
	'black_captured': self.black_captured,
	'white_captured': self.white_captured,
	'move_count': self.move_count
	}
	self.move_history.append(state)
	# 限制历史长度
	if len(self.move_history) > 350:
	self.move_history.pop(0)
	def undo_move(self) -> bool:
	"""
	悔棋
	流程：
	1. 从历史记录恢复上一个状态
	2. 同步更新游戏节点树
	3. 清除劫点
	4. 更新手数
	"""
	if len(self.move_history) <= 1:
	return False
	self.move_history.pop()
	prev_state = self.move_history[-1]
	# 恢复状态
	self.board = [row[:] for row in prev_state['board']]
	self.move_numbers = [row[:] for row in prev_state['move_numbers']]
	# ...
	return True

---

3.4 core/game.py - 游戏规则实现 ⭐核心

文件位置: core/game.py
文件大小: 34517 字节 (800 行)
功能: 围棋游戏的核心规则实现，包括落子规则、提子、劫争等

3.4.1 BaseGame类 - 游戏基类

	class BaseGame:
	"""
	围棋游戏基类
	主要属性：
	- katago: KataGo引擎引用
	- root: 游戏树的根节点（GameNode）
	- current_node: 当前游戏节点
	- board: 棋盘状态（二维数组，存储chain id）
	- chains: 棋块列表（每个棋块是一个Move列表）
	- prisoners: 被提掉的棋子列表
	"""
	def __init__(self, size=19, rules='chinese', komi=7.5):
	self.size = size
	self.rules = rules
	self.komi = komi
	self.board = [[0] * size for _ in range(size)]
	self.chains = []
	self.prisoners = []
	self.root = GameNode()
	self.current_node = self.root

3.4.2 核心方法详解

_validate_move_and_update_chains(move, ignore_ko) - 落子验证与棋块更新

	def _validate_move_and_update_chains(
	self,
	move: Move,
	ignore_ko: bool = False
	) -> Tuple[bool, Optional[str]]:
	"""
	落子验证与棋块更新
	检查步骤：
	1. 检查落子是否合法（位置是否被占用）
	2. 合并相邻的同色棋块
	3. 检查并执行提子（对方无气的棋块）
	4. 检查劫争（ko规则）
	5. 检查自杀（根据规则集判断是否允许）
	返回：(是否合法, 错误信息)
	"""
	# 位置检查
	if self.board[move.y][move.x] != 0:
	return False, "位置已被占用"
	# 合并相邻同色棋块
	adjacent_chains = self._get_adjacent_chains(move)
	new_chain = [move]
	for chain_id in adjacent_chains:
	if self.chains[chain_id][0].player == move.player:
	new_chain.extend(self.chains[chain_id])
	# 检查提子
	opponent = 'W' if move.player == 'B' else 'B'
	captured = self._check_captures(move, opponent)
	# 检查劫争
	if not ignore_ko and self._is_ko(move, captured):
	return False, "劫争禁着"
	# 检查自杀
	if self._is_suicide(new_chain):
	if self.rules == 'chinese':
	return False, "自杀禁着"
	return True, None

play(move, ignore_ko=False) - 执行落子

	def play(self, move: Move, ignore_ko: bool = False) -> Optional[GameNode]:
	"""
	执行落子
	流程：
	1. 验证落子合法性
	2. 更新棋盘状态
	3. 创建新的游戏节点
	4. 返回新节点
	参数：
	- move: 落子对象（包含坐标和颜色）
	- ignore_ko: 是否忽略劫争检查
	返回：新游戏节点，如果非法则返回None
	"""
	valid, error = self._validate_move_and_update_chains(move, ignore_ko)
	if not valid:
	return None
	# 更新棋盘
	self.board[move.y][move.x] = len(self.chains)
	# 创建新节点
	new_node = self.current_node.play(move)
	self.current_node = new_node
	return new_node

undo(n_times=1) - 悔棋

	def undo(self, n_times: int = 1) -> bool:
	"""
	悔棋
	支持：
	- 多次悔棋
	- 悔到分支点
	- 悔到主分支
	参数：
	- n_times: 悔棋次数
	返回：是否成功
	"""
	for _ in range(n_times):
	if self.current_node.parent is None:
	return False
	self.current_node = self.current_node.parent
	# 重建棋盘状态
	self._rebuild_board_from_node(self.current_node)
	return True

redo(n_times=1) - 重做

	def redo(self, n_times: int = 1) -> bool:
	"""
	重做
	支持：
	- 前进到下一个节点
	- 快捷节点跳转
	参数：
	- n_times: 重做次数
	返回：是否成功
	"""
	for _ in range(n_times):
	if not self.current_node.children:
	return False
	self.current_node = self.current_node.children[0]
	self._rebuild_board_from_node(self.current_node)
	return True

3.4.3 Game类 - 扩展游戏类

	class Game(BaseGame):
	"""
	扩展游戏类，集成KataGo分析引擎
	新增功能：
	- 集成KataGo分析引擎
	- 支持AI分析和建议
	- 支持棋谱分析（sweep、equalize、alternative等模式）
	- 支持自我对弈（selfplay）
	"""
	def analyze_all_nodes(
	self,
	priority: int = 0,
	analyze_fast: bool = False,
	even_if_present: bool = False
	):
	"""
	分析所有节点
	参数：
	- priority: 分析优先级
	- analyze_fast: 是否快速分析
	- even_if_present: 是否强制重新分析
	流程：
	1. 遍历游戏树的所有节点
	2. 对每个节点调用KataGo进行分析
	3. 存储分析结果
	"""
	for node in self.root.nodes_in_tree:
	if not node.analysis or even_if_present:
	node.analyze(
	self.katago,
	priority=priority,
	fast=analyze_fast
	)
	def analyze_extra(self, mode: str, **kwargs):
	"""
	额外分析
	模式：
	- ponder: 持续思考模式
	- extra: 增加访问次数的深度分析
	- game: 全局分析
	- sweep: 扫描所有可能着法
	- equalize: 均衡化分析
	- alternative: 寻找替代着法
	"""
	if mode == 'ponder':
	self._analyze_ponder(**kwargs)
	elif mode == 'extra':
	self._analyze_extra(**kwargs)
	# ...
	def selfplay(
	self,
	until_move: int = 100,
	target_b_advantage: float = None
	):
	"""
	自我对弈
	参数：
	- until_move: 下到第几手
	- target_b_advantage: 目标黑方优势（用于教学）
	流程：
	AI自动下棋直到指定手数或达到目标分数
	"""
	while self.current_node.move_number < until_move:
	best_move = self.get_best_move()
	self.play(best_move)

---

3.5 core/ai.py - AI策略系统 ⭐核心

文件位置: core/ai.py
文件大小: 83404 字节 (2193 行)
功能: 多种AI对弈策略，策略注册机制，AI等级评估

3.5.1 策略注册机制

	# 策略注册表
	STRATEGIES = {}
	def register_strategy(name: str):
	"""
	策略注册装饰器
	使用方式：
	@register_strategy("my_strategy")
	def ai_my_strategy(game, **kwargs):
	# 实现你的策略
	return best_move
	"""
	def decorator(func):
	STRATEGIES[name] = func
	return func
	return decorator

3.5.2 内置策略详解

1. AI_DEFAULT - 默认策略

	@register_strategy("default")
	def ai_default(game: Game, **kwargs) -> Optional[Move]:
	"""
	默认策略：选择KataGo推荐的最佳着法
	流程：
	1. 获取当前节点的分析结果
	2. 提取最佳着法
	3. 返回Move对象
	"""
	analysis = game.current_node.analysis
	if not analysis or not analysis.get('moveInfos'):
	return None
	best_move_info = analysis['moveInfos'][0]
	move = Move.from_gtp(best_move_info['move'], game.current_player)
	return move

2. AI_RANK - 段位匹配策略

	@register_strategy("rank")
	def ai_rank(game: Game, user_rank: int = 1, **kwargs) -> Optional[Move]:
	"""
	段位匹配策略：根据用户段位调整AI强度
	参数：
	- user_rank: 用户段位（1-10段）
	策略：
	- 低段位（1-3段）：选择第3-5候选着法
	- 中段位（4-6段）：选择第2-3候选着法
	- 高段位（7-10段）：选择最佳着法
	"""
	analysis = game.current_node.analysis
	move_infos = analysis['moveInfos']
	# 根据段位选择候选着法索引
	if user_rank <= 3:
	idx = min(2, len(move_infos) - 1) # 第3候选
	elif user_rank <= 6:
	idx = min(1, len(move_infos) - 1) # 第2候选
	else:
	idx = 0 # 最佳着法
	move = Move.from_gtp(move_infos[idx]['move'], game.current_player)
	return move

3. AI_HUMAN - 人类风格

	@register_strategy("human")
	def ai_human(game: Game, **kwargs) -> Optional[Move]:
	"""
	人类风格：模拟人类思考时间和着法选择
	特点：
	- 模拟思考时间（1-3秒）
	- 偶尔选择非最优着法（10%概率）
	- 避免过于精确的着法
	"""
	import time
	import random
	# 模拟思考时间
	think_time = random.uniform(1, 3)
	time.sleep(think_time)
	analysis = game.current_node.analysis
	move_infos = analysis['moveInfos']
	# 10%概率选择次优着法
	if random.random() < 0.1 and len(move_infos) > 1:
	idx = 1
	else:
	idx = 0
	move = Move.from_gtp(move_infos[idx]['move'], game.current_player)
	return move

4. AI_PRO - 职业棋手风格

	@register_strategy("pro")
	def ai_pro(game: Game, **kwargs) -> Optional[Move]:
	"""
	职业棋手风格：高访问次数，选择最强着法
	特点：
	- 高访问次数（10000+）
	- 选择最强着法
	- 深度分析
	"""
	# 强制深度分析
	game.current_node.analyze(
	game.katago,
	max_visits=10000,
	priority=10
	)
	analysis = game.current_node.analysis
	best_move_info = analysis['moveInfos'][0]
	move = Move.from_gtp(best_move_info['move'], game.current_player)
	return move

5. AI_WEIGHTED - 加权选择

	@register_strategy("weighted")
	def ai_weighted(game: Game, **kwargs) -> Optional[Move]:
	"""
	加权选择：根据policy概率加权选择
	特点：
	- 根据policyPrior加权
	- 增加变化
	- 避免重复着法
	"""
	analysis = game.current_node.analysis
	move_infos = analysis['moveInfos'][:10] # 前10候选
	# 提取policy权重
	weights = [info['policyPrior'] for info in move_infos]
	total = sum(weights)
	weights = [w / total for w in weights]
	# 加权随机选择
	import random
	r = random.random()
	cumsum = 0
	for i, w in enumerate(weights):
	cumsum += w
	if r <= cumsum:
	move = Move.from_gtp(move_infos[i]['move'], game.current_player)
	return move
	# 默认返回最佳
	return Move.from_gtp(move_infos[0]['move'], game.current_player)

3.5.3 更多策略

	# AI_LOCAL: 局部战斗型
	@register_strategy("local")
	def ai_local(game, **kwargs):
	"""优先选择局部战斗着法"""
	pass
	# AI_TENUKI: 脱先型
	@register_strategy("tenuki")
	def ai_tenuki(game, **kwargs):
	"""优先选择脱先着法"""
	pass
	# AI_INFLUENCE: 势力型
	@register_strategy("influence")
	def ai_influence(game, **kwargs):
	"""优先选择扩张势力的着法"""
	pass
	# AI_TERRITORY: 实地型
	@register_strategy("territory")
	def ai_territory(game, **kwargs):
	"""优先选择围空的着法"""
	pass
	# AI_ATTACK: 攻击型
	@register_strategy("attack")
	def ai_attack(game, **kwargs):
	"""优先选择攻击着法"""
	pass
	# AI_DEFEND: 防守型
	@register_strategy("defend")
	def ai_defend(game, **kwargs):
	"""优先选择防守着法"""
	pass
	# AI_BALANCE: 均衡型
	@register_strategy("balance")
	def ai_balance(game, **kwargs):
	"""均衡发展"""
	pass
	# AI_SABAKI: 治孤型
	@register_strategy("sabaki")
	def ai_sabaki(game, **kwargs):
	"""优先选择治孤着法"""
	pass
	# AI_THICK: 厚势型
	@register_strategy("thick")
	def ai_thick(game, **kwargs):
	"""优先选择厚势着法"""
	pass
	# AI_LIGHT: 轻灵型
	@register_strategy("light")
	def ai_light(game, **kwargs):
	"""优先选择轻灵着法"""
	pass
	# AI_HEAVY: 重型
	@register_strategy("heavy")
	def ai_heavy(game, **kwargs):
	"""优先选择重型着法"""
	pass
	# AI_RANDOM: 随机型
	@register_strategy("random")
	def ai_random(game, **kwargs):
	"""随机选择候选着法"""
	pass

3.5.4 ELO等级系统

	# 策略ELO等级
	STRATEGY_ELO = {
	'default': 2800, # 职业强豪
	'pro': 2900, # 顶尖职业
	'rank': 2700, # 职业中游
	'human': 2600, # 业余强豪
	'weighted': 2750, # 职业上游
	'local': 2650, # 业余强手
	'tenuki': 2600, # 业余强手
	'influence': 2680, # 业余顶尖
	'territory': 2680, # 业余顶尖
	'attack': 2700, # 职业入门
	'defend': 2650, # 业余强手
	'balance': 2720, # 职业中游
	'sabaki': 2670, # 业余顶尖
	'thick': 2690, # 业余顶尖
	'light': 2660, # 业余强手
	'heavy': 2640, # 业余强手
	'random': 2000, # 业余初段
	}
	def get_strategy_by_elo(target_elo: int) -> str:
	"""
	根据目标ELO选择最接近的策略
	用于匹配不同水平的玩家
	"""
	closest = min(
	STRATEGY_ELO.items(),
	key=lambda x: abs(x[1] - target_elo)
	)
	return closest[0]

---

3.6 core/connect.py - WebSocket连接管理

文件位置: core/connect.py
文件大小: 35048 字节 (862 行)
功能: 管理WebSocket连接、房间系统、断线重连

3.6.1 ConnectionManager类 - 核心属性

	class ConnectionManager:
	"""
	WebSocket连接管理器
	配置参数：
	- MAX_AI_ROOM_USERS: AI房间最大用户数（1）
	- SESSION_TIMEOUT: 会话过期时间（1800秒）
	- RECONNECT_TIMEOUT: 断线重连超时（180秒）
	- MAX_DISCONNECT_COUNT: 最大断线次数（2）
	- MOVE_TIMEOUT: 落子超时（60秒）
	- MAX_TIMEOUT_COUNT: 最大超时次数（3）
	- MAX_REPLAY_ANALYSIS: 打谱点目最大次数（3）
	"""
	def __init__(self):
	# 房间管理
	self.rooms: Dict[str, Dict[str, WebSocket]] = {}
	# room_id -> {'black': ws1, 'white': ws2}
	# 游戏状态
	self.games: Dict[str, GameState] = {}
	# room_id -> GameState
	# 在线用户
	self.online_users: Dict[str, WebSocket] = {}
	# username -> WebSocket
	# 用户对局状态
	self.user_game_status: Dict[str, Dict] = {}
	# username -> {'room_id': str, 'color': str, 'opponent': str}
	# 断线用户（支持重连）
	self.disconnected_users: Dict[str, Dict] = {}
	# username -> {'room_id', 'color', 'opponent', 'disconnect_time', 'reconnect_timer'}
	# AI房间用户管理
	self.ai_room_users: Dict[str, List[str]] = {}
	# room_id -> [username1, username2, ...]

3.6.2 房间管理

join_room(room_id, websocket, username) - 加入房间

	async def join_room(
	self,
	room_id: str,
	websocket: WebSocket,
	username: str
	) -> Tuple[bool, Optional[str], Optional[str]]:
	"""
	加入房间
	流程：
	1. 重连检测
	- 检查用户是否在对局中
	- 检查是否在断线重连超时内
	- 检查断线次数是否超限
	2. AI房间限制检查
	- VIP用户无限制
	- 普通用户每日限制次数
	- 检查AI是否已被占用
	3. 初始化房间和游戏
	- 创建新的GameState
	- 如果是AI房间，启用AI
	4. 重连处理
	- 恢复用户颜色
	- 取消重连等待计时器
	- 替换旧的WebSocket连接
	5. 新用户处理
	- 分配颜色（黑/白）
	- 记录用户状态
	- 广播用户列表更新
	返回：(成功, 颜色, 错误信息)
	"""
	# 重连检测
	if username in self.disconnected_users:
	disconnect_info = self.disconnected_users[username]
	elapsed = time.time() - disconnect_info['disconnect_time']
	if elapsed < RECONNECT_TIMEOUT:
	# 重连成功
	return await self._handle_reconnect(username, websocket)
	# AI房间限制
	if room_id.startswith('ai:'):
	if not self._check_ai_room_permission(username, room_id):
	return False, None, "AI房间使用次数已达上限"
	# 初始化房间
	if room_id not in self.rooms:
	self.rooms[room_id] = {}
	self.games[room_id] = GameState()
	# AI房间启用AI
	if room_id.startswith('ai:'):
	self.games[room_id].ai_enabled = True
	self.games[room_id].ai_color = 2 # 白方
	# 分配颜色
	if 'black' not in self.rooms[room_id]:
	color = 'black'
	self.rooms[room_id]['black'] = websocket
	elif 'white' not in self.rooms[room_id]:
	color = 'white'
	self.rooms[room_id]['white'] = websocket
	else:
	return False, None, "房间已满"
	# 记录用户状态
	self.user_game_status[username] = {
	'room_id': room_id,
	'color': color,
	'opponent': self._get_opponent(room_id, color)
	}
	return True, color, None

3.6.3 断线重连机制

handle_disconnect(room_id, color_str, username) - 处理断线

	async def handle_disconnect(
	self,
	room_id: str,
	color_str: str,
	username: str
	):
	"""
	处理断线
	流程：
	1. 记录断线信息（时间、房间、颜色、对手）
	2. 启动重连等待计时器（180秒）
	3. 如果超时未重连：
	- 增加断线次数
	- 如果超过最大次数（2次），判负
	- 否则，等待下次重连
	重连恢复：
	- 恢复用户颜色和房间
	- 同步当前游戏状态
	- 通知对手重连成功
	"""
	# 记录断线信息
	self.disconnected_users[username] = {
	'room_id': room_id,
	'color': color_str,
	'opponent': self._get_opponent(room_id, color_str),
	'disconnect_time': time.time(),
	'disconnect_count': self._get_disconnect_count(username) + 1
	}
	# 启动重连计时器
	async def reconnect_timer():
	await asyncio.sleep(RECONNECT_TIMEOUT)
	# 超时未重连
	if username in self.disconnected_users:
	disconnect_info = self.disconnected_users[username]
	# 检查断线次数
	if disconnect_info['disconnect_count'] >= MAX_DISCONNECT_COUNT:
	# 判负
	await self._handle_timeout_loss(room_id, color_str, username)
	else:
	# 等待下次重连
	del self.disconnected_users[username]
	asyncio.create_task(reconnect_timer())

3.6.4 落子超时机制

start_move_timer(room_id, color) - 启动落子计时器

	def start_move_timer(self, room_id: str, color: str):
	"""
	启动落子计时器
	流程：
	1. 设置60秒超时
	2. 如果超时未落子：
	- 增加超时次数
	- 如果超过最大次数（3次），判负
	- 否则，继续等待
	"""
	async def timeout_handler():
	await asyncio.sleep(MOVE_TIMEOUT)
	game = self.games.get(room_id)
	if not game:
	return
	# 检查是否轮到该颜色
	if (color == 'black' and game.black_turn) or \
	(color == 'white' and not game.black_turn):
	# 超时
	timeout_count = self._get_timeout_count(room_id, color) + 1
	self._set_timeout_count(room_id, color, timeout_count)
	if timeout_count >= MAX_TIMEOUT_COUNT:
	# 判负
	await self._handle_timeout_loss(room_id, color)
	else:
	# 继续等待
	self.start_move_timer(room_id, color)
	asyncio.create_task(timeout_handler())

3.6.5 AI对弈管理

_schedule_ai_move(room_id) - 调度AI落子

	async def _schedule_ai_move(self, room_id: str):
	"""
	调度AI落子
	流程：
	1. 获取当前游戏状态
	2. 调用GoAI获取最佳着法
	3. 执行AI落子
	4. 广播状态更新
	5. 如果游戏未结束，继续等待玩家落子
	"""
	game = self.games.get(room_id)
	if not game or not game.ai_enabled:
	return
	# 获取AI着法
	ai_move = await self._get_ai_move(game)
	if not ai_move:
	return
	# 执行落子
	success = game.place_stone(ai_move[0], ai_move[1])
	if not success:
	return
	# 广播状态更新
	await self._broadcast_state(room_id)
	# 检查游戏是否结束
	if game.game_over:
	await self._handle_game_over(room_id)

---

3.7 core/auth.py - 用户认证与授权

文件位置: core/auth.py
文件大小: 27438 字节 (834 行)
功能: 用户认证、权限管理、数据库操作

3.7.1 数据库表结构

users表 - 用户信息

	CREATE TABLE users (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	username TEXT UNIQUE NOT NULL,
	hashed_password TEXT NOT NULL,
	email TEXT,
	score INTEGER DEFAULT 0, -- 积分
	rank INTEGER DEFAULT 1, -- 段位（1-10段）
	role TEXT DEFAULT 'user', -- 角色：user/admin
	is_vip BOOLEAN DEFAULT FALSE, -- VIP状态
	vip_expire_at DATETIME, -- VIP到期时间
	reset_token TEXT, -- 密码重置token
	reset_token_expires DATETIME, -- token过期时间
	created_at DATETIME DEFAULT CURRENT_TIMESTAMP
	);

games表 - 对局记录

	CREATE TABLE games (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	black_player TEXT NOT NULL,
	white_player TEXT NOT NULL,
	winner TEXT,
	final_score TEXT, -- 如 "B+3.5"
	end_reason TEXT, -- 如 "resign", "timeout"
	sgf_content TEXT, -- SGF棋谱内容
	created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
	FOREIGN KEY (black_player) REFERENCES users(username),
	FOREIGN KEY (white_player) REFERENCES users(username)
	);

vip_orders表 - VIP订单

	CREATE TABLE vip_orders (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	username TEXT NOT NULL,
	plan_type TEXT NOT NULL, -- 如 "monthly", "yearly"
	amount REAL NOT NULL,
	status TEXT DEFAULT 'pending', -- pending/paid/failed
	trade_no TEXT UNIQUE, -- 商户订单号
	transaction_id TEXT, -- 微信支付交易号
	created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
	FOREIGN KEY (username) REFERENCES users(username)
	);

vip_settings表 - VIP功能开关

	CREATE TABLE vip_settings (
	key TEXT PRIMARY KEY,
	value BOOLEAN,
	description TEXT
	);

doc_markdown_files表 - 文档管理

	CREATE TABLE doc_markdown_files (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	filename TEXT NOT NULL,
	content TEXT NOT NULL,
	uploaded_by TEXT NOT NULL,
	approved BOOLEAN DEFAULT FALSE,
	created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
	FOREIGN KEY (uploaded_by) REFERENCES users(username)
	);

3.7.2 核心功能

用户注册与登录

	def register_user(username: str, password: str, email: str = None) -> bool:
	"""
	用户注册
	流程：
	1. 检查用户名是否已存在
	2. 使用bcrypt加密密码
	3. 插入数据库
	"""
	# 检查用户名
	if get_user(username):
	return False
	# 加密密码
	hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
	# 插入数据库
	conn = sqlite3.connect(DB_PATH)
	cursor = conn.cursor()
	cursor.execute(
	"INSERT INTO users (username, hashed_password, email) VALUES (?, ?, ?)",
	(username, hashed, email)
	)
	conn.commit()
	conn.close()
	return True
	def login_user(username: str, password: str) -> Optional[str]:
	"""
	用户登录
	流程：
	1. 查询用户
	2. 验证密码
	3. 生成JWT token
	返回：JWT token，失败返回None
	"""
	user = get_user(username)
	if not user:
	return None
	# 验证密码
	if not bcrypt.checkpw(password.encode(), user['hashed_password']):
	return None
	# 生成JWT token
	token = jwt.encode({
	'username': username,
	'exp': datetime.utcnow() + timedelta(hours=24)
	}, SECRET_KEY, algorithm='HS256')
	return token

权限管理

	def is_admin(username: str) -> bool:
	"""检查是否为管理员"""
	user = get_user(username)
	return user and user['role'] == 'admin'
	def is_vip(username: str) -> bool:
	"""检查是否为VIP"""
	user = get_user(username)
	if not user or not user['is_vip']:
	return False
	# 检查是否过期
	if user['vip_expire_at']:
	expire_at = datetime.fromisoformat(user['vip_expire_at'])
	return datetime.utcnow() < expire_at
	return False
	def is_vip_or_admin(username: str) -> bool:
	"""检查是否为VIP或管理员"""
	return is_vip(username) or is_admin(username)

积分与段位

	def update_score(username: str, delta: int):
	"""
	更新积分
	积分规则：
	- 同段位：胜方+1，负方-1
	- 不同段位：
	- 高段位胜：不加积分
	- 高段位负：-1
	- 低段位胜：+1
	- 低段位负：不减积分
	- AI固定为9段，不计算积分
	- 10段积分到100后：胜不加，负减1
	"""
	user = get_user(username)
	new_score = user['score'] + delta
	# 更新段位
	new_rank = calculate_rank(new_score)
	# 更新数据库
	conn = sqlite3.connect(DB_PATH)
	cursor = conn.cursor()
	cursor.execute(
	"UPDATE users SET score = ?, rank = ? WHERE username = ?",
	(new_score, new_rank, username)
	)
	conn.commit()
	conn.close()
	def calculate_rank(score: int) -> int:
	"""
	根据积分计算段位
	段位规则：
	- 1段：0-9分
	- 2段：10-19分
	- ...
	- 10段：90+分
	"""
	return min(10, max(1, score // 10 + 1))

---

3.8 core/engine.py - KataGo引擎接口

文件位置: core/engine.py
文件大小: 19815 字节 (455 行)
功能: 与KataGo AI引擎通信

3.8.1 KataGoEngine类

	class KataGoEngine:
	"""
	KataGo引擎接口
	主要方法：
	- start(): 启动KataGo进程
	- request_analysis(): 请求分析
	- stop_pondering(): 停止思考
	"""
	def __init__(self, config_path: str = "data/config.json"):
	self.config = self._load_config(config_path)
	self.process = None
	self.query_id = 0
	self.pending_queries = {}

3.8.2 启动引擎

	def start(self):
	"""
	启动KataGo进程
	流程：
	1. 加载配置文件
	2. 启动子进程
	3. 建立管道通信
	"""
	# 构建命令
	cmd = [
	self.config['katago_path'],
	'analysis',
	'-config', self.config['config_path'],
	'-model', self.config['model_path']
	]
	# 启动进程
	self.process = subprocess.Popen(
	cmd,
	stdin=subprocess.PIPE,
	stdout=subprocess.PIPE,
	stderr=subprocess.PIPE,
	text=True,
	bufsize=1
	)
	# 启动结果读取线程
	threading.Thread(target=self._read_results, daemon=True).start()

3.8.3 请求分析

	def request_analysis(
	self,
	node: GameNode,
	callback: Callable,
	priority: int = 0,
	**kwargs
	):
	"""
	请求分析
	参数：
	- node: 游戏节点
	- callback: 结果回调函数
	- priority: 优先级（越高越优先）
	- kwargs: 其他参数（max_visits, include_policy等）
	分析请求格式：
	{
	"id": "query-id",
	"moves": ["BQ16", "WD4", ...],
	"rules": "chinese",
	"komi": 7.5,
	"boardXSize": 19,
	"boardYSize": 19,
	"maxVisits": 1000,
	"includePolicy": true,
	"includeOwnership": true
	}
	"""
	# 构建请求
	self.query_id += 1
	query_id = f"query-{self.query_id}"
	moves = [node.move.gtp() for node in node.nodes_from_root if node.move]
	request = {
	"id": query_id,
	"moves": moves,
	"rules": self.config['rules'],
	"komi": self.config['komi'],
	"boardXSize": self.config['size'],
	"boardYSize": self.config['size'],
	"maxVisits": kwargs.get('max_visits', 1000),
	"includePolicy": kwargs.get('include_policy', True),
	"includeOwnership": kwargs.get('include_ownership', True)
	}
	# 保存回调
	self.pending_queries[query_id] = callback
	# 发送请求
	self.process.stdin.write(json.dumps(request) + '\n')
	self.process.stdin.flush()

3.8.4 结果处理

	def _read_results(self):
	"""
	读取结果线程
	分析结果格式：
	{
	"id": "query-id",
	"moveInfos": [
	{
	"move": "Q16",
	"visits": 500,
	"winrate": 0.65,
	"scoreLead": 3.5,
	"policyPrior": 0.1
	},
	...
	],
	"ownership": [0.9, 0.8, -0.7, ...],
	"rootInfo": {
	"winrate": 0.6,
	"scoreLead": 2.5
	}
	}
	"""
	while True:
	line = self.process.stdout.readline()
	if not line:
	break
	try:
	result = json.loads(line)
	query_id = result['id']
	# 调用回调
	if query_id in self.pending_queries:
	callback = self.pending_queries.pop(query_id)
	callback(result)
	except Exception as e:
	logging.error(f"解析结果失败: {e}")

---

3.9 core/game_node.py - 游戏树节点

文件位置: core/game_node.py
文件大小: 18857 字节 (452 行)
功能: 棋谱树结构管理

3.9.1 GameNode类

	class GameNode:
	"""
	游戏树节点
	主要属性：
	- parent: 父节点
	- children: 子节点列表
	- move: 本节点着法（Move对象）
	- properties: SGF属性字典
	- analysis: 分析结果（从KataGo获取）
	"""
	def __init__(self, parent=None, move=None):
	self.parent = parent
	self.children = []
	self.move = move
	self.properties = {}
	self.analysis = None
	self.move_number = 0
	if parent:
	self.move_number = parent.move_number + 1

3.9.2 核心方法

	def play(self, move: Move) -> 'GameNode':
	"""
	执行着法
	流程：
	1. 查找是否已有该着法的子节点
	2. 如果有，返回现有节点
	3. 如果没有，创建新节点
	返回：子节点
	"""
	# 查找现有子节点
	for child in self.children:
	if child.move and child.move.equals(move):
	return child
	# 创建新节点
	new_node = GameNode(parent=self, move=move)
	self.children.append(new_node)
	return new_node
	@property
	def nodes_from_root(self) -> List['GameNode']:
	"""
	从根节点到本节点的路径
	返回：节点列表
	"""
	nodes = []
	current = self
	while current:
	nodes.insert(0, current)
	current = current.parent
	return nodes
	@property
	def nodes_in_tree(self) -> List['GameNode']:
	"""
	树中的所有节点
	递归遍历所有子孙节点
	返回：节点列表
	"""
	nodes = [self]
	for child in self.children:
	nodes.extend(child.nodes_in_tree)
	return nodes
	def analyze(self, engine: KataGoEngine, **kwargs):
	"""
	分析本节点
	参数：
	- engine: KataGo引擎
	- kwargs: 分析参数
	流程：
	1. 调用KataGo引擎
	2. 存储分析结果
	"""
	def callback(result):
	self.analysis = result
	engine.request_analysis(self, callback, **kwargs)

3.9.3 分析数据压缩

	# 分析结果可能很大（数百KB）
	# 使用压缩存储减少内存占用
	import gzip
	import pickle
	def set_analysis(self, analysis: dict):
	"""压缩存储分析结果"""
	self._analysis_compressed = gzip.compress(
	pickle.dumps(analysis)
	)
	def get_analysis(self) -> dict:
	"""解压获取分析结果"""
	if hasattr(self, '_analysis_compressed'):
	return pickle.loads(
	gzip.decompress(self._analysis_compressed)
	)
	return None

---

3.10 core/sgf_parser.py - SGF棋谱解析

文件位置: core/sgf_parser.py
文件大小: 26249 字节 (713 行)
功能: 解析和生成SGF格式棋谱

3.10.1 Move类 - 棋步

	class Move:
	"""
	棋步类
	坐标系统：
	- GTP: A1-T19（列字母+行数字）
	- SGF: aa-ss（双字母，左上角为aa）
	- 内部: (x, y) 元组，x=列，y=行
	"""
	def __init__(self, coords: Optional[Tuple[int, int]], player: str):
	"""
	参数：
	- coords: (x, y) 或 None（pass）
	- player: 'B' 或 'W'
	"""
	self.coords = coords
	self.player = player
	def gtp(self) -> str:
	"""转换为GTP坐标：如 "Q16" """
	if not self.coords:
	return "pass"
	x, y = self.coords
	col = chr(ord('A') + x)
	row = 19 - y
	return f"{col}{row}"
	def sgf(self) -> str:
	"""转换为SGF坐标：如 "pd" """
	if not self.coords:
	return ""
	x, y = self.coords
	col = chr(ord('a') + x)
	row = chr(ord('a') + y)
	return f"{col}{row}"
	@staticmethod
	def from_gtp(gtp: str, player: str) -> 'Move':
	"""从GTP坐标创建"""
	if gtp.lower() == 'pass':
	return Move(None, player)
	col = gtp[0].upper()
	row = int(gtp[1:])
	x = ord(col) - ord('A')
	y = 19 - row
	return Move((x, y), player)
	@staticmethod
	def from_sgf(sgf: str, player: str) -> 'Move':
	"""从SGF坐标创建"""
	if not sgf:
	return Move(None, player)
	x = ord(sgf[0]) - ord('a')
	y = ord(sgf[1]) - ord('a')
	return Move((x, y), player)

3.10.2 SGF类 - SGF文件解析

	class SGF:
	"""
	SGF文件解析
	SGF属性：
	- GM: 游戏类型（1=围棋）
	- FF: 文件格式（4）
	- SZ: 棋盘大小（19）
	- KM: 贴目（7.5）
	- RU: 规则（chinese）
	- PB/PW: 黑/白方姓名
	- DT: 日期
	- RE: 结果（如 "B+3.5"）
	"""
	@staticmethod
	def parse(sgf_content: str) -> GameNode:
	"""
	解析SGF文件
	流程：
	1. 解析SGF属性（如GM、FF、SZ、KM等）
	2. 构建游戏树（支持分支）
	3. 返回根节点（GameNode）
	"""
	# 解析属性
	properties = SGF._parse_properties(sgf_content)
	# 创建根节点
	root = GameNode()
	root.properties = properties
	# 解析着法
	current = root
	SGF._parse_moves(sgf_content, current)
	return root
	@staticmethod
	def generate(root: GameNode) -> str:
	"""
	生成SGF文件
	返回：SGF格式字符串
	"""
	sgf = "(;"
	# 添加属性
	for key, value in root.properties.items():
	sgf += f"{key}[{value}]"
	# 添加着法
	sgf += SGF._generate_moves(root)
	sgf += ")"
	return sgf

---

3.11 core/goai.py - AI封装接口

文件位置: core/goai.py
文件大小: 7230 字节 (173 行)
功能: 简化AI调用接口

3.11.1 GoAI类

	class GoAI:
	"""
	围棋AI封装
	主要方法：
	- get_best_move(): 获取最佳着法
	"""
	def __init__(self, strategy: str = 'default'):
	self.strategy = strategy
	def get_best_move(self, game: Game, color: int) -> Optional[Move]:
	"""
	获取最佳着法
	流程：
	1. 优先提子逻辑
	- 检查是否能提掉对方≥3子的棋块
	- 如果能，直接提子（避免错失提子机会）
	2. 调用KataGo分析
	- 获取当前节点的分析结果
	- 提取候选着法列表
	3. 合法落子检查
	- 过滤掉非法着法（如自杀、劫）
	- 返回最佳合法着法
	参数：
	- game: 游戏对象
	- color: 颜色（1=黑，2=白）
	返回：Move对象，失败返回None
	"""
	# 提子优先
	capture_move = self._check_capture_opportunity(game, color)
	if capture_move:
	return capture_move
	# 调用策略
	strategy_func = STRATEGIES[self.strategy]
	return strategy_func(game)
	def _check_capture_opportunity(
	self,
	game: Game,
	color: int
	) -> Optional[Move]:
	"""
	检查提子机会
	如果能提掉对方≥3子的棋块，直接提子
	"""
	# ...（详细实现见源码）
	pass

---

3.12 core/wechat_pay.py - 微信支付集成

文件位置: core/wechat_pay.py
文件大小: 8538 字节 (273 行)
功能: 微信支付V3 API封装

3.12.1 核心功能

	class WeChatPay:
	"""
	微信支付V3
	主要功能：
	1. Native支付（扫码支付）
	2. 订单管理
	3. 支付回调
	"""
	def __init__(self, app_id, mch_id, api_key):
	self.app_id = app_id
	self.mch_id = mch_id
	self.api_key = api_key
	def create_order(
	self,
	out_trade_no: str,
	total_amount: int,
	description: str
	) -> str:
	"""
	创建订单
	参数：
	- out_trade_no: 商户订单号
	- total_amount: 金额（分）
	- description: 商品描述
	返回：支付二维码链接
	"""
	# 构建请求
	# ...
	# 返回code_url
	return code_url
	def query_order(self, out_trade_no: str) -> dict:
	"""
	查询订单状态
	返回：
	{
	'trade_state': 'SUCCESS', # SUCCESS/NOTPAY/CLOSED
	'transaction_id': '...',
	...
	}
	"""
	pass
	def close_order(self, out_trade_no: str):
	"""关闭订单"""
	pass
	def verify_callback(self, headers: dict, body: str) -> dict:
	"""
	验证支付回调
	安全措施：
	- 使用微信平台证书验证签名
	- 敏感数据加密传输（AES-256-GCM）
	- 订单幂等性处理
	返回：解密后的回调数据
	"""
	# 验证签名
	# ...
	# 解密数据
	# ...
	return decrypted_data

---

3.13 其他核心模块

3.13.1 core/constants.py - 全局常量

	# AI策略类型
	AI_DEFAULT = 'default'
	AI_RANK = 'rank'
	AI_HUMAN = 'human'
	AI_PRO = 'pro'
	AI_WEIGHTED = 'weighted'
	# ... 更多策略
	# 输出级别
	OUTPUT_RAW = 'raw'
	OUTPUT_ANALYSIS = 'analysis'
	OUTPUT_OWNERSHIP = 'ownership'
	# 优先级
	PRIORITY_HIGH = 10
	PRIORITY_NORMAL = 5
	PRIORITY_LOW = 1
	# 游戏模式
	MODE_PLAY = 'play'
	MODE_ANALYZE = 'analyze'
	MODE_SELFPLAY = 'selfplay'

3.13.2 core/utils.py - 工具函数

	def coord_to_gtp(x: int, y: int) -> str:
	"""坐标转GTP格式"""
	col = chr(ord('A') + x)
	row = 19 - y
	return f"{col}{row}"
	def weighted_choice(items: List, weights: List[float]) -> Any:
	"""加权随机选择"""
	total = sum(weights)
	r = random.random() * total
	cumsum = 0
	for item, weight in zip(items, weights):
	cumsum += weight
	if r <= cumsum:
	return item
	return items[-1]
	def truncate_json_array(data: dict, key: str, max_items: int):
	"""截断JSON数组"""
	if key in data and isinstance(data[key], list):
	data[key] = data[key][:max_items]

3.13.3 core/email_config.py - 邮件服务

	import smtplib
	from email.mime.text import MIMEText
	def send_email(to: str, subject: str, body: str):
	"""
	发送邮件
	用于：
	- 密码重置
	- VIP到期提醒
	"""
	msg = MIMEText(body)
	msg['Subject'] = subject
	msg['From'] = SMTP_USER
	msg['To'] = to
	with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
	server.starttls()
	server.login(SMTP_USER, SMTP_PASSWORD)
	server.send_message(msg)

3.13.4 core/bj_time.py - 北京时间

	from datetime import datetime, timezone, timedelta
	def get_beijing_time() -> datetime:
	"""获取当前北京时间"""
	utc_now = datetime.utcnow()
	beijing_tz = timezone(timedelta(hours=8))
	return utc_now.replace(tzinfo=beijing_tz)

3.13.5 core/katabase.py - KataGo配置管理

	class KatagoBase:
	"""
	KataGo基础配置
	主要属性：
	- size: 棋盘大小
	- komi: 贴目
	- rules: 规则集
	- max_visits: 最大访问次数
	"""
	def __init__(self, config_path: str = "data/config.json"):
	self.config = self._load_config(config_path)
	self.size = self.config.get('size', 19)
	self.komi = self.config.get('komi', 7.5)
	self.rules = self.config.get('rules', 'chinese')
	self.max_visits = self.config.get('max_visits', 1000)
	class SimpleJsonStore:
	"""轻量级JSON存储器"""
	def __init__(self, filepath: str):
	self.filepath = filepath
	self.data = self._load()
	def save(self):
	"""保存到文件"""
	with open(self.filepath, 'w') as f:
	json.dump(self.data, f, indent=2)

---

四、路由模块详解

4.1 routers/ws.py - WebSocket路由 ⭐核心

文件位置: routers/ws.py
文件大小: 31361 字节 (755 行)
功能: WebSocket路由和实时对弈处理

4.1.1 端点定义

	@router.get("/")
	async def get_home():
	"""主页路由"""
	return FileResponse('static/index.html')
	@router.websocket("/ws/{room_id}")
	async def websocket_endpoint(
	websocket: WebSocket,
	room_id: str,
	token: str = Query(...)
	):
	"""
	WebSocket连接端点
	参数：
	- room_id: 房间ID
	- token: JWT认证token
	流程：
	1. 验证token
	2. 接受WebSocket连接
	3. 加入房间
	4. 循环处理消息
	5. 处理断线
	"""
	# 验证token
	username = decode_token(token)
	if not username:
	await websocket.close(code=4001)
	return
	# 接受连接
	await websocket.accept()
	# 加入房间
	success, color, error = await manager.join_room(
	room_id, websocket, username
	)
	if not success:
	await websocket.send_json({'type': 'error', 'message': error})
	await websocket.close()
	return
	try:
	# 消息循环
	while True:
	data = await websocket.receive_json()
	await handle_websocket_message(room_id, username, data, websocket)
	except WebSocketDisconnect:
	# 处理断线
	await manager.handle_disconnect(room_id, color, username)

4.1.2 消息类型

类型	方向	数据	说明
move	客户端→服务器		落子
pass	客户端→服务器		虚手
undo	客户端→服务器	{}	悔棋
resign	客户端→服务器		认输
territory_request	客户端→服务器	{}	请求点目
state_update	服务器→客户端		状态更新
game_over	服务器→客户端		游戏结束
error	服务器→客户端		错误消息

4.1.3 消息处理

	async def handle_websocket_message(
	room_id: str,
	username: str,
	data: dict,
	websocket: WebSocket
	):
	"""
	处理WebSocket消息
	消息类型：
	- move: 落子
	- pass: 虚手
	- undo: 悔棋
	- resign: 认输
	- territory_request: 点目请求
	"""
	msg_type = data.get('type')
	game = manager.games.get(room_id)
	if msg_type == 'move':
	# 验证轮次
	color = data.get('color')
	if not _is_player_turn(game, color):
	await websocket.send_json({
	'type': 'error',
	'message': '不是你的回合'
	})
	return
	# 执行落子
	x, y = data['x'], data['y']
	success = game.place_stone(x, y)
	if success:
	# 广播状态更新
	await manager.broadcast_state(room_id)
	# 检查游戏是否结束
	if game.game_over:
	await manager.handle_game_over(room_id)
	# AI落子
	if game.ai_enabled and not game.game_over:
	await manager.schedule_ai_move(room_id)
	else:
	await websocket.send_json({
	'type': 'error',
	'message': '非法落子'
	})
	elif msg_type == 'pass':
	# 执行虚手
	game.pass_turn()
	# 检查是否双pass（游戏结束）
	if game.consecutive_passes >= 2:
	await manager.handle_game_over(room_id)
	else:
	await manager.broadcast_state(room_id)
	elif msg_type == 'undo':
	# 执行悔棋
	success = game.undo_move()
	if success:
	await manager.broadcast_state(room_id)
	else:
	await websocket.send_json({
	'type': 'error',
	'message': '无法悔棋'
	})
	elif msg_type == 'resign':
	# 处理认输
	winner = 'W' if data['color'] == 'black' else 'B'
	await manager.handle_game_over(room_id, winner=winner, reason='resign')
	elif msg_type == 'territory_request':
	# 执行点目
	result = game.calculate_final_score()
	await websocket.send_json({
	'type': 'territory_result',
	'data': result
	})

4.1.4 积分计算

	def calculate_score_change(
	winner_username: str,
	loser_username: str
	) -> Tuple[int, int]:
	"""
	计算积分变化
	规则：
	- 同段位：胜方+1，负方-1
	- 不同段位：
	- 高段位胜：不加积分
	- 高段位负：-1
	- 低段位胜：+1
	- 低段位负：不减积分
	- AI固定为9段，不计算积分
	- 10段积分到100后：胜不加，负减1
	返回：(胜方积分变化, 负方积分变化)
	"""
	winner = get_user(winner_username)
	loser = get_user(loser_username)
	# AI不计算积分
	if winner_username.startswith('AI_') or loser_username.startswith('AI_'):
	return 0, 0
	winner_rank = winner['rank']
	loser_rank = loser['rank']
	if winner_rank == loser_rank:
	# 同段位
	return 1, -1
	elif winner_rank > loser_rank:
	# 高段位胜
	if winner['score'] >= 100 and winner_rank == 10:
	return 0, -1
	return 0, -1
	else:
	# 低段位胜
	return 1, 0

---

4.2 routers/game.py - 游戏路由

文件位置: routers/game.py
文件大小: 13261 字节 (362 行)
功能: 游戏相关路由

4.2.1 端点列表

方法	路径	功能	权限
GET	/api/online-users	获取在线用户	登录
GET	/api/generate-room	生成房间ID	登录
GET	/api/user-status	用户对局状态	登录
POST	/api/invite	发送对弈邀请	登录
POST	/api/accept-invite	接受邀请	登录
POST	/api/reject-invite	拒绝邀请	登录
GET	/api/games/history	棋谱历史	登录
GET	/api/games/{id}/sgf	下载SGF	登录

4.2.2 核心实现

	@router.get("/api/online-users")
	async def get_online_users(token: str = Query(...)):
	"""获取在线用户列表"""
	username = decode_token(token)
	if not username:
	raise HTTPException(status_code=401, detail="未登录")
	users = manager.get_online_users()
	return {"users": users}
	@router.post("/api/invite")
	async def send_invite(
	data: InviteRequest,
	token: str = Query(...)
	):
	"""
	发送对弈邀请
	流程：
	1. 验证token
	2. 检查目标用户是否在线
	3. 发送邀请通知（WebSocket）
	"""
	username = decode_token(token)
	if not username:
	raise HTTPException(status_code=401, detail="未登录")
	# 检查目标用户
	if data.to not in manager.online_users:
	raise HTTPException(status_code=404, detail="用户不在线")
	# 发送邀请
	websocket = manager.online_users[data.to]
	await websocket.send_json({
	'type': 'invite',
	'from': username,
	'room_id': data.room_id
	})
	return {"message": "邀请已发送"}
	@router.get("/api/games/history")
	async def get_game_history(token: str = Query(...)):
	"""获取棋谱历史"""
	username = decode_token(token)
	if not username:
	raise HTTPException(status_code=401, detail="未登录")
	games = get_user_games(username)
	return {"games": games}
	@router.get("/api/games/{game_id}/sgf")
	async def download_sgf(game_id: int, token: str = Query(...)):
	"""下载SGF棋谱"""
	username = decode_token(token)
	if not username:
	raise HTTPException(status_code=401, detail="未登录")
	sgf_content = get_game_sgf(game_id)
	if not sgf_content:
	raise HTTPException(status_code=404, detail="棋谱不存在")
	return Response(
	content=sgf_content,
	media_type="application/x-go-sgf",
	headers={
	"Content-Disposition": f"attachment; filename=game_{game_id}.sgf"
	}
	)

---

4.3 routers/auth.py - 认证路由

文件位置: routers/auth.py
文件大小: 3307 字节 (107 行)
功能: 认证相关路由

4.3.1 端点列表

方法	路径	功能	权限
POST	/api/register	用户注册	公开
POST	/api/login	用户登录	公开
POST	/api/change-password	修改密码	登录
POST	/api/update-email	更新邮箱	登录
POST	/api/request-reset-password	请求重置密码	公开
POST	/api/reset-password	重置密码	公开

4.3.2 核心实现

	@router.post("/api/register")
	async def register(data: RegisterRequest):
	"""
	用户注册
	请求：
	{
	"username": "player1",
	"password": "secure123",
	"email": "player1@example.com"
	}
	响应：
	{
	"message": "注册成功",
	"username": "player1"
	}
	"""
	success = register_user(data.username, data.password, data.email)
	if not success:
	raise HTTPException(status_code=400, detail="用户名已存在")
	return {"message": "注册成功", "username": data.username}
	@router.post("/api/login")
	async def login(data: LoginRequest):
	"""
	用户登录
	请求：
	{
	"username": "player1",
	"password": "secure123"
	}
	响应：
	{
	"access_token": "eyJ...",
	"token_type": "bearer",
	"username": "player1",
	"role": "user",
	"is_vip": false
	}
	"""
	token = login_user(data.username, data.password)
	if not token:
	raise HTTPException(status_code=401, detail="用户名或密码错误")
	user = get_user(data.username)
	return {
	"access_token": token,
	"token_type": "bearer",
	"username": data.username,
	"role": user['role'],
	"is_vip": is_vip(data.username)
	}

---

4.4 routers/admin.py - 管理员路由

文件位置: routers/admin.py
文件大小: 3268 字节 (98 行)
功能: 管理员相关路由

4.4.1 端点列表

方法	路径	功能	权限
GET	/api/admin/users	用户列表	管理员
PUT	/api/admin/users/{username}/role	修改角色	管理员
DELETE	/api/admin/users/{username}	删除用户	管理员
GET	/api/admin/games	棋谱列表	管理员
DELETE	/api/admin/games/{id}	删除棋谱	管理员
POST	/api/admin/init	初始化管理员	公开

---

4.5 routers/vip.py - VIP路由

文件位置: routers/vip.py
文件大小: 6017 字节 (171 行)
功能: VIP相关路由

4.5.1 端点列表

方法	路径	功能	权限
GET	/api/vip/info	VIP信息	登录
POST	/api/vip/create-order	创建订单	登录
GET	/api/vip/order-status	订单状态	登录
POST	/api/vip/notify	支付回调	微信
GET	/api/vip/settings	功能开关	管理员

---

4.6 routers/doc.py - 文档路由

文件位置: routers/doc.py
文件大小: 3900 字节 (105 行)
功能: 文档管理路由

4.6.1 端点列表

方法	路径	功能	权限
POST	/api/doc/markdown/upload	上传文档	登录
GET	/api/doc/markdown/files	文档列表	登录
GET	/api/doc/markdown/{id}	获取文档	登录
PUT	/api/doc/markdown/{id}	更新文档	所有者
DELETE	/api/doc/markdown/{id}	删除文档	管理员
POST	/api/doc/markdown/{id}/approve	审核文档	管理员

4.6.2 核心实现

	@router.post("/api/doc/markdown/upload")
	async def upload_markdown(data: DocMarkdownUpload, token: str):
	"""
	上传Markdown文档
	请求：
	{
	"filename": "tutorial.md",
	"content": "# 教程\n\n..."
	}
	"""
	username = check_doc_permission(token)
	file_id = upload_doc_markdown(data.filename, data.content, username)
	return {"message": "上传成功", "file_id": file_id}
	@router.put("/api/doc/markdown/{file_id}")
	async def update_markdown(file_id: int, data: DocMarkdownUpdate, token: str):
	"""
	更新Markdown文档
	权限：仅文档所有者可编辑
	"""
	username = check_doc_permission(token)
	file_data = get_doc_markdown_content(file_id)
	if not file_data:
	raise HTTPException(status_code=404, detail="文件不存在")
	# 权限检查
	if file_data.get("uploaded_by") != username:
	raise HTTPException(status_code=403, detail="只有文件所有者可以编辑")
	success = update_doc_markdown(file_id, data.content)
	if not success:
	raise HTTPException(status_code=500, detail="更新失败")
	return {"message": "更新成功"}

---

4.7 routers/deps.py - 路由依赖项

文件位置: routers/deps.py
文件大小: 1278 字节 (36 行)
功能: 路由依赖项

	_manager: ConnectionManager = None
	def init_manager(manager: ConnectionManager):
	"""初始化连接管理器"""
	global _manager
	_manager = manager
	def get_manager() -> ConnectionManager:
	"""获取连接管理器"""
	return _manager
	async def get_current_user(token: str = Query(...)) -> str:
	"""获取当前用户"""
	username = decode_token(token)
	if not username:
	raise HTTPException(status_code=401, detail="未登录")
	return username
	async def require_admin(token: str = Query(...)) -> str:
	"""要求管理员权限"""
	username = decode_token(token)
	if not username:
	raise HTTPException(status_code=401, detail="未登录")
	if not is_admin(username):
	raise HTTPException(status_code=403, detail="需要管理员权限")
	return username

---

五、前端实现

5.1 static/script.js - 前端主逻辑 ⭐核心

文件位置: static/script.js
文件大小: 194157 字节 (4371 行)
功能: 前端主逻辑脚本

5.1.1 核心模块

1. WebSocket连接管理

	let ws = null;
	let reconnectTimer = null;
	function connectWebSocket(roomId) {
	/**
	* 连接WebSocket
	*
	* 流程：
	* 1. 创建WebSocket连接
	* 2. 设置事件处理器
	* 3. 启动心跳检测
	*/
	const wsUrl = `ws://${window.location.host}/ws/${roomId}?token=${authToken}`;
	ws = new WebSocket(wsUrl);
	ws.onopen = function() {
	console.log('WebSocket连接成功');
	startHeartbeat();
	};
	ws.onmessage = function(event) {
	const data = JSON.parse(event.data);
	handleMessage(data);
	};
	ws.onclose = function() {
	console.log('WebSocket断开');
	scheduleReconnect();
	};
	ws.onerror = function(error) {
	console.error('WebSocket错误:', error);
	};
	}
	function startHeartbeat() {
	/**
	* 心跳检测
	* 每30秒发送一次ping
	*/
	setInterval(() => {
	if (ws && ws.readyState === WebSocket.OPEN) {
	ws.send(JSON.stringify({type: 'ping'}));
	}
	}, 30000);
	}
	function scheduleReconnect() {
	/**
	* 重连机制
	* 5秒后尝试重连
	*/
	if (reconnectTimer) {
	clearTimeout(reconnectTimer);
	}
	reconnectTimer = setTimeout(() => {
	console.log('尝试重连...');
	connectWebSocket(currentRoomId);
	}, 5000);
	}

2. 棋盘渲染

	const canvas = document.getElementById('board');
	const ctx = canvas.getContext('2d');
	function renderBoard(gameState) {
	/**
	* 渲染棋盘
	*
	* 流程：
	* 1. 绘制棋盘网格
	* 2. 绘制星位
	* 3. 绘制棋子
	* 4. 绘制手数标记
	* 5. 绘制领地标记
	*/
	// 清空画布
	ctx.clearRect(0, 0, canvas.width, canvas.height);
	// 绘制网格
	drawGrid();
	// 绘制星位
	drawStarPoints();
	// 绘制棋子
	for (let y = 0; y < 19; y++) {
	for (let x = 0; x < 19; x++) {
	if (gameState.board[y][x] !== 0) {
	drawStone(x, y, gameState.board[y][x]);
	}
	}
	}
	// 绘制手数标记
	if (showMoveNumbers) {
	drawMoveNumbers(gameState.move_numbers);
	}
	// 绘制领地标记
	if (gameState.territory_board) {
	drawTerritory(gameState.territory_board);
	}
	}
	function drawStone(x, y, color) {
	/**
	* 绘制棋子
	*
	* 参数：
	* - x, y: 棋盘坐标
	* - color: 1=黑，2=白
	*/
	const px = x * CELL_SIZE + OFFSET;
	const py = y * CELL_SIZE + OFFSET;
	ctx.beginPath();
	ctx.arc(px, py, STONE_RADIUS, 0, 2 * Math.PI);
	if (color === 1) {
	// 黑子
	ctx.fillStyle = '#000';
	} else {
	// 白子
	ctx.fillStyle = '#fff';
	ctx.strokeStyle = '#000';
	ctx.lineWidth = 1;
	}
	ctx.fill();
	ctx.stroke();
	}

3. 游戏逻辑

	function handleClick(event) {
	/**
	* 处理点击事件
	*
	* 流程：
	* 1. 计算棋盘坐标
	* 2. 检查是否轮到自己
	* 3. 发送落子消息
	*/
	const rect = canvas.getBoundingClientRect();
	const x = Math.round((event.clientX - rect.left - OFFSET) / CELL_SIZE);
	const y = Math.round((event.clientY - rect.top - OFFSET) / CELL_SIZE);
	// 检查坐标合法性
	if (x < 0 || x >= 19 || y < 0 || y >= 19) {
	return;
	}
	// 检查是否轮到自己
	if (!isMyTurn()) {
	return;
	}
	// 发送落子消息
	ws.send(JSON.stringify({
	type: 'move',
	x: x,
	y: y,
	color: myColor
	}));
	}
	function handleUndo() {
	/**
	* 处理悔棋
	*/
	ws.send(JSON.stringify({
	type: 'undo'
	}));
	}
	function handleResign() {
	/**
	* 处理认输
	*/
	if (confirm('确定要认输吗？')) {
	ws.send(JSON.stringify({
	type: 'resign',
	color: myColor
	}));
	}
	}
	function requestTerritory() {
	/**
	* 请求点目
	*/
	ws.send(JSON.stringify({
	type: 'territory_request'
	}));
	}

4. UI交互

	function handleMessage(data) {
	/**
	* 处理WebSocket消息
	*
	* 消息类型：
	* - state_update: 状态更新
	* - game_over: 游戏结束
	* - error: 错误消息
	* - invite: 对弈邀请
	* - territory_result: 点目结果
	*/
	switch (data.type) {
	case 'state_update':
	updateGameState(data.state);
	break;
	case 'game_over':
	showGameOver(data.winner, data.score);
	break;
	case 'error':
	showError(data.message);
	break;
	case 'invite':
	showInvite(data.from, data.room_id);
	break;
	case 'territory_result':
	showTerritoryResult(data.data);
	break;
	}
	}
	function updateGameState(state) {
	/**
	* 更新游戏状态
	*
	* 流程：
	* 1. 更新棋盘
	* 2. 更新信息显示
	* 3. 重新渲染
	*/
	gameState = state;
	renderBoard(state);
	updateInfo(state);
	}

5. 国际化支持

	const i18n = {
	'zh-CN': {
	'login': '登录',
	'register': '注册',
	'play': '对弈',
	'undo': '悔棋',
	'resign': '认输',
	'pass': '虚手',
	// ... 更多翻译
	},
	'en-US': {
	'login': 'Login',
	'register': 'Register',
	'play': 'Play',
	'undo': 'Undo',
	'resign': 'Resign',
	'pass': 'Pass',
	// ... 更多翻译
	}
	};
	let currentLang = 'zh-CN';
	function t(key) {
	/**
	* 翻译函数
	*
	* 用法：t('login') -> '登录'
	*/
	return i18n[currentLang][key] || key;
	}
	function switchLanguage(lang) {
	/**
	* 切换语言
	*/
	currentLang = lang;
	updateUI();
	}

6. Markdown渲染

	function renderMarkdown(content) {
	/**
	* 渲染Markdown
	*
	* 使用marked.js库
	*/
	const html = marked.parse(content);
	document.getElementById('docContent').innerHTML = html;
	// 代码高亮
	document.querySelectorAll('pre code').forEach(block => {
	hljs.highlightElement(block);
	});
	}

---

5.2 static/style.css - 样式表

文件位置: static/style.css
文件大小: 33983 字节 (1668 行)
功能: 样式表

5.2.1 核心样式

	/* 棋盘样式 */
	#board {
	border: 2px solid #000;
	background: #DCB35C;
	cursor: pointer;
	}
	/* 棋子样式 */
	.stone {
	border-radius: 50%;
	box-shadow: 2px 2px 4px rgba(0,0,0,0.3);
	}
	.stone.black {
	background: #000;
	}
	.stone.white {
	background: #fff;
	border: 1px solid #000;
	}
	/* 响应式布局 */
	@media (max-width: 768px) {
	#board {
	width: 100%;
	height: auto;
	}
	.sidebar {
	display: none;
	}
	}
	/* 动画效果 */
	@keyframes fadeIn {
	from { opacity: 0; }
	to { opacity: 1; }
	}
	.fade-in {
	animation: fadeIn 0.3s ease-in;
	}

---

5.3 static/views/ - 页面视图模板

5.3.1 modals.html - 模态框模板

	<!-- 修改密码模态框 -->
	<div id="changePasswordModal" class="modal">
	<div class="modal-content">
	<h2>修改密码</h2>
	<form onsubmit="handleChangePassword(event)">
	<input type="password" id="oldPassword" placeholder="旧密码" required>
	<input type="password" id="newPassword" placeholder="新密码" required>
	<button type="submit">确认</button>
	</form>
	</div>
	</div>
	<!-- VIP购买模态框 -->
	<div id="vipModal" class="modal">
	<div class="modal-content">
	<h2>VIP会员</h2>
	<div class="vip-plans">
	<div class="plan" onclick="buyVIP('monthly')">
	<h3>月度会员</h3>
	<p>¥30/月</p>
	</div>
	<div class="plan" onclick="buyVIP('yearly')">
	<h3>年度会员</h3>
	<p>¥300/年</p>
	</div>
	</div>
	</div>
	</div>

5.3.2 game.html - 对弈页面

	<div id="gameView" class="view-container">
	<!-- 棋盘区域 -->
	<div class="board-container">
	<canvas id="board" width="570" height="570"></canvas>
	</div>
	<!-- 信息面板 -->
	<div class="info-panel">
	<div class="player-info">
	<div class="black">
	<span class="name">黑方</span>
	<span class="captures">提子: 0</span>
	</div>
	<div class="white">
	<span class="name">白方</span>
	<span class="captures">提子: 0</span>
	</div>
	</div>
	<!-- 控制按钮 -->
	<div class="controls">
	<button onclick="handleUndo()">悔棋</button>
	<button onclick="handlePass()">虚手</button>
	<button onclick="handleResign()">认输</button>
	<button onclick="requestTerritory()">点目</button>
	</div>
	</div>
	</div>

5.3.3 doc.html - 文档页面

	<div id="docView" class="view-container">
	<!-- 文档工具栏 -->
	<div id="docToolbar">
	<button id="docEditBtn" onclick="editDocContent()">✏️ 编辑</button>
	<button id="docSaveBtn" onclick="saveDocContent()">💾 保存</button>
	<button id="docDownloadBtn" onclick="downloadDocContent()">📥 下载</button>
	<button id="docPrintBtn" onclick="printDocContent()">🖨️ 打印</button>
	</div>
	<!-- 文档内容 -->
	<div id="docMarkdownContent">
	<!-- Markdown渲染区域 -->
	</div>
	<!-- 文件列表 -->
	<div id="docFileList">
	<!-- 文件列表 -->
	</div>
	</div>

---

六、配置文件

6.1 data/config.json - 主配置文件

文件位置: data/config.json
文件大小: 6572 字节 (251 行)
功能: 主配置文件

	{
	"game": {
	"size": 19,
	"komi": 7.5,
	"rules": "chinese",
	"handicap": 0
	},
	"engine": {
	"katago_path": "data/katago",
	"model_path": "data/b18c384nbt-s*.bin.gz",
	"config_path": "data/analysis_config.cfg",
	"max_visits": 1000,
	"fast_visits": 100,
	"time_limit": 10.0
	},
	"ai": {
	"default_strategy": "default",
	"rank_strategy": "rank",
	"human_strategy": "human"
	},
	"ui": {
	"show_move_numbers": false,
	"show_coordinates": true,
	"sound_enabled": true
	},
	"timer": {
	"move_timeout": 60,
	"max_timeout_count": 3
	}
	}

6.2 data/analysis_config.cfg - KataGo分析配置

文件位置: data/analysis_config.cfg
文件大小: 13353 字节
功能: KataGo分析引擎配置

	# 日志配置
	logToStdout = true
	logAllRequests = false
	# 搜索参数
	maxVisits = 1000
	maxTime = 10.0
	# 规则设置
	rules = chinese
	komi = 7.5
	# 性能优化
	numSearchThreads = 4
	maxConcurrentEvals = 8

---

七、关键技术特性

7.1 实时对弈

WebSocket通信:

• 低延迟实时通信
• 双向消息推送
• 自动重连机制

断线重连:

• 3分钟内自动重连
• 最多允许2次断线
• 第3次断线直接判负

7.2 AI引擎集成

KataGo引擎:

• 强大的围棋AI引擎
• 支持多种规则集
• 提供精确的形势判断

AI策略:

• 16种不同策略
• 段位匹配
• 人类风格模拟

7.3 精确点目

Benson算法:

• 识别无条件活棋
• 比传统方法更精确
• 支持大眼活棋、借劲活棋

死子识别:

• 优先使用KataGo的ownership数据
• 回退到启发式算法
• 综合考虑多个因素

7.4 权限系统

三级权限:

• 普通用户：基本功能
• VIP用户：无限AI对弈、专属功能
• 管理员：用户管理、棋谱管理

VIP功能开关:

• AI对弈限制
• 优先匹配
• 专属功能

7.5 支付集成

微信支付V3:

• Native支付（扫码）
• 订单管理
• 支付回调

安全措施:

• 签名验证
• 数据加密
• 幂等性处理

---

八、性能优化

8.1 数据库优化

• 使用索引加速查询
• 定期清理过期数据
• 连接池管理

8.2 内存优化

• 分析数据压缩存储
• 按需解压
• 定期清理缓存

8.3 网络优化

• WebSocket长连接
• 消息压缩
• 心跳检测

---

九、安全措施

9.1 认证安全

• JWT token认证
• 密码bcrypt加密
• Token过期机制

9.2 数据安全

• SQL注入防护（参数化查询）
• XSS防护（输入过滤）
• CSRF防护（token验证）

9.3 支付安全

• 签名验证
• 数据加密
• 幂等性处理

---

十、部署说明

10.1 环境要求

• Python 3.8+
• SQLite 3
• KataGo引擎

10.2 安装步骤

	# 1. 安装依赖
	pip install -r requirements.txt
	# 2. 初始化数据库
	python -c "from core.auth import init_db; init_db()"
	# 3. 下载KataGo引擎
	从 https://github.com/lightvector/KataGo 下载
	# 4. 启动服务
	python main.py

10.3 生产部署

	# 使用uvicorn
	uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
	# 或使用gunicorn
	gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

---

十一、开发指南

11.1 代码风格

• 遵循PEP 8规范
• 使用类型注解
• 编写文档字符串

11.2 测试

	# 运行测试
	pytest tests/
	# 代码覆盖率
	pytest --cov=core tests/

11.3 调试

	# 启用调试日志
	import logging
	logging.basicConfig(level=logging.DEBUG)

---

十二、常见问题

Q1: 如何添加新的AI策略？

在 core/ai.py 中使用装饰器注册：

	@register_strategy("my_strategy")
	def ai_my_strategy(game, **kwargs):
	# 实现你的策略
	return best_move

Q2: 如何修改点目算法？

修改 core/state.py 中的 calculate_final_score() 方法。

Q3: 如何添加新的WebSocket消息类型？

在 routers/ws.py 的 handle_websocket_message() 中添加新的处理分支。

---

十三、更新日志

v1.0.0 (2026-01-01)

• 初始版本发布
• 支持人机对弈、人人对弈
• 集成KataGo引擎