From 5b5b06903343efb3b3d3e9e3d8fac40236cd7d59 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Fri, 22 Aug 2025 16:49:30 +0200 Subject: [PATCH] feat: simple lichess bot --- .gitignore | 206 +++++++++++++++++++++++- PYTHON/lichess_bot/README.md | 74 +++++++++ PYTHON/lichess_bot/__init__.py | 1 + PYTHON/lichess_bot/engine.py | 17 ++ PYTHON/lichess_bot/lichess_api.py | 121 ++++++++++++++ PYTHON/lichess_bot/main.py | 119 ++++++++++++++ PYTHON/lichess_bot/requirements.txt | 6 + PYTHON/lichess_bot/run.sh | 51 ++++++ PYTHON/lichess_bot/tests/conftest.py | 8 + PYTHON/lichess_bot/tests/test_engine.py | 10 ++ PYTHON/lichess_bot/tests/test_utils.py | 21 +++ PYTHON/lichess_bot/utils.py | 15 ++ 12 files changed, 648 insertions(+), 1 deletion(-) create mode 100644 PYTHON/lichess_bot/README.md create mode 100644 PYTHON/lichess_bot/__init__.py create mode 100644 PYTHON/lichess_bot/engine.py create mode 100644 PYTHON/lichess_bot/lichess_api.py create mode 100644 PYTHON/lichess_bot/main.py create mode 100644 PYTHON/lichess_bot/requirements.txt create mode 100755 PYTHON/lichess_bot/run.sh create mode 100644 PYTHON/lichess_bot/tests/conftest.py create mode 100644 PYTHON/lichess_bot/tests/test_engine.py create mode 100644 PYTHON/lichess_bot/tests/test_utils.py create mode 100644 PYTHON/lichess_bot/utils.py diff --git a/.gitignore b/.gitignore index 11841bb..65179fe 100644 --- a/.gitignore +++ b/.gitignore @@ -41,4 +41,208 @@ testem.log .DS_Store Thumbs.db *.mkv -imageviewer \ No newline at end of file +imageviewer + +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[codz] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py.cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# UV +# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +#uv.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock +#poetry.toml + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python. +# https://pdm-project.org/en/latest/usage/project/#working-with-version-control +#pdm.lock +#pdm.toml +.pdm-python +.pdm-build/ + +# pixi +# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control. +#pixi.lock +# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one +# in the .venv directory. It is recommended not to include this directory in version control. +.pixi + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.envrc +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ + +# Abstra +# Abstra is an AI-powered process automation framework. +# Ignore directories containing user credentials, local state, and settings. +# Learn more at https://abstra.io/docs +.abstra/ + +# Visual Studio Code +# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore +# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore +# and can be added to the global gitignore or merged into this file. However, if you prefer, +# you could uncomment the following to ignore the entire vscode folder +# .vscode/ + +# Ruff stuff: +.ruff_cache/ + +# PyPI configuration file +.pypirc + +# Marimo +marimo/_static/ +marimo/_lsp/ +__marimo__/ + +# Streamlit +.streamlit/secrets.toml \ No newline at end of file diff --git a/PYTHON/lichess_bot/README.md b/PYTHON/lichess_bot/README.md new file mode 100644 index 0000000..56ec9e7 --- /dev/null +++ b/PYTHON/lichess_bot/README.md @@ -0,0 +1,74 @@ +# Lichess Bot (minimal) + +A small Lichess BOT that accepts standard challenges and plays quick random legal moves using python-chess. It demonstrates the Lichess Board API basics with a simple, readable implementation. + +## Features + +- Connects to Lichess Board API via streaming NDJSON +- Accepts only standard chess challenges (bullet/blitz/rapid/classical) +- Spawns a thread per active game +- Plays random legal moves (swap in a stronger engine later) +- Simple logging and basic retries on transient network errors + +## Requirements + +- Python 3.9+ +- A Lichess account that is activated as a BOT +- A Lichess API access token with at least the scopes: + - bot:play + - challenge:read + - challenge:write + +Install dependencies: + +```bash +pip install -r PYTHON/lichess_bot/requirements.txt +``` + +## Activate BOT and get a token + +1) Create or use an existing Lichess account for your bot. +2) Activate it as a BOT (one-time): https://lichess.org/api#tag/Bot + - If not already BOT, you need to convert the account; follow Lichess docs. +3) Create a personal API token: https://lichess.org/account/oauth/token/create + - Grant scopes: bot:play, challenge:read, challenge:write + +Export the token in your shell (recommended): + +```bash +export LICHESS_TOKEN="your_bot_token_here" +``` + +## Run + +From the repo root: + +```bash +python -m PYTHON.lichess_bot.main +``` + +Optional flags: + +- `--log-level INFO|DEBUG|WARNING|ERROR` (default: INFO) +- `--decline-correspondence` (declines correspondence challenges) + +You can also use the helper script: + +```bash +bash PYTHON/lichess_bot/run.sh +``` + +## Notes + +- The engine is intentionally weak (random moves). Swap it with a UCI engine or implement a better search in `engine.py`. +- Network calls hit real Lichess endpoints. Keep the bot polite; respect rate limits. + +## Development + +- Small unit tests are in `tests/` and only cover local helpers (no network). Run: + +```bash +python -m pytest PYTHON/lichess_bot/tests -q +``` + +If you add tests requiring third-party packages, install them in your environment first. \ No newline at end of file diff --git a/PYTHON/lichess_bot/__init__.py b/PYTHON/lichess_bot/__init__.py new file mode 100644 index 0000000..a9a2c5b --- /dev/null +++ b/PYTHON/lichess_bot/__init__.py @@ -0,0 +1 @@ +__all__ = [] diff --git a/PYTHON/lichess_bot/engine.py b/PYTHON/lichess_bot/engine.py new file mode 100644 index 0000000..3252f3c --- /dev/null +++ b/PYTHON/lichess_bot/engine.py @@ -0,0 +1,17 @@ +import random +from typing import Optional + +import chess + + +class RandomEngine: + """Picks a random legal move. + + You can replace this with a UCI engine wrapper or a better search. + """ + + def choose_move(self, board: chess.Board) -> Optional[chess.Move]: + moves = list(board.legal_moves) + if not moves: + return None + return random.choice(moves) diff --git a/PYTHON/lichess_bot/lichess_api.py b/PYTHON/lichess_bot/lichess_api.py new file mode 100644 index 0000000..a547d08 --- /dev/null +++ b/PYTHON/lichess_bot/lichess_api.py @@ -0,0 +1,121 @@ +import json +import logging +import time +from typing import Dict, Generator, Optional, Tuple + +import requests +import chess + + +LICHESS_API = "https://lichess.org" + + +class LichessAPI: + def __init__(self, token: str, session: Optional[requests.Session] = None): + self.token = token + self.session = session or requests.Session() + self.session.headers.update({ + "Authorization": f"Bearer {self.token}", + "Accept": "application/json", + "User-Agent": "minimal-lichess-bot/0.1 (+https://lichess.org)" + }) + + def stream_events(self) -> Generator[Dict, None, None]: + url = f"{LICHESS_API}/api/stream/event" + 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: {line}") + + def accept_challenge(self, challenge_id: str) -> None: + url = f"{LICHESS_API}/api/challenge/{challenge_id}/accept" + r = self.session.post(url, timeout=30) + r.raise_for_status() + + def decline_challenge(self, challenge_id: str, reason: str = "generic") -> None: + url = f"{LICHESS_API}/api/challenge/{challenge_id}/decline" + data = {"reason": reason} + r = self.session.post(url, data=data, timeout=30) + 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 + 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): + if not line: + continue + try: + 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() + if me == white_id: + color = "white" + elif me == black_id: + color = "black" + + # Load initial state + state = event.get("state", {}) + moves = state.get("moves", "") + if moves: + for m in moves.split(): + try: + board.push_uci(m) + except Exception: + pass + break + elif t == "gameState": + # may see gameState first in rare cases; skip until gameFull + continue + return board, color + + 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) + if r.status_code == 429: + time.sleep(0.5) + r = self.session.post(url, timeout=30) + 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. + return None + + def get_my_user_id(self) -> Optional[str]: + url = f"{LICHESS_API}/api/account" + r = self.session.get(url, timeout=30) + if r.status_code == 200: + return r.json().get("id") + return None diff --git a/PYTHON/lichess_bot/main.py b/PYTHON/lichess_bot/main.py new file mode 100644 index 0000000..525a6fe --- /dev/null +++ b/PYTHON/lichess_bot/main.py @@ -0,0 +1,119 @@ +import argparse +import json +import logging +import os +import threading +import time +from typing import Optional + +from .engine import RandomEngine +from .lichess_api import LichessAPI +from .utils import backoff_sleep + + +def run_bot(log_level: str = "INFO", decline_correspondence: bool = False) -> None: + logging.basicConfig( + level=getattr(logging, log_level.upper(), logging.INFO), + format="[%(asctime)s] %(levelname)s %(threadName)s: %(message)s", + ) + + token = os.getenv("LICHESS_TOKEN") + if not token: + raise RuntimeError("LICHESS_TOKEN environment variable is required") + + logging.info("Token present. Initializing client and engine...") + api = LichessAPI(token) + engine = RandomEngine() + + game_threads = {} + + def handle_game(game_id: str, my_color: Optional[str] = None): + logging.info(f"Starting game thread for {game_id}") + try: + board, color = api.join_game_stream(game_id, my_color) + logging.info(f"Game {game_id}: joined as {color}") + + 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)") + 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: + 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: + logging.info(f"Ending game thread for {game_id}") + + # Main event stream: challenge and game start events + logging.info("Connecting to Lichess event stream. Waiting for challenges...") + backoff = 0 + while True: + try: + for event in api.stream_events(): + if event.get("type") == "challenge": + challenge = event["challenge"] + ch_id = challenge["id"] + variant = challenge.get("variant", {}).get("key", "standard") + speed = challenge.get("speed") + perf_ok = speed in {"bullet", "blitz", "rapid", "classical"} + not_corr = challenge.get("speed") != "correspondence" or not decline_correspondence + if variant == "standard" and perf_ok and not_corr: + logging.info(f"Accepting challenge {ch_id} ({speed})") + api.accept_challenge(ch_id) + else: + logging.info(f"Declining challenge {ch_id} (variant={variant}, speed={speed})") + api.decline_challenge(ch_id) + + elif event.get("type") == "gameStart": + game_id = event["game"]["id"] + # Spin up a game thread + if game_id not in game_threads or not game_threads[game_id].is_alive(): + t = threading.Thread(target=handle_game, args=(game_id,), name=f"game-{game_id}") + t.daemon = True + game_threads[game_id] = t + t.start() + + elif event.get("type") == "gameFinish": + game_id = event["game"]["id"] + logging.info(f"Game finished event: {game_id}") + else: + logging.debug(f"Unhandled event: {json.dumps(event)}") + # If stream ends normally, reset backoff + backoff = 0 + except Exception as e: + logging.warning(f"Event stream error: {e}") + backoff = backoff_sleep(backoff) + + +def main(): + parser = argparse.ArgumentParser(description="Run a minimal Lichess bot") + parser.add_argument("--log-level", default="INFO", help="Logging level (default: INFO)") + parser.add_argument("--decline-correspondence", action="store_true", help="Decline correspondence challenges") + args = parser.parse_args() + run_bot(args.log_level, args.decline_correspondence) + + +if __name__ == "__main__": + main() diff --git a/PYTHON/lichess_bot/requirements.txt b/PYTHON/lichess_bot/requirements.txt new file mode 100644 index 0000000..87fe6a8 --- /dev/null +++ b/PYTHON/lichess_bot/requirements.txt @@ -0,0 +1,6 @@ +python-chess==1.999 +requests==2.32.3 +urllib3==2.2.3 +certifi>=2024.2.2 +chardet>=5.2.0 +idna>=3.7 diff --git a/PYTHON/lichess_bot/run.sh b/PYTHON/lichess_bot/run.sh new file mode 100755 index 0000000..7b590b2 --- /dev/null +++ b/PYTHON/lichess_bot/run.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Resolve script directory and repo root +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd -- "$SCRIPT_DIR/../.." && pwd)" + +# Load a local .env if present +if [[ -f "$SCRIPT_DIR/.env" ]]; then + set -a + # shellcheck disable=SC1090 + source "$SCRIPT_DIR/.env" + set +a +fi + +# Optional: --token to set for this run +if [[ "${1:-}" == "--token" || "${1:-}" == "-t" ]]; then + if [[ "${2:-}" == "" ]]; then + echo "--token requires a value" + exit 2 + fi + export LICHESS_TOKEN="$2" + shift 2 +fi + +# Ask for token if not set and export it for this run +if [[ -z "${LICHESS_TOKEN:-}" ]]; then + printf "Paste your Lichess API token and press Enter: " + read -r token + export LICHESS_TOKEN="$token" + if [[ -z "$LICHESS_TOKEN" ]]; then + echo "No token provided. Aborting." + exit 1 + fi + echo "Token received." +fi + +# Choose python: prefer local venv +PY="$SCRIPT_DIR/.venv/bin/python" +if [[ ! -x "$PY" ]]; then + PY="python" +fi + +cd "$REPO_ROOT" +echo "Using Python: $PY" +echo "Repository root: $REPO_ROOT" +echo "Starting Lichess bot..." +echo "Tip: Open another terminal to watch logs; press Ctrl+C here to stop." + +trap 'echo; echo "Stopping bot (Ctrl+C)."' INT +"$PY" -m PYTHON.lichess_bot.main "$@" diff --git a/PYTHON/lichess_bot/tests/conftest.py b/PYTHON/lichess_bot/tests/conftest.py new file mode 100644 index 0000000..f3309ab --- /dev/null +++ b/PYTHON/lichess_bot/tests/conftest.py @@ -0,0 +1,8 @@ +import os +import sys + +# Add repository root to sys.path so 'import PYTHON.*' works when running +# pytest with a subdirectory as rootdir. +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..")) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) diff --git a/PYTHON/lichess_bot/tests/test_engine.py b/PYTHON/lichess_bot/tests/test_engine.py new file mode 100644 index 0000000..fa0c845 --- /dev/null +++ b/PYTHON/lichess_bot/tests/test_engine.py @@ -0,0 +1,10 @@ +import chess +from PYTHON.lichess_bot.engine import RandomEngine + + +def test_random_engine_returns_move_on_start_position(): + board = chess.Board() + eng = RandomEngine() + move = eng.choose_move(board) + assert move is not None + assert move in board.legal_moves diff --git a/PYTHON/lichess_bot/tests/test_utils.py b/PYTHON/lichess_bot/tests/test_utils.py new file mode 100644 index 0000000..ed184a6 --- /dev/null +++ b/PYTHON/lichess_bot/tests/test_utils.py @@ -0,0 +1,21 @@ +import builtins + +from PYTHON.lichess_bot.utils import backoff_sleep + + +def test_backoff_sleep_increments_and_caps(monkeypatch): + slept = [] + + def fake_sleep(sec): + slept.append(sec) + + monkeypatch.setattr("time.sleep", fake_sleep) + + b = 0 + b = backoff_sleep(b, base=0.1, cap=0.3) + b = backoff_sleep(b, base=0.1, cap=0.3) + b = backoff_sleep(b, base=0.1, cap=0.3) + assert b >= 1 + assert len(slept) == 3 + # 0.1, 0.2, 0.3 (capped) + assert slept[0] == 0.1 and slept[1] == 0.2 and slept[2] == 0.3 diff --git a/PYTHON/lichess_bot/utils.py b/PYTHON/lichess_bot/utils.py new file mode 100644 index 0000000..bb4ced4 --- /dev/null +++ b/PYTHON/lichess_bot/utils.py @@ -0,0 +1,15 @@ +import logging +import time + + +def backoff_sleep(current_backoff: int, base: float = 0.5, cap: float = 8.0) -> int: + """Sleep with exponential backoff. Returns the next backoff step. + + - current_backoff: number of consecutive failures + - base: base delay in seconds + - cap: maximum delay in seconds + """ + delay = min(cap, base * (2 ** current_backoff)) + logging.info(f"Backing off for {delay:.1f}s") + time.sleep(delay) + return min(current_backoff + 1, 10)