重构: aiohttp 全异步 + chessdb 云库 + uv 管理

xiangqi_bot/cloud.py

@@ -0,0 +1,88 @@
+# -*- coding: utf-8 -*-
+"""chessdb.cn 云库客户端(aiohttp 异步, 带缓存)。"""
+import asyncio
+import logging
+import time
+import urllib.parse
+
+import aiohttp
+
+log = logging.getLogger('xiangqi.cloud')
+
+API_URL = 'https://www.chessdb.cn/chessdb.php'
+CACHE_TTL = 7 * 24 * 3600  # 已评局面基本不变, 缓存一周
+
+
+class CloudBook:
+    """chessdb 云库查询: 返回 (move, score_from_mover_perspective) 或 None。"""
+
+    def __init__(self, timeout=2.5, min_score=15, cache_size=4096):
+        self.min_score = min_score   # 云库着法采纳的最低分(红方视角, 厘兵)
+        self._cache = {}             # fen -> (expire_ts, result|None)
+        self._cache_size = cache_size
+        self._sem = asyncio.Semaphore(2)  # 礼貌并发限制
+        self.timeout = aiohttp.ClientTimeout(total=timeout)
+        self._session = None
+        self.ok_count = 0
+        self.miss_count = 0
+
+    async def _get_session(self):
+        if self._session is None or self._session.closed:
+            self._session = aiohttp.ClientSession(timeout=self.timeout)
+        return self._session
+
+    async def close(self):
+        if self._session and not self._session.closed:
+            await self._session.close()
+
+    async def query_all(self, fen):
+        """返回 [(move, score_red_perspective)]；云库无局面/出错返回 []。"""
+        now = time.time()
+        hit = self._cache.get(fen)
+        if hit and hit[0] > now:
+            return hit[1]
+        url = f'{API_URL}?action=queryall&board={urllib.parse.quote(fen)}&learn=1'
+        try:
+            async with self._sem:
+                session = await self._get_session()
+                async with session.get(url) as resp:
+                    text = (await resp.text()).strip()
+        except Exception as e:
+            log.warning('cloud request failed: %s', e)
+            return []
+        result = []
+        if text.startswith('move:'):
+            for item in text.split('|'):
+                fields = dict(kv.split(':', 1) for kv in item.split(',') if ':' in kv)
+                try:
+                    result.append((fields['move'], int(fields.get('score', '-30000'))))
+                except (KeyError, ValueError):
+                    continue
+            self.ok_count += 1
+        else:
+            # unknown / nobestmove / checkmate / stalemate / invalid board
+            self.miss_count += 1
+        if len(self._cache) >= self._cache_size:
+            self._cache.clear()
+        self._cache[fen] = (now + CACHE_TTL, result)
+        return result
+
+    async def best(self, fen, moves_count=0):
+        """返回 (move, score_mover, depth_note)。
+
+        云库分数为红方视角; 轮黑时取负。分数不足 min_score 时放弃(交给引擎)。
+        moves_count 用于开局阶段放宽采纳阈值(前 8 回合云库几乎必赢)。
+        """
+        entries = await self.query_all(fen)
+        if not entries:
+            return None
+        side_is_red = (fen.split()[1] == 'w')
+        sign = 1 if side_is_red else -1
+        scored = [(mv, sign * sc) for mv, sc in entries]
+        scored.sort(key=lambda x: -x[1])
+        mv, sc = scored[0]
+        # 开局阈值低(云库碾压引擎), 中后局要求云库分足够好
+        threshold = self.min_score if moves_count >= 16 else 0
+        if sc < threshold:
+            return None
+        return mv, sc, f'{len(scored)}候选'