项目概述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 文件21263.4 KBcore/ Python 文件1510,989418 KBrouters/ Python 文件81,64364 KBstatic/ 前端文件67,425393 KBstatic/views/ 模板752531 KBdata/ 配置文件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 # 主页HTML36行│ ├── 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. 应用生命周期管理asynccontextmanagerasync def lifespan(app: FastAPI):启动时创建ConnectionManager关闭时取消清理任务manager ConnectionManager()init_manager(manager)cleanup_task asyncio.create_task(session_cleanup_task(manager))logging.info([LIFESPAN] 应用启动)yieldcleanup_task.cancel()logging.info([LIFESPAN] 应用关闭)# 3. 创建应用并注册路由app FastAPI(lifespanlifespan)app.mount(/static, StaticFiles(directorystatic), namestatic)# 注册路由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# 或使用gunicorngunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker3.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(usernameadmin, roleadmin)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, selfplayai_enabled: bool # 是否启用AIai_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] stoneself.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 True3.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 或 Wmargin: 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: 不在对杀中
x01.weiqi.15: AI 对弈
项目概述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 文件21263.4 KBcore/ Python 文件1510,989418 KBrouters/ Python 文件81,64364 KBstatic/ 前端文件67,425393 KBstatic/views/ 模板752531 KBdata/ 配置文件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 # 主页HTML36行│ ├── 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. 应用生命周期管理asynccontextmanagerasync def lifespan(app: FastAPI):启动时创建ConnectionManager关闭时取消清理任务manager ConnectionManager()init_manager(manager)cleanup_task asyncio.create_task(session_cleanup_task(manager))logging.info([LIFESPAN] 应用启动)yieldcleanup_task.cancel()logging.info([LIFESPAN] 应用关闭)# 3. 创建应用并注册路由app FastAPI(lifespanlifespan)app.mount(/static, StaticFiles(directorystatic), namestatic)# 注册路由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# 或使用gunicorngunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker3.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(usernameadmin, roleadmin)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, selfplayai_enabled: bool # 是否启用AIai_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] stoneself.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 True3.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 或 Wmargin: 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: 不在对杀中
