mirror of
https://github.com/kuhyx/testsAndMisc-archive.git
synced 2026-07-04 19:03:11 +02:00
647 lines
20 KiB
Python
647 lines
20 KiB
Python
"""Main entry point for the Lichess bot."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
from dataclasses import dataclass, field
|
|
import datetime
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from typing import TYPE_CHECKING
|
|
|
|
import chess
|
|
import chess.pgn
|
|
import requests
|
|
|
|
from python_pkg.lichess_bot.engine import RandomEngine
|
|
from python_pkg.lichess_bot.lichess_api import LichessAPI
|
|
from python_pkg.lichess_bot.utils import backoff_sleep, get_and_increment_version
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Iterator
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
# Regex for parsing ply numbers from analysis output
|
|
_PLY_LINE_RE = re.compile(r"^\s*(\d+)\s")
|
|
|
|
# Game end statuses
|
|
_GAME_END_STATUSES = frozenset({"mate", "resign", "stalemate", "timeout", "draw"})
|
|
|
|
|
|
@dataclass
|
|
class GameMeta:
|
|
"""Metadata for a game (for PGN headers and logging)."""
|
|
|
|
game_id: str
|
|
bot_version: int
|
|
site_url: str | None = None
|
|
date_iso: str | None = None
|
|
white_name: str | None = None
|
|
black_name: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class GameState:
|
|
"""Mutable state for an ongoing game."""
|
|
|
|
board: chess.Board = field(default_factory=chess.Board)
|
|
color: str | None = None
|
|
last_handled_len: int = -1
|
|
my_ms: int | None = None
|
|
opp_ms: int | None = None
|
|
inc_ms: int = 0
|
|
log_path: Path | None = None
|
|
|
|
|
|
@dataclass
|
|
class BotContext:
|
|
"""Shared context for bot operations."""
|
|
|
|
api: LichessAPI
|
|
engine: RandomEngine
|
|
bot_version: int
|
|
decline_correspondence: bool = False
|
|
|
|
|
|
def _apply_move_to_board(board: chess.Board, move: str, game_id: str) -> None:
|
|
"""Apply a single move to the board, logging errors."""
|
|
try:
|
|
board.push_uci(move)
|
|
except ValueError:
|
|
_logger.debug(f"Game {game_id}: could not apply move {move}")
|
|
|
|
|
|
def _init_game_log(game_id: str, bot_version: int) -> Path | None:
|
|
"""Initialize the game log file."""
|
|
game_log_path = Path.cwd() / f"lichess_bot_game_{game_id}.log"
|
|
try:
|
|
with open(game_log_path, "w") as lf:
|
|
lf.write(f"game {game_id} started\n")
|
|
lf.write(f"bot_version v{bot_version}\n")
|
|
except OSError:
|
|
return None
|
|
return game_log_path
|
|
|
|
|
|
def _extract_game_full_data(
|
|
event: dict[str, object],
|
|
state: GameState,
|
|
meta: GameMeta,
|
|
api: LichessAPI,
|
|
) -> tuple[str, str | None]:
|
|
"""Extract data from a gameFull event.
|
|
|
|
Returns:
|
|
Tuple of (moves_string, status).
|
|
"""
|
|
state_data = event.get("state", {})
|
|
if not isinstance(state_data, dict):
|
|
state_data = {}
|
|
moves = str(state_data.get("moves", ""))
|
|
status = state_data.get("status")
|
|
|
|
# Update clocks - values are int milliseconds from API
|
|
wtime = state_data.get("wtime")
|
|
btime = state_data.get("btime")
|
|
if state.color == "white":
|
|
state.my_ms = int(wtime) if isinstance(wtime, int | float) else None
|
|
state.opp_ms = int(btime) if isinstance(btime, int | float) else None
|
|
else:
|
|
state.my_ms = int(btime) if isinstance(btime, int | float) else None
|
|
state.opp_ms = int(wtime) if isinstance(wtime, int | float) else None
|
|
inc = state_data.get("winc") or state_data.get("binc")
|
|
state.inc_ms = int(inc) if isinstance(inc, int | float) else 0
|
|
|
|
# Extract player info
|
|
white_data = event.get("white", {})
|
|
black_data = event.get("black", {})
|
|
if isinstance(white_data, dict) and isinstance(black_data, dict):
|
|
white_id = white_data.get("id")
|
|
black_id = black_data.get("id")
|
|
meta.white_name = str(white_data.get("name") or white_id or "?")
|
|
meta.black_name = str(black_data.get("name") or black_id or "?")
|
|
|
|
# Determine color
|
|
me = api.get_my_user_id()
|
|
if me == white_id:
|
|
state.color = "white"
|
|
elif me == black_id:
|
|
state.color = "black"
|
|
|
|
# Extract date
|
|
with contextlib.suppress(Exception):
|
|
created_ms = event.get("createdAt") or event.get("createdAtDate")
|
|
if created_ms is not None:
|
|
meta.date_iso = datetime.datetime.fromtimestamp(
|
|
int(str(created_ms)) / 1000,
|
|
tz=datetime.timezone.utc,
|
|
).strftime("%Y.%m.%d")
|
|
|
|
meta.site_url = f"https://lichess.org/{meta.game_id}"
|
|
|
|
return moves, str(status) if status else None
|
|
|
|
|
|
def _extract_game_state_data(
|
|
event: dict[str, object], state: GameState
|
|
) -> tuple[str, str | None]:
|
|
"""Extract data from a gameState event.
|
|
|
|
Returns:
|
|
Tuple of (moves_string, status).
|
|
"""
|
|
moves = str(event.get("moves", ""))
|
|
status = event.get("status")
|
|
|
|
# Update clocks based on color
|
|
if state.color == "white":
|
|
state.my_ms = event.get("wtime", state.my_ms) # type: ignore[assignment]
|
|
state.opp_ms = event.get("btime", state.opp_ms) # type: ignore[assignment]
|
|
state.inc_ms = event.get("winc", state.inc_ms) # type: ignore[assignment]
|
|
elif state.color == "black":
|
|
state.my_ms = event.get("btime", state.my_ms) # type: ignore[assignment]
|
|
state.opp_ms = event.get("wtime", state.opp_ms) # type: ignore[assignment]
|
|
state.inc_ms = event.get("binc", state.inc_ms) # type: ignore[assignment]
|
|
|
|
return moves, str(status) if status else None
|
|
|
|
|
|
def _calculate_time_budget(
|
|
state: GameState, board: chess.Board, max_time_sec: float
|
|
) -> float:
|
|
"""Calculate time budget for the next move."""
|
|
est_moves_left = max(10, min(60, 30 - board.fullmove_number // 2))
|
|
time_left_sec = (state.my_ms or 0) / 1000.0
|
|
inc_sec = (state.inc_ms or 0) / 1000.0
|
|
budget = 0.6 * (time_left_sec / max(1, est_moves_left)) + 0.5 * inc_sec
|
|
# Double the budget for more thoughtful moves
|
|
budget *= 2.0
|
|
return max(0.05, min(max_time_sec, budget))
|
|
|
|
|
|
def _log_move_to_file(
|
|
log_path: Path | None, ply: int, move: chess.Move, reason: str
|
|
) -> None:
|
|
"""Log a move to the game log file."""
|
|
if log_path:
|
|
with open(log_path, "a") as lf:
|
|
lf.write(f"ply {ply}: {move.uci()}\n{reason}\n\n")
|
|
|
|
|
|
def _attempt_move(
|
|
ctx: BotContext,
|
|
state: GameState,
|
|
meta: GameMeta,
|
|
board: chess.Board,
|
|
) -> bool:
|
|
"""Attempt to make a move. Returns True if game should continue."""
|
|
budget = _calculate_time_budget(state, board, ctx.engine.max_time_sec)
|
|
move, reason = ctx.engine.choose_move_with_explanation(
|
|
board, time_budget_sec=budget
|
|
)
|
|
|
|
if move is None:
|
|
_logger.info(f"Game {meta.game_id}: no legal moves (game likely over)")
|
|
return False
|
|
|
|
time_left_sec = (state.my_ms or 0) / 1000.0
|
|
inc_sec = (state.inc_ms or 0) / 1000.0
|
|
|
|
try:
|
|
if move not in board.legal_moves:
|
|
_logger.info(
|
|
f"Game {meta.game_id}: selected move no longer legal; skipping send"
|
|
)
|
|
else:
|
|
_logger.info(
|
|
f"Game {meta.game_id}: playing {move.uci()} "
|
|
f"(budget={budget:.2f}s, my_time_left={time_left_sec:.1f}s, "
|
|
f"inc={inc_sec:.2f}s)"
|
|
)
|
|
_log_move_to_file(state.log_path, state.last_handled_len + 1, move, reason)
|
|
ctx.api.make_move(meta.game_id, move)
|
|
except requests.RequestException as e:
|
|
_logger.warning(f"Game {meta.game_id}: move {move.uci()} failed: {e}")
|
|
|
|
return True
|
|
|
|
|
|
def _is_my_turn(board: chess.Board, color: str | None) -> bool:
|
|
"""Check if it's our turn to move."""
|
|
is_white_turn = board.turn
|
|
return (is_white_turn and color == "white") or (
|
|
(not is_white_turn) and color == "black"
|
|
)
|
|
|
|
|
|
def _rebuild_board_from_moves(moves_list: list[str], game_id: str) -> chess.Board:
|
|
"""Rebuild board from list of moves."""
|
|
board = chess.Board()
|
|
for m in moves_list:
|
|
_apply_move_to_board(board, m, game_id)
|
|
return board
|
|
|
|
|
|
def _handle_move_if_needed(
|
|
ctx: BotContext,
|
|
state: GameState,
|
|
meta: GameMeta,
|
|
et: str,
|
|
new_len: int,
|
|
) -> bool:
|
|
"""Handle making a move if it's our turn. Returns False if game ends."""
|
|
my_turn = _is_my_turn(state.board, state.color)
|
|
turn_str = "white" if state.board.turn else "black"
|
|
_logger.info(f"Game {meta.game_id}: turn={turn_str}, my_turn={my_turn}")
|
|
|
|
# Move policy
|
|
allow_move = (et == "gameState") or (et == "gameFull" and new_len == 0)
|
|
|
|
if my_turn and allow_move and not _attempt_move(ctx, state, meta, state.board):
|
|
return False
|
|
|
|
# Mark position as handled
|
|
if et == "gameState" or (my_turn and allow_move):
|
|
state.last_handled_len = new_len
|
|
|
|
return True
|
|
|
|
|
|
def _process_game_event(
|
|
event: dict[str, object],
|
|
ctx: BotContext,
|
|
state: GameState,
|
|
meta: GameMeta,
|
|
) -> bool:
|
|
"""Process a single game event. Returns False if game should end."""
|
|
et = event.get("type")
|
|
|
|
if et not in ("gameFull", "gameState"):
|
|
return True # Continue processing other events
|
|
|
|
# At this point et is guaranteed to be a string
|
|
event_type = str(et)
|
|
|
|
# Extract moves and status based on event type
|
|
if event_type == "gameFull":
|
|
moves, status = _extract_game_full_data(event, state, meta, ctx.api)
|
|
_logger.info(f"Game {meta.game_id}: joined as {state.color} (gameFull)")
|
|
else:
|
|
moves, status = _extract_game_state_data(event, state)
|
|
|
|
moves_list = moves.split() if moves else []
|
|
new_len = len(moves_list)
|
|
|
|
_logger.info(
|
|
f"Game {meta.game_id}: event={event_type}, moves={new_len}, color={state.color}"
|
|
)
|
|
|
|
if new_len == state.last_handled_len:
|
|
_logger.debug(
|
|
f"Game {meta.game_id}: position unchanged (len={new_len}), skipping"
|
|
)
|
|
return True
|
|
|
|
# Rebuild board from moves
|
|
state.board = _rebuild_board_from_moves(moves_list, meta.game_id)
|
|
|
|
if state.color is None:
|
|
_logger.info(f"Game {meta.game_id}: color unknown yet; waiting for gameFull")
|
|
if event_type == "gameState":
|
|
state.last_handled_len = new_len
|
|
return True
|
|
|
|
if not _handle_move_if_needed(ctx, state, meta, event_type, new_len):
|
|
return False
|
|
|
|
# Check for game end
|
|
if status in _GAME_END_STATUSES:
|
|
_logger.info(f"Game {meta.game_id} finished: {status}")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def _write_pgn_to_log(log_path: Path, board: chess.Board, meta: GameMeta) -> None:
|
|
"""Write PGN to the game log file."""
|
|
game = chess.pgn.Game.from_board(board)
|
|
with contextlib.suppress(Exception):
|
|
game.headers["BotVersion"] = f"v{meta.bot_version}"
|
|
if meta.site_url:
|
|
game.headers["Site"] = meta.site_url
|
|
if meta.date_iso:
|
|
game.headers["Date"] = meta.date_iso
|
|
if meta.white_name:
|
|
game.headers["White"] = meta.white_name
|
|
if meta.black_name:
|
|
game.headers["Black"] = meta.black_name
|
|
|
|
with open(log_path, "a") as lf:
|
|
lf.write("\nPGN:\n")
|
|
exporter = chess.pgn.StringExporter(
|
|
headers=True, variations=False, comments=False
|
|
)
|
|
lf.write(game.accept(exporter))
|
|
lf.write("\n")
|
|
|
|
|
|
def _run_analysis_subprocess(
|
|
game_id: str, log_path: Path, total_plies: int
|
|
) -> str | None:
|
|
"""Run the analysis script and return output text."""
|
|
analyze_script = (
|
|
Path(__file__).resolve().parent.parent
|
|
/ "stockfish_analysis"
|
|
/ "analyze_chess_game.py"
|
|
)
|
|
|
|
if not analyze_script.is_file():
|
|
_logger.info(
|
|
f"Game {game_id}: analysis script not found at {analyze_script}; "
|
|
"skipping analysis"
|
|
)
|
|
return None
|
|
|
|
_logger.info(f"Game {game_id}: starting post-game analysis ({total_plies} plies)")
|
|
|
|
proc = subprocess.Popen( # noqa: S603 - trusted internal analysis script
|
|
[sys.executable, "-u", str(analyze_script), str(log_path)],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
bufsize=1,
|
|
)
|
|
|
|
analyzed = 0
|
|
lines: list[str] = []
|
|
|
|
# stdout/stderr are guaranteed non-None with PIPE
|
|
assert proc.stdout is not None # noqa: S101
|
|
assert proc.stderr is not None # noqa: S101
|
|
|
|
for line in proc.stdout:
|
|
lines.append(line)
|
|
m = _PLY_LINE_RE.match(line)
|
|
if m:
|
|
analyzed += 1
|
|
_log_analysis_progress(game_id, analyzed, total_plies)
|
|
|
|
stderr_text = proc.stderr.read() or ""
|
|
ret = proc.wait()
|
|
analysis_text = "".join(lines)
|
|
|
|
if ret != 0:
|
|
_logger.warning(f"Game {game_id}: analysis script exited with code {ret}")
|
|
if stderr_text:
|
|
analysis_text += "\n[stderr]\n" + stderr_text
|
|
|
|
_logger.info(f"Game {game_id}: analysis complete")
|
|
return analysis_text
|
|
|
|
|
|
def _log_analysis_progress(game_id: str, analyzed: int, total_plies: int) -> None:
|
|
"""Log analysis progress."""
|
|
if total_plies:
|
|
left = max(0, total_plies - analyzed)
|
|
pct = analyzed / total_plies * 100.0
|
|
_logger.info(
|
|
f"Game {game_id}: analysis progress "
|
|
f"{analyzed}/{total_plies} ({pct:.0f}%), left {left}"
|
|
)
|
|
else:
|
|
_logger.info(
|
|
f"Game {game_id}: analysis progress {analyzed} plies (total unknown)"
|
|
)
|
|
|
|
|
|
def _insert_analysis_into_log(
|
|
log_path: Path, analysis_text: str, meta: GameMeta
|
|
) -> None:
|
|
"""Insert analysis text into the log file before PGN section."""
|
|
try:
|
|
with open(log_path, encoding="utf-8", errors="replace") as f:
|
|
content = f.read()
|
|
|
|
# Find insertion point (before PGN)
|
|
insert_idx = 0
|
|
p = content.find("\nPGN:\n")
|
|
if p != -1:
|
|
insert_idx = p + 1
|
|
elif content.startswith("PGN:\n"):
|
|
insert_idx = 0
|
|
else:
|
|
insert_idx = len(content)
|
|
|
|
# Build meta block
|
|
meta_lines = []
|
|
if meta.date_iso:
|
|
meta_lines.append(f"Date: {meta.date_iso}")
|
|
if meta.white_name or meta.black_name:
|
|
meta_lines.append(
|
|
f"Players: {meta.white_name or '?'} vs {meta.black_name or '?'}"
|
|
)
|
|
meta_block = "\n".join(meta_lines) + "\n" if meta_lines else ""
|
|
|
|
analysis_block = f"{meta_block}ANALYSIS:\n{analysis_text.rstrip()}\n\n"
|
|
new_content = content[:insert_idx] + analysis_block + content[insert_idx:]
|
|
|
|
with open(log_path, "w", encoding="utf-8") as f:
|
|
f.write(new_content)
|
|
except OSError as e:
|
|
_logger.debug(f"Game {meta.game_id}: could not write analysis to log: {e}")
|
|
|
|
|
|
def _finalize_game(state: GameState, meta: GameMeta) -> None:
|
|
"""Finalize game: write PGN and run analysis."""
|
|
if not state.log_path:
|
|
return
|
|
|
|
try:
|
|
_write_pgn_to_log(state.log_path, state.board, meta)
|
|
except OSError as e:
|
|
_logger.debug(f"Game {meta.game_id}: could not write PGN: {e}")
|
|
return
|
|
|
|
# Run analysis
|
|
try:
|
|
total_plies = len(state.board.move_stack)
|
|
except TypeError:
|
|
total_plies = 0
|
|
|
|
try:
|
|
analysis_text = _run_analysis_subprocess(
|
|
meta.game_id, state.log_path, total_plies
|
|
)
|
|
if analysis_text:
|
|
_insert_analysis_into_log(state.log_path, analysis_text, meta)
|
|
except (subprocess.SubprocessError, OSError) as e:
|
|
_logger.debug(f"Game {meta.game_id}: analysis run failed: {e}")
|
|
|
|
|
|
def _handle_game(game_id: str, ctx: BotContext, my_color: str | None = None) -> None:
|
|
"""Handle a single game from start to finish."""
|
|
_logger.info(f"Starting game thread for {game_id} [bot v{ctx.bot_version}]")
|
|
|
|
meta = GameMeta(game_id=game_id, bot_version=ctx.bot_version)
|
|
state = GameState(color=my_color)
|
|
state.log_path = _init_game_log(game_id, ctx.bot_version)
|
|
|
|
try:
|
|
for event in ctx.api.stream_game_events(game_id):
|
|
et = event.get("type")
|
|
if et in ("chatLine", "opponentGone"):
|
|
continue
|
|
if not _process_game_event(event, ctx, state, meta):
|
|
break
|
|
except requests.RequestException:
|
|
_logger.exception(f"Game {game_id} thread error")
|
|
finally:
|
|
_finalize_game(state, meta)
|
|
_logger.info(f"Ending game thread for {game_id}")
|
|
|
|
|
|
def _handle_challenge(
|
|
challenge: dict[str, object], api: LichessAPI, *, decline_correspondence: bool
|
|
) -> None:
|
|
"""Handle an incoming challenge."""
|
|
ch_id = challenge.get("id", "")
|
|
variant_data = challenge.get("variant", {})
|
|
variant = (
|
|
variant_data.get("key", "standard")
|
|
if isinstance(variant_data, dict)
|
|
else "standard"
|
|
)
|
|
speed = challenge.get("speed")
|
|
|
|
perf_ok = speed in {"bullet", "blitz", "rapid", "classical"}
|
|
not_corr = speed != "correspondence" or not decline_correspondence
|
|
|
|
if variant == "standard" and perf_ok and not_corr:
|
|
_logger.info(f"Accepting challenge {ch_id} ({speed})")
|
|
api.accept_challenge(str(ch_id))
|
|
else:
|
|
_logger.info(f"Declining challenge {ch_id} (variant={variant}, speed={speed})")
|
|
api.decline_challenge(str(ch_id))
|
|
|
|
|
|
def _process_bot_event(
|
|
event: dict[str, object],
|
|
ctx: BotContext,
|
|
game_threads: dict[str, threading.Thread],
|
|
) -> None:
|
|
"""Process a single bot event (challenge, gameStart, etc.)."""
|
|
event_type = event.get("type")
|
|
|
|
if event_type == "challenge":
|
|
challenge = event.get("challenge", {})
|
|
if isinstance(challenge, dict):
|
|
_handle_challenge(
|
|
challenge, ctx.api, decline_correspondence=ctx.decline_correspondence
|
|
)
|
|
|
|
elif event_type == "gameStart":
|
|
game_data = event.get("game", {})
|
|
if isinstance(game_data, dict):
|
|
game_id = str(game_data.get("id", ""))
|
|
if game_id and (
|
|
game_id not in game_threads or not game_threads[game_id].is_alive()
|
|
):
|
|
t = threading.Thread(
|
|
target=_handle_game,
|
|
args=(game_id, ctx),
|
|
name=f"game-{game_id}",
|
|
)
|
|
t.daemon = True
|
|
game_threads[game_id] = t
|
|
t.start()
|
|
|
|
elif event_type == "gameFinish":
|
|
game_data = event.get("game", {})
|
|
if isinstance(game_data, dict):
|
|
game_id = game_data.get("id", "")
|
|
_logger.info(f"Game finished event: {game_id}")
|
|
|
|
else:
|
|
_logger.debug(f"Unhandled event: {json.dumps(event)}")
|
|
|
|
|
|
def _stream_bot_events(ctx: BotContext) -> Iterator[dict[str, object]]:
|
|
"""Stream events from Lichess API with type hints."""
|
|
yield from ctx.api.stream_events()
|
|
|
|
|
|
def _run_event_loop_iteration(
|
|
ctx: BotContext, game_threads: dict[str, threading.Thread]
|
|
) -> int:
|
|
"""Run one iteration of the event loop.
|
|
|
|
Returns:
|
|
New backoff value (0 on success).
|
|
"""
|
|
for event in _stream_bot_events(ctx):
|
|
_process_bot_event(event, ctx, game_threads)
|
|
return 0
|
|
|
|
|
|
def run_bot(log_level: str = "INFO", *, decline_correspondence: bool = False) -> None:
|
|
"""Start the bot and listen for incoming events."""
|
|
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:
|
|
msg = "LICHESS_TOKEN environment variable is required"
|
|
raise RuntimeError(msg)
|
|
|
|
_logger.info("Token present. Initializing client and engine...")
|
|
bot_version = get_and_increment_version()
|
|
_logger.info(f"Bot version: v{bot_version}")
|
|
|
|
ctx = BotContext(
|
|
api=LichessAPI(token),
|
|
engine=RandomEngine(),
|
|
bot_version=bot_version,
|
|
decline_correspondence=decline_correspondence,
|
|
)
|
|
|
|
game_threads: dict[str, threading.Thread] = {}
|
|
|
|
_logger.info("Connecting to Lichess event stream. Waiting for challenges...")
|
|
backoff = 0
|
|
|
|
while True:
|
|
try:
|
|
backoff = _run_event_loop_iteration(ctx, game_threads)
|
|
except requests.RequestException as e: # noqa: PERF203 - intentional reconnection loop
|
|
_logger.warning(f"Event stream error: {e}")
|
|
backoff = backoff_sleep(backoff)
|
|
|
|
|
|
def main() -> None:
|
|
"""Parse arguments and run the Lichess bot."""
|
|
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, decline_correspondence=args.decline_correspondence)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|