kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-07-04 16:43:05 +02:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

fix: not able to start game as white

Browse Source
This commit is contained in:
Krzysztof kuhy Rudnicki 2025-08-22 17:04:42 +02:00
parent 5b5b069033
commit 77f66e6bb4
3 changed files with 104 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
PYTHON/lichess_bot/lichess_api.py
View File

@ -44,11 +44,11 @@ class LichessAPI:
r.raise_for_status()
def join_game_stream(self, game_id: str, my_color: Optional[str]) -> Tuple[chess.Board, str]:
# Join board stream once to detect initial state and my color
"""Deprecated: use stream_game_events and parse initial state there."""
# Fallback to initial behavior for compatibility
url = f"{LICHESS_API}/api/board/game/stream/{game_id}"
board = chess.Board()
color = my_color or "white"
with self.session.get(url, stream=True, timeout=60) as r:
r.raise_for_status()
for line in r.iter_lines(decode_unicode=True):
@ -58,10 +58,8 @@ class LichessAPI:
event = json.loads(line)
except json.JSONDecodeError:
continue
t = event.get("type")
if t == "gameFull":
# set my color
white_id = event["white"].get("id")
black_id = event["black"].get("id")
me = self.get_my_user_id()
@ -69,8 +67,6 @@ class LichessAPI:
color = "white"
elif me == black_id:
color = "black"
# Load initial state
state = event.get("state", {})
moves = state.get("moves", "")
if moves:
@ -80,11 +76,20 @@ class LichessAPI:
except Exception:
pass
break
elif t == "gameState":
# may see gameState first in rare cases; skip until gameFull
continue
return board, color
def stream_game_events(self, game_id: str) -> Generator[Dict, None, None]:
url = f"{LICHESS_API}/api/board/game/stream/{game_id}"
with self.session.get(url, stream=True, timeout=60) as r:
r.raise_for_status()
for line in r.iter_lines(decode_unicode=True):
if not line:
continue
try:
yield json.loads(line)
except json.JSONDecodeError:
logging.debug(f"Skipping non-JSON line in game {game_id}: {line}")
def make_move(self, game_id: str, move: chess.Move) -> None:
url = f"{LICHESS_API}/api/board/game/{game_id}/move/{move.uci()}"
r = self.session.post(url, timeout=30)
@ -94,23 +99,7 @@ class LichessAPI:
r.raise_for_status()
def get_game_state(self, game_id: str) -> Optional[Dict]:
url = f"{LICHESS_API}/api/board/game/stream/{game_id}"
# Use a short-lived request to read a single line update
with self.session.get(url, stream=True, timeout=10) as r:
if r.status_code >= 400:
return None
for line in r.iter_lines(decode_unicode=True):
if not line:
continue
try:
event = json.loads(line)
except json.JSONDecodeError:
continue
if event.get("type") == "gameState":
return event
if event.get("type") == "gameFull":
return event.get("state")
# If we get other events, keep looping; this request is short-lived anyway.
"""Deprecated: use stream_game_events in a persistent loop."""
return None
def get_my_user_id(self) -> Optional[str]:

98
PYTHON/lichess_bot/main.py
View File

@ -6,6 +6,8 @@ import threading
import time
from typing import Optional
import chess
from .engine import RandomEngine
from .lichess_api import LichessAPI
from .utils import backoff_sleep
@ -29,38 +31,78 @@ def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> No
def handle_game(game_id: str, my_color: Optional[str] = None):
logging.info(f"Starting game thread for {game_id}")
board = chess.Board()
color: Optional[str] = my_color
# Track how many moves we have already processed; start at -1 so we act on the first state (0 moves)
last_handled_len = -1
try:
board, color = api.join_game_stream(game_id, my_color)
logging.info(f"Game {game_id}: joined as {color}")
for event in api.stream_game_events(game_id):
et = event.get("type")
if et in ("gameFull", "gameState"):
# Determine moves list and optional status
if et == "gameFull":
state = event.get("state", {})
moves = state.get("moves", "")
status = state.get("status")
# Discover my color from gameFull
white_id = event["white"].get("id")
black_id = event["black"].get("id")
me = api.get_my_user_id()
if me == white_id:
color = "white"
elif me == black_id:
color = "black"
logging.info(f"Game {game_id}: joined as {color} (gameFull)")
else:
moves = event.get("moves", "")
status = event.get("status")
while True:
# If it's our turn, pick and play a move
if (board.turn and color == "white") or (not board.turn and color == "black"):
move = engine.choose_move(board)
if move is None:
logging.info(f"Game {game_id}: no legal moves (game likely over)")
moves_list = moves.split() if moves else []
new_len = len(moves_list)
logging.info(
f"Game {game_id}: event={et}, moves={new_len}, color={color}"
)
if new_len == last_handled_len:
logging.debug(f"Game {game_id}: position unchanged (len={new_len}), skipping")
continue
# Rebuild board from moves
board = chess.Board()
for m in moves_list:
try:
board.push_uci(m)
except Exception:
logging.debug(f"Game {game_id}: could not apply move {m}")
if color is None:
logging.info(f"Game {game_id}: color unknown yet; waiting for gameFull")
last_handled_len = new_len
continue
is_white_turn = board.turn
my_turn = (is_white_turn and color == "white") or ((not is_white_turn) and color == "black")
logging.info(
f"Game {game_id}: turn={'white' if is_white_turn else 'black'}, my_turn={my_turn}"
)
if my_turn:
move = engine.choose_move(board)
if move is None:
logging.info(f"Game {game_id}: no legal moves (game likely over)")
break
try:
logging.info(f"Game {game_id}: playing {move.uci()}")
api.make_move(game_id, move)
except Exception as e:
logging.warning(f"Game {game_id}: move {move.uci()} failed: {e}")
# Mark this position as handled (whether or not we moved)
last_handled_len = new_len
if status in {"mate", "resign", "stalemate", "timeout", "draw"}:
logging.info(f"Game {game_id} finished: {status}")
break
api.make_move(game_id, move)
# Sleep briefly to avoid hammering the API
time.sleep(0.2)
# Poll for updates to keep board in sync
updates = api.get_game_state(game_id)
if updates is None:
elif et == "chatLine":
continue
elif et == "opponentGone":
continue
# Apply last move if present
last_move_uci = updates.get("lastMove")
if last_move_uci:
try:
board.push_uci(last_move_uci)
except Exception:
# It may already be applied; ignore
pass
# Check for game end
if updates.get("status") in {"mate", "resign", "stalemate", "timeout", "draw"}:
logging.info(f"Game {game_id} finished: {updates.get('status')}")
break
except Exception as e:
logging.exception(f"Game {game_id} thread error: {e}")
finally:

19
PYTHON/lichess_bot/run.sh
View File

@ -13,6 +13,21 @@ if [[ -f "$SCRIPT_DIR/.env" ]]; then
set +a
fi
# Helper to persist the token to .env
save_token() {
local env_file="$SCRIPT_DIR/.env"
# Keep other lines, remove existing LICHESS_TOKEN, then append new one
if [[ -f "$env_file" ]]; then
grep -v '^LICHESS_TOKEN=' "$env_file" > "$env_file.tmp" || true
printf 'LICHESS_TOKEN=%s\n' "$LICHESS_TOKEN" >> "$env_file.tmp"
mv "$env_file.tmp" "$env_file"
else
printf 'LICHESS_TOKEN=%s\n' "$LICHESS_TOKEN" > "$env_file"
fi
chmod 600 "$env_file" 2>/dev/null || true
echo "Saved token to $env_file"
}
# Optional: --token <TOKEN> to set for this run
if [[ "${1:-}" == "--token" || "${1:-}" == "-t" ]]; then
if [[ "${2:-}" == "" ]]; then
@ -33,6 +48,10 @@ if [[ -z "${LICHESS_TOKEN:-}" ]]; then
exit 1
fi
echo "Token received."
save_token
else
# If token came from CLI or env, still persist so next run won't prompt
save_token
fi
# Choose python: prefer local venv