3.3 KiB
Technical Design
1. SQLite → aiosqlite
\\python
room_store.py
import aiosqlite
class RoomStore: async def _get_conn(self) -> aiosqlite.Connection: db_path = Path(self.config.DB_PATH) db_path.parent.mkdir(parents=True, exist_ok=True) conn = await aiosqlite.connect(str(db_path)) conn.row_factory = aiosqlite.Row await conn.execute('''CREATE TABLE IF NOT EXISTS race_history (...)''') await conn.commit() return conn
async def get_last_horse_name(self, user_id: str) -> Optional[str]:
conn = await self._get_conn()
try:
cursor = await conn.execute("SELECT horse_name FROM ...", (user_id,))
row = await cursor.fetchone()
return row[0] if row else None
finally:
await conn.close()
\\
每次操作独立连接和关闭保持与原逻辑一致，仅将同步调用替换为 async。
2. 积分重试加延时 + 区分失败原因
\\python
points_service.py
async def spend_bet_points(self, user_id: str, amount: int, reason: str = "赛马下注"): success, balance = await points_api.spend_points(user_id, amount, "horse_race", reason) if success: return True, balance # 余额不足不再重试 if balance < amount: return False, balance # 网络/其他失败，短暂等待后重试一次 await asyncio.sleep(1.0) success, balance = await points_api.spend_points(user_id, amount, "horse_race", reason) return success, balance \\
3. 消息发送加日志
\\python
commands.py
import logging logger = logging("group_horse_racing")
async def send_to_scope(bot: Bot, scope: str, message: ...): try: if scope.startswith("group"): await bot.send_group_msg(...) elif scope.startswith("test_"): await bot.send_private_msg(...) except Exception: logger.warning(f"发送消息到 {scope} 失败", exc_info=True) \\
4. 赔率快照（结算时锁内固定）
在比赛结束结算时，将赔率在 Room Lock 内计算后传入结算函数，保证赔率与下注分布一致： \\python
在 start_race 的 lock 内计算赔率快照
odds_snapshot = {} for horse_name, horse in room.horses.items(): odds_snapshot[horse_name] = room.calculate_odds(horse_name) # 锁内计算
传入结算函数使用快照赔率
await _settle_race(bot, event, room, config, odds_snapshot) \\
5. 补存积分变化
\\python
room_store.py - save_race_result
await conn.execute( "INSERT INTO race_history (...point_changes, point_change_summaries) VALUES (..., ?, ?)", (..., json.dumps(result.point_changes), json.dumps(result.point_change_summaries)) ) \\
需在 CREATE TABLE 中新增两个 TEXT 字段，使用 \IF NOT EXISTS\ 保证兼容。
6. 测试数据隔离
\\python
test_commands.py
from .commands import get_scope, check_access, race_engine, _rooms as prod_rooms from .room_store import RoomStore as _RealRoomStore
测试命令用独立 store
_test_store = _RealRoomStore(config) # 或用内存 store \\
不再共享生产 room_store 实例，测试操作独立数据。
7. 马名去重统一
将 dict key 统一用 normalized name（casefold），或在注册时统一用原始名但比较用 casefold。