core/game_node.py - 游戏树节点

文件大小: 18857 字节 (452 行)功能: 棋谱树结构管理3.9.1 GameNode类class GameNode:游戏树节点主要属性- parent: 父节点- children: 子节点列表- move: 本节点着法Move对象- properties: SGF属性字典- analysis: 分析结果从KataGo获取def __init__(self, parentNone, moveNone):self.parent parentself.children []self.move moveself.properties {}self.analysis Noneself.move_number 0if parent:self.move_number parent.move_number 13.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(parentself, movemove)self.children.append(new_node)return new_nodepropertydef nodes_from_root(self) - List[GameNode]:从根节点到本节点的路径返回节点列表nodes []current selfwhile current:nodes.insert(0, current)current current.parentreturn nodespropertydef nodes_in_tree(self) - List[GameNode]:树中的所有节点递归遍历所有子孙节点返回节点列表nodes [self]for child in self.children:nodes.extend(child.nodes_in_tree)return nodesdef analyze(self, engine: KataGoEngine, **kwargs):分析本节点参数- engine: KataGo引擎- kwargs: 分析参数流程1. 调用KataGo引擎2. 存储分析结果def callback(result):self.analysis resultengine.request_analysis(self, callback, **kwargs)3.9.3 分析数据压缩# 分析结果可能很大数百KB# 使用压缩存储减少内存占用import gzipimport pickledef 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 None3.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) 或 Nonepass- player: B 或 Wself.coords coordsself.player playerdef gtp(self) - str:转换为GTP坐标如 Q16 if not self.coords:return passx, y self.coordscol chr(ord(A) x)row 19 - yreturn f{col}{row}def sgf(self) - str:转换为SGF坐标如 pd if not self.coords:return x, y self.coordscol chr(ord(a) x)row chr(ord(a) y)return f{col}{row}staticmethoddef 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 - rowreturn Move((x, y), player)staticmethoddef 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: 结果如 B3.5staticmethoddef 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 rootSGF._parse_moves(sgf_content, current)return rootstaticmethoddef 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 sgf3.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 strategydef 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子的棋块直接提子# ...详细实现见源码pass3.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_idself.mch_id mch_idself.api_key api_keydef create_order(self,out_trade_no: str,total_amount: int,description: str) - str:创建订单参数- out_trade_no: 商户订单号- total_amount: 金额分- description: 商品描述返回支付二维码链接# 构建请求# ...# 返回code_urlreturn code_urldef query_order(self, out_trade_no: str) - dict:查询订单状态返回{trade_state: SUCCESS, # SUCCESS/NOTPAY/CLOSEDtransaction_id: ...,...}passdef close_order(self, out_trade_no: str):关闭订单passdef verify_callback(self, headers: dict, body: str) - dict:验证支付回调安全措施- 使用微信平台证书验证签名- 敏感数据加密传输AES-256-GCM- 订单幂等性处理返回解密后的回调数据# 验证签名# ...# 解密数据# ...return decrypted_data3.13 其他核心模块3.13.1 core/constants.py - 全局常量# AI策略类型AI_DEFAULT defaultAI_RANK rankAI_HUMAN humanAI_PRO proAI_WEIGHTED weighted# ... 更多策略# 输出级别OUTPUT_RAW rawOUTPUT_ANALYSIS analysisOUTPUT_OWNERSHIP ownership# 优先级PRIORITY_HIGH 10PRIORITY_NORMAL 5PRIORITY_LOW 1# 游戏模式MODE_PLAY playMODE_ANALYZE analyzeMODE_SELFPLAY selfplay3.13.2 core/utils.py - 工具函数def coord_to_gtp(x: int, y: int) - str:坐标转GTP格式col chr(ord(A) x)row 19 - yreturn f{col}{row}def weighted_choice(items: List, weights: List[float]) - Any:加权随机选择total sum(weights)r random.random() * totalcumsum 0for item, weight in zip(items, weights):cumsum weightif r cumsum:return itemreturn 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 smtplibfrom email.mime.text import MIMETextdef send_email(to: str, subject: str, body: str):发送邮件用于- 密码重置- VIP到期提醒msg MIMEText(body)msg[Subject] subjectmsg[From] SMTP_USERmsg[To] towith 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, timedeltadef get_beijing_time() - datetime:获取当前北京时间utc_now datetime.utcnow()beijing_tz timezone(timedelta(hours8))return utc_now.replace(tzinfobeijing_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 filepathself.data self._load()def save(self):保存到文件with open(self.filepath, w) as f:json.dump(self.data, f, indent2)四、路由模块详解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. 验证token2. 接受WebSocket连接3. 加入房间4. 循环处理消息5. 处理断线# 验证tokenusername decode_token(token)if not username:await websocket.close(code4001)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()returntry:# 消息循环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)