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

xiangqi_bot/server.py

@@ -0,0 +1,120 @@
+# -*- coding: utf-8 -*-
+"""本地 HTTP 引擎服务: 云库优先, 皮卡鱼兜底。供油猴脚本调用。"""
+import argparse
+import logging
+
+from aiohttp import web
+
+from .cloud import CloudBook
+from .coords import fen_from_moves, site_to_ucci, ucci_to_site
+from .pikafish import Pikafish
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s %(name)s %(message)s')
+log = logging.getLogger('xiangqi.server')
+
+
+class Service:
+    def __init__(self, args):
+        self.args = args
+        self.cloud = CloudBook(min_score=args.cloud_min_score)
+        self.engine = None  # 惰性启动(首个请求前/启动时)
+
+    async def startup(self, app):
+        self.engine = await Pikafish(
+            threads=self.args.threads, hash_mb=self.args.hash,
+            movetime=self.args.movetime,
+            log=(lambda m: log.debug(m)) if self.args.quiet else (lambda m: log.info(m)),
+        ).start()
+
+    async def shutdown(self, app):
+        if self.engine:
+            await self.engine.quit()
+        await self.cloud.close()
+
+    async def bestmove(self, moves_site, fen=None, movetime=None, depth=None):
+        moves_ucci = [site_to_ucci(m) for m in moves_site]
+        fen = fen or fen_from_moves(moves_ucci)
+
+        # 1. 云库优先
+        try:
+            cloud_hit = await self.cloud.best(fen, moves_count=len(moves_ucci))
+        except Exception as e:
+            log.warning('cloud error: %s', e)
+            cloud_hit = None
+        if cloud_hit and not depth:
+            mv, score, note = cloud_hit
+            return {'ok': True, 'source': 'cloud', 'move': ucci_to_site(mv),
+                    'score': f'cp {score}', 'note': note, 'ms': 0}
+
+        # 2. 皮卡鱼兜底
+        try:
+            mv, score = await self.engine.bestmove(moves=moves_ucci, fen=fen,
+                                                   movetime=movetime, depth=depth)
+        except Exception:
+            log.exception('engine crashed, restarting')
+            await self.engine.restart()
+            raise
+        return {'ok': True, 'source': 'engine', 'move': ucci_to_site(mv),
+                'score': score, 'note': None}
+
+
+def make_app(args):
+    svc = Service(args)
+
+    async def health(_):
+        return web.json_response({
+            'ok': True,
+            'engine': svc.engine.path if svc.engine else None,
+            'cloud': {'min_score': svc.cloud.min_score,
+                      'hit': svc.cloud.ok_count, 'miss': svc.cloud.miss_count},
+        })
+
+    async def bestmove(request):
+        try:
+            req = await request.json()
+        except Exception:
+            return web.json_response({'ok': False, 'error': 'bad json'}, status=400)
+        moves = req.get('moves', [])
+        if not isinstance(moves, list) or not all(isinstance(m, str) for m in moves):
+            return web.json_response({'ok': False, 'error': 'moves 格式错误'}, status=400)
+        try:
+            out = await svc.bestmove(
+                moves,
+                fen=req.get('fen'),
+                movetime=req.get('movetime'),
+                depth=req.get('depth'),
+            )
+            return web.json_response(out)
+        except Exception as e:
+            return web.json_response({'ok': False, 'error': str(e)}, status=500)
+
+    app = web.Application()
+    app.router.add_get('/health', health)
+    app.router.add_post('/bestmove', bestmove)
+    app.on_startup.append(svc.startup)
+    app.on_cleanup.append(svc.shutdown)
+    return app
+
+
+def main():
+    p = argparse.ArgumentParser(description='皮卡鱼 + 云库本地服务')
+    p.add_argument('--port', type=int, default=9377)
+    p.add_argument('--movetime', type=int, default=800, help='每步思考毫秒(深度未指定时)')
+    p.add_argument('--depth', type=int, default=None, help='固定深度(优先于 movetime)')
+    p.add_argument('--threads', type=int, default=None, help='线程数(默认=CPU核心数)')
+    p.add_argument('--hash', type=int, default=256, help='哈希表 MB')
+    p.add_argument('--cloud-min-score', type=int, default=15,
+                   help='云库着法采纳阈值(厘兵, 红方视角)')
+    p.add_argument('--quiet', action='store_true', help='不打印引擎 info')
+    args = p.parse_args()
+
+    app = make_app(args)
+    mode = f'depth={args.depth}' if args.depth else f'movetime={args.movetime}ms'
+    print(f'[server] http://127.0.0.1:{args.port} ({mode}, '
+          f'threads={args.threads or "max"}, hash={args.hash}MB, '
+          f'cloud_min_score={args.cloud_min_score})', flush=True)
+    web.run_app(app, host='127.0.0.1', port=args.port, print=None)
+
+
+if __name__ == '__main__':
+    main()