87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
# -*- 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, any_score=False):
|
|
"""返回 (move, score_mover, note)。
|
|
|
|
实测 chessdb 的 score 就是行棋方视角(与皮卡鱼 UCI 口径一致), 不需翻转。
|
|
any_score=True(全程云库): 无条件采纳云库最佳着法。
|
|
否则: 开局(≤16步)直接采纳, 中后局要求分数 ≥ min_score。
|
|
"""
|
|
entries = await self.query_all(fen)
|
|
if not entries:
|
|
return None
|
|
scored = sorted(entries, key=lambda x: -x[1])
|
|
mv, sc = scored[0]
|
|
threshold = (-10 ** 9) if any_score else (
|
|
self.min_score if moves_count >= 16 else 0)
|
|
if sc < threshold:
|
|
return None
|
|
return mv, sc, f'{len(scored)}候选'
|