文件大小: 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) |
