# 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):随机选择候选着法pass3.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(),keylambda 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: 打谱点目最大次数3def __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房间启用AI4. 重连处理- 恢复用户颜色- 取消重连等待计时器- 替换旧的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房间启用AIif room_id.startswith(ai:):self.games[room_id].ai_enabled Trueself.games[room_id].ai_color 2 # 白方# 分配颜色if black not in self.rooms[room_id]:color blackself.rooms[room_id][black] websocketelif white not in self.rooms[room_id]:color whiteself.rooms[room_id][white] websocketelse:return False, None, 房间已满# 记录用户状态self.user_game_status[username] {room_id: room_id,color: color,opponent: self._get_opponent(room_id, color)}return True, color, None3.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) 1self._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/adminis_vip BOOLEAN DEFAULT FALSE, -- VIP状态vip_expire_at DATETIME, -- VIP到期时间reset_token TEXT, -- 密码重置tokenreset_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, -- 如 B3.5end_reason TEXT, -- 如 resign, timeoutsgf_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, yearlyamount REAL NOT NULL,status TEXT DEFAULT pending, -- pending/paid/failedtrade_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 Truedef login_user(username: str, password: str) - Optional[str]:用户登录流程1. 查询用户2. 验证密码3. 生成JWT token返回JWT token失败返回Noneuser get_user(username)if not user:return None# 验证密码if not bcrypt.checkpw(password.encode(), user[hashed_password]):return None# 生成JWT tokentoken jwt.encode({username: username,exp: datetime.utcnow() timedelta(hours24)}, SECRET_KEY, algorithmHS256)return token权限管理def is_admin(username: str) - bool:检查是否为管理员user get_user(username)return user and user[role] admindef is_vip(username: str) - bool:检查是否为VIPuser 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_atreturn Falsedef 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后胜不加负减1user 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 Noneself.query_id 0self.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,stdinsubprocess.PIPE,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,textTrue,bufsize1)# 启动结果读取线程threading.Thread(targetself._read_results, daemonTrue).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 1query_id fquery-{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:breaktry: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})
更多局部战斗型
# 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):随机选择候选着法pass3.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(),keylambda 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: 打谱点目最大次数3def __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房间启用AI4. 重连处理- 恢复用户颜色- 取消重连等待计时器- 替换旧的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房间启用AIif room_id.startswith(ai:):self.games[room_id].ai_enabled Trueself.games[room_id].ai_color 2 # 白方# 分配颜色if black not in self.rooms[room_id]:color blackself.rooms[room_id][black] websocketelif white not in self.rooms[room_id]:color whiteself.rooms[room_id][white] websocketelse:return False, None, 房间已满# 记录用户状态self.user_game_status[username] {room_id: room_id,color: color,opponent: self._get_opponent(room_id, color)}return True, color, None3.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) 1self._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/adminis_vip BOOLEAN DEFAULT FALSE, -- VIP状态vip_expire_at DATETIME, -- VIP到期时间reset_token TEXT, -- 密码重置tokenreset_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, -- 如 B3.5end_reason TEXT, -- 如 resign, timeoutsgf_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, yearlyamount REAL NOT NULL,status TEXT DEFAULT pending, -- pending/paid/failedtrade_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 Truedef login_user(username: str, password: str) - Optional[str]:用户登录流程1. 查询用户2. 验证密码3. 生成JWT token返回JWT token失败返回Noneuser get_user(username)if not user:return None# 验证密码if not bcrypt.checkpw(password.encode(), user[hashed_password]):return None# 生成JWT tokentoken jwt.encode({username: username,exp: datetime.utcnow() timedelta(hours24)}, SECRET_KEY, algorithmHS256)return token权限管理def is_admin(username: str) - bool:检查是否为管理员user get_user(username)return user and user[role] admindef is_vip(username: str) - bool:检查是否为VIPuser 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_atreturn Falsedef 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后胜不加负减1user 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 Noneself.query_id 0self.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,stdinsubprocess.PIPE,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,textTrue,bufsize1)# 启动结果读取线程threading.Thread(targetself._read_results, daemonTrue).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 1query_id fquery-{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:breaktry: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})
