From 5e95d1e86748031f60e4d38357cdf8a2a4b96a38 Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Thu, 25 Sep 2025 21:14:45 +0100 Subject: [PATCH 01/11] Add stats command --- api.py | 17 +++++++++++ user_interface.py | 76 ++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 92 insertions(+), 1 deletion(-) diff --git a/api.py b/api.py index d67473f5..5246414a 100644 --- a/api.py +++ b/api.py @@ -2,6 +2,7 @@ import json import logging import time +from collections.abc import AsyncGenerator from typing import Any import aiohttp @@ -174,6 +175,22 @@ async def get_account(self) -> dict[str, Any]: raise RuntimeError(f"Account error: {json_response['error']}") return json_response + async def get_user_games(self, username: str, since: int | None = None, until: int | None = None, max_games: int | None = None) -> AsyncGenerator[dict[str, Any], None]: + params = {} + if since is not None: + params['since'] = since + if until is not None: + params['until'] = until + if max_games is not None: + params['max'] = max_games + params['finished'] = 'true' + async with self.lichess_session.get(f"/api/games/user/{username}", params=params, headers={'Accept': 'application/x-ndjson'}) as response: + response.raise_for_status() + async for line in response.content: + line = line.decode('utf-8').strip() + if line: + yield json.loads(line) + async def get_chessdb_eval(self, fen: str, best_move: bool, timeout: int) -> dict[str, Any] | None: try: async with self.external_session.get( diff --git a/user_interface.py b/user_interface.py index 5ea9e538..e63fa335 100755 --- a/user_interface.py +++ b/user_interface.py @@ -4,13 +4,15 @@ import os import signal import sys +from datetime import datetime, timezone, timedelta +from engine import Engine from enum import StrEnum from typing import TypeVar from api import API from botli_dataclasses import Challenge_Request from config import Config -from engine import Engine +from collections import defaultdict from enums import Challenge_Color, Perf_Type, Variant from event_handler import Event_Handler from game_manager import Game_Manager @@ -33,6 +35,7 @@ "quit": "Exits the bot.", "rechallenge": "Challenges the opponent to the last received challenge.", "reset": "Resets matchmaking. Usage: reset PERF_TYPE", + "stats": "Shows bot statistics and performance information.", "stop": "Stops matchmaking mode.", "tournament": "Joins tournament. Usage: tournament ID [TEAM_ID] [PASSWORD]", "whitelist": "Temporarily whitelists a user. Use config for permanent whitelisting. Usage: whitelist USERNAME", @@ -158,6 +161,8 @@ async def _handle_command(self, command: list[str]) -> None: self._rechallenge() case "reset": self._reset(command) + case "stats": + await self._stats() case "stop" | "s": self._stop() case "tournament" | "t": @@ -286,6 +291,75 @@ def _reset(self, command: list[str]) -> None: self.game_manager.matchmaking.opponents.reset_release_time(perf_type) print("Matchmaking has been reset.") + async def _stats(self) -> None: + """Display bot statistics and performance information.""" + print("\n=== Bot Statistics ===") + + # Account information + try: + account = await self.api.get_account() + print(f"👤 Username: {account['username']}") + + # Rating information + perfs = account.get('perfs', {}) + if perfs: + print("\nRatings:") + for perf_type, perf_data in perfs.items(): + if 'rating' in perf_data: + rating = perf_data['rating'] + provisional = "?" if perf_data.get('provisional', False) else "" + print(f" {perf_type.title()}: {rating}{provisional}") + except Exception as e: + print(f"Could not retrieve account info: {e}") + + # Total games today + start_of_day = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0) + end_of_day = start_of_day + timedelta(days=1) + since = int(start_of_day.timestamp() * 1000) + until = int(end_of_day.timestamp() * 1000) + total_games_today = 0 + rating_changes = defaultdict(int) + try: + async for game in self.api.get_user_games(account['username'], since=since, until=until): + total_games_today += 1 + user_id = account['id'] + if game['players']['white']['user']['id'] == user_id: + player = game['players']['white'] + elif game['players']['black']['user']['id'] == user_id: + player = game['players']['black'] + else: + continue + if 'ratingDiff' in player: + if game['variant'] == 'standard': + key = game['speed'] + else: + key = game['variant'] + rating_changes[key] += player['ratingDiff'] + except Exception as e: + print(f"Could not retrieve games today: {e}") + total_games_today = "N/A" + print(f"\nTotal Games Today: {total_games_today}") + if rating_changes and total_games_today != "N/A": + print("\nToday's Rating Changes:") + for key, diff in rating_changes.items(): + sign = "+" if diff > 0 else "" + print(f" {key.title()}: {sign}{diff}") + + # System info + import psutil + import time + + # Memory usage + process = psutil.Process() + memory_mb = process.memory_info().rss / 1024 / 1024 + print(f"\nMemory Usage: {memory_mb:.1f} MB") + + # CPU usage + cpu_percent = process.cpu_percent(interval=0.1) + print(f"CPU Usage: {cpu_percent:.1f}%") + + print("=" * 25) + def _stop(self) -> None: if self.game_manager.stop_matchmaking(): print("Stopping matchmaking ...") From 5ff8844e9e97a6594885cd1a0fe2715a342c6c0b Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Fri, 26 Sep 2025 18:39:44 +0100 Subject: [PATCH 02/11] remove unused command --- user_interface.py | 35 +++++++++++++++++++++++------------ 1 file changed, 23 insertions(+), 12 deletions(-) diff --git a/user_interface.py b/user_interface.py index e63fa335..670add83 100755 --- a/user_interface.py +++ b/user_interface.py @@ -8,6 +8,7 @@ from engine import Engine from enum import StrEnum from typing import TypeVar +import psutil from api import API from botli_dataclasses import Challenge_Request @@ -318,36 +319,46 @@ async def _stats(self) -> None: since = int(start_of_day.timestamp() * 1000) until = int(end_of_day.timestamp() * 1000) total_games_today = 0 - rating_changes = defaultdict(int) + rating_changes = defaultdict(lambda: {'diff': 0, 'wins': 0, 'losses': 0, 'draws': 0}) try: async for game in self.api.get_user_games(account['username'], since=since, until=until): total_games_today += 1 user_id = account['id'] if game['players']['white']['user']['id'] == user_id: player = game['players']['white'] + bot_color = 'white' elif game['players']['black']['user']['id'] == user_id: player = game['players']['black'] + bot_color = 'black' else: continue + if game['variant'] == 'standard': + key = game['speed'] + else: + key = game['variant'] if 'ratingDiff' in player: - if game['variant'] == 'standard': - key = game['speed'] - else: - key = game['variant'] - rating_changes[key] += player['ratingDiff'] + rating_changes[key]['diff'] += player['ratingDiff'] + # Determine result + winner = game.get('winner') + if winner == bot_color: + rating_changes[key]['wins'] += 1 + elif winner and winner != bot_color: + rating_changes[key]['losses'] += 1 + elif game.get('status') == 'draw' or not winner: + rating_changes[key]['draws'] += 1 except Exception as e: print(f"Could not retrieve games today: {e}") total_games_today = "N/A" print(f"\nTotal Games Today: {total_games_today}") if rating_changes and total_games_today != "N/A": print("\nToday's Rating Changes:") - for key, diff in rating_changes.items(): + for key, data in rating_changes.items(): + diff = data['diff'] + wins = data['wins'] + losses = data['losses'] + draws = data['draws'] sign = "+" if diff > 0 else "" - print(f" {key.title()}: {sign}{diff}") - - # System info - import psutil - import time + print(f" {key.title()}: {sign}{diff} (W: {wins} L: {losses} D: {draws})") # Memory usage process = psutil.Process() From 9d1cfd6da8fc2482cc77fe375979ade1b0cd9e70 Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Fri, 26 Sep 2025 18:51:56 +0100 Subject: [PATCH 03/11] Lint code with ruff and fix pyright errors --- api.py | 17 ++++++++---- user_interface.py | 67 ++++++++++++++++++++++++----------------------- 2 files changed, 46 insertions(+), 38 deletions(-) diff --git a/api.py b/api.py index 5246414a..87c800b0 100644 --- a/api.py +++ b/api.py @@ -175,7 +175,10 @@ async def get_account(self) -> dict[str, Any]: raise RuntimeError(f"Account error: {json_response['error']}") return json_response - async def get_user_games(self, username: str, since: int | None = None, until: int | None = None, max_games: int | None = None) -> AsyncGenerator[dict[str, Any], None]: + async def get_user_games(self, username: str, + since: int | None = None, + until: int | None = None, + max_games: int | None = None) -> AsyncGenerator[dict[str, Any], None]: params = {} if since is not None: params['since'] = since @@ -184,12 +187,16 @@ async def get_user_games(self, username: str, since: int | None = None, until: i if max_games is not None: params['max'] = max_games params['finished'] = 'true' - async with self.lichess_session.get(f"/api/games/user/{username}", params=params, headers={'Accept': 'application/x-ndjson'}) as response: + async with self.lichess_session.get( + f"/api/games/user/{username}", + params=params, + headers={'Accept': 'application/x-ndjson'} + ) as response: response.raise_for_status() async for line in response.content: - line = line.decode('utf-8').strip() - if line: - yield json.loads(line) + decoded_line = line.decode('utf-8').strip() + if decoded_line: + yield json.loads(decoded_line) async def get_chessdb_eval(self, fen: str, best_move: bool, timeout: int) -> dict[str, Any] | None: try: diff --git a/user_interface.py b/user_interface.py index 670add83..ce828641 100755 --- a/user_interface.py +++ b/user_interface.py @@ -4,16 +4,17 @@ import os import signal import sys -from datetime import datetime, timezone, timedelta -from engine import Engine +from collections import defaultdict +from datetime import UTC, datetime, timedelta from enum import StrEnum from typing import TypeVar + import psutil from api import API from botli_dataclasses import Challenge_Request from config import Config -from collections import defaultdict +from engine import Engine from enums import Challenge_Color, Perf_Type, Variant from event_handler import Event_Handler from game_manager import Game_Manager @@ -296,7 +297,7 @@ async def _stats(self) -> None: """Display bot statistics and performance information.""" print("\n=== Bot Statistics ===") - # Account information + account = None try: account = await self.api.get_account() print(f"👤 Username: {account['username']}") @@ -314,40 +315,40 @@ async def _stats(self) -> None: print(f"Could not retrieve account info: {e}") # Total games today - start_of_day = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0) + start_of_day = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0) end_of_day = start_of_day + timedelta(days=1) since = int(start_of_day.timestamp() * 1000) until = int(end_of_day.timestamp() * 1000) total_games_today = 0 rating_changes = defaultdict(lambda: {'diff': 0, 'wins': 0, 'losses': 0, 'draws': 0}) - try: - async for game in self.api.get_user_games(account['username'], since=since, until=until): - total_games_today += 1 - user_id = account['id'] - if game['players']['white']['user']['id'] == user_id: - player = game['players']['white'] - bot_color = 'white' - elif game['players']['black']['user']['id'] == user_id: - player = game['players']['black'] - bot_color = 'black' - else: - continue - if game['variant'] == 'standard': - key = game['speed'] - else: - key = game['variant'] - if 'ratingDiff' in player: - rating_changes[key]['diff'] += player['ratingDiff'] - # Determine result - winner = game.get('winner') - if winner == bot_color: - rating_changes[key]['wins'] += 1 - elif winner and winner != bot_color: - rating_changes[key]['losses'] += 1 - elif game.get('status') == 'draw' or not winner: - rating_changes[key]['draws'] += 1 - except Exception as e: - print(f"Could not retrieve games today: {e}") + if account: + try: + async for game in self.api.get_user_games(account['username'], since=since, until=until): + total_games_today += 1 + user_id = account['id'] + if game['players']['white']['user']['id'] == user_id: + player = game['players']['white'] + bot_color = 'white' + elif game['players']['black']['user']['id'] == user_id: + player = game['players']['black'] + bot_color = 'black' + else: + continue + key = game['speed'] if game['variant'] == 'standard' else game['variant'] + if 'ratingDiff' in player: + rating_changes[key]['diff'] += player['ratingDiff'] + # Determine result + winner = game.get('winner') + if winner == bot_color: + rating_changes[key]['wins'] += 1 + elif winner and winner != bot_color: + rating_changes[key]['losses'] += 1 + elif game.get('status') == 'draw' or not winner: + rating_changes[key]['draws'] += 1 + except Exception as e: + print(f"Could not retrieve games today: {e}") + total_games_today = "N/A" + else: total_games_today = "N/A" print(f"\nTotal Games Today: {total_games_today}") if rating_changes and total_games_today != "N/A": From 5a83c343f861b5a36c36bc1929d219564c2a22d9 Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Wed, 1 Oct 2025 16:53:50 +0100 Subject: [PATCH 04/11] Add new API methods for fetching user rating history and activity --- api.py | 13 +++++++ user_interface.py | 97 +++++++++++++++++++++++++++-------------------- 2 files changed, 69 insertions(+), 41 deletions(-) diff --git a/api.py b/api.py index 87c800b0..c07ed1f0 100644 --- a/api.py +++ b/api.py @@ -297,6 +297,12 @@ async def get_tournament_info(self, tournament_id: str) -> dict[str, Any]: async with self.lichess_session.get(f"/api/tournament/{tournament_id}") as response: return await response.json() + @retry(**JSON_RETRY_CONDITIONS) + async def get_rating_history(self, username: str) -> list[dict[str, Any]]: + async with self.lichess_session.get(f"/api/user/{username}/rating-history") as response: + response.raise_for_status() + return await response.json() + @retry(**JSON_RETRY_CONDITIONS) async def get_user_status(self, username: str) -> dict[str, Any]: async with self.lichess_session.get("/api/users/status", params={"ids": username}) as response: @@ -406,3 +412,10 @@ async def withdraw_tournament(self, tournament_id: str) -> bool: except aiohttp.ClientResponseError as e: print(e) return False + + @retry(**JSON_RETRY_CONDITIONS) + async def get_user_activity(self, username: str) -> list: + """Fetches the activity feed of a user from Lichess.""" + async with self.lichess_session.get(f"/api/user/{username}/activity") as response: + response.raise_for_status() + return await response.json() diff --git a/user_interface.py b/user_interface.py index ce828641..fe99cb70 100755 --- a/user_interface.py +++ b/user_interface.py @@ -1,11 +1,10 @@ import argparse import asyncio +import datetime import logging import os import signal import sys -from collections import defaultdict -from datetime import UTC, datetime, timedelta from enum import StrEnum from typing import TypeVar @@ -314,52 +313,68 @@ async def _stats(self) -> None: except Exception as e: print(f"Could not retrieve account info: {e}") - # Total games today - start_of_day = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0) - end_of_day = start_of_day + timedelta(days=1) - since = int(start_of_day.timestamp() * 1000) - until = int(end_of_day.timestamp() * 1000) - total_games_today = 0 - rating_changes = defaultdict(lambda: {'diff': 0, 'wins': 0, 'losses': 0, 'draws': 0}) + # Use new API functions for stats if account: try: + today = datetime.datetime.now(datetime.UTC).date() + start_of_day = datetime.datetime.combine(today, datetime.datetime.min.time(), tzinfo=datetime.UTC) + end_of_day = start_of_day + datetime.timedelta(days=1) + since = int(start_of_day.timestamp() * 1000) + until = int(end_of_day.timestamp() * 1000) + stats = {} + total_games = 0 + # Calculate rating changes for today by variant + rating_diffs = {} async for game in self.api.get_user_games(account['username'], since=since, until=until): - total_games_today += 1 + variant = game.get('variant', 'standard').title() + perf = game.get('perf', variant) user_id = account['id'] - if game['players']['white']['user']['id'] == user_id: - player = game['players']['white'] - bot_color = 'white' - elif game['players']['black']['user']['id'] == user_id: - player = game['players']['black'] - bot_color = 'black' + if perf not in stats: + stats[perf] = {'games': 0, 'wins': 0, 'draws': 0, 'losses': 0} + stats[perf]['games'] += 1 + total_games += 1 + # Rating diff + player = None + if 'players' in game: + if 'white' in game['players'] and game['players']['white']['user']['id'] == user_id: + player = game['players']['white'] + elif 'black' in game['players'] and game['players']['black']['user']['id'] == user_id: + player = game['players']['black'] + if player and 'ratingDiff' in player: + rating_diffs.setdefault(perf, 0) + rating_diffs[perf] += player['ratingDiff'] + # Results + result = game.get('winner') + if result: + if game['players'][result]['user']['id'] == user_id: + stats[perf]['wins'] += 1 + else: + stats[perf]['losses'] += 1 + elif game.get('status') == 'draw': + stats[perf]['draws'] += 1 + print(f"\nTotal Games Today: {total_games}") + for perf, data in stats.items(): + line = f"Played {data['games']} {perf} games" + details = [] + if data['wins']: + details.append(f"{data['wins']} win{'s' if data['wins'] != 1 else ''}") + if data['draws']: + details.append(f"{data['draws']} draw{'s' if data['draws'] != 1 else ''}") + if data['losses']: + details.append(f"{data['losses']} loss{'es' if data['losses'] != 1 else ''}") + # Add rating diff + diff = rating_diffs.get(perf, 0) + if diff > 0: + details.append(f"+{diff} rating") + elif diff < 0: + details.append(f"{diff} rating") else: - continue - key = game['speed'] if game['variant'] == 'standard' else game['variant'] - if 'ratingDiff' in player: - rating_changes[key]['diff'] += player['ratingDiff'] - # Determine result - winner = game.get('winner') - if winner == bot_color: - rating_changes[key]['wins'] += 1 - elif winner and winner != bot_color: - rating_changes[key]['losses'] += 1 - elif game.get('status') == 'draw' or not winner: - rating_changes[key]['draws'] += 1 + details.append("no rating change") + if details: + line += ": " + " ".join(details) + print(line) except Exception as e: print(f"Could not retrieve games today: {e}") - total_games_today = "N/A" - else: - total_games_today = "N/A" - print(f"\nTotal Games Today: {total_games_today}") - if rating_changes and total_games_today != "N/A": - print("\nToday's Rating Changes:") - for key, data in rating_changes.items(): - diff = data['diff'] - wins = data['wins'] - losses = data['losses'] - draws = data['draws'] - sign = "+" if diff > 0 else "" - print(f" {key.title()}: {sign}{diff} (W: {wins} L: {losses} D: {draws})") # Memory usage process = psutil.Process() From e9d09e47277642ac59f316fe41be9b48f52eed1f Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Wed, 1 Oct 2025 17:17:29 +0100 Subject: [PATCH 05/11] Remove unused methods --- api.py | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/api.py b/api.py index c07ed1f0..8f56449e 100644 --- a/api.py +++ b/api.py @@ -297,11 +297,6 @@ async def get_tournament_info(self, tournament_id: str) -> dict[str, Any]: async with self.lichess_session.get(f"/api/tournament/{tournament_id}") as response: return await response.json() - @retry(**JSON_RETRY_CONDITIONS) - async def get_rating_history(self, username: str) -> list[dict[str, Any]]: - async with self.lichess_session.get(f"/api/user/{username}/rating-history") as response: - response.raise_for_status() - return await response.json() @retry(**JSON_RETRY_CONDITIONS) async def get_user_status(self, username: str) -> dict[str, Any]: @@ -413,9 +408,3 @@ async def withdraw_tournament(self, tournament_id: str) -> bool: print(e) return False - @retry(**JSON_RETRY_CONDITIONS) - async def get_user_activity(self, username: str) -> list: - """Fetches the activity feed of a user from Lichess.""" - async with self.lichess_session.get(f"/api/user/{username}/activity") as response: - response.raise_for_status() - return await response.json() From 2ca107a00a9ace303bc50c6c32a04dbff4927790 Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Wed, 1 Oct 2025 21:18:53 +0100 Subject: [PATCH 06/11] Remove comments --- user_interface.py | 8 -------- 1 file changed, 8 deletions(-) diff --git a/user_interface.py b/user_interface.py index fe99cb70..e4650c57 100755 --- a/user_interface.py +++ b/user_interface.py @@ -301,7 +301,6 @@ async def _stats(self) -> None: account = await self.api.get_account() print(f"👤 Username: {account['username']}") - # Rating information perfs = account.get('perfs', {}) if perfs: print("\nRatings:") @@ -313,7 +312,6 @@ async def _stats(self) -> None: except Exception as e: print(f"Could not retrieve account info: {e}") - # Use new API functions for stats if account: try: today = datetime.datetime.now(datetime.UTC).date() @@ -323,7 +321,6 @@ async def _stats(self) -> None: until = int(end_of_day.timestamp() * 1000) stats = {} total_games = 0 - # Calculate rating changes for today by variant rating_diffs = {} async for game in self.api.get_user_games(account['username'], since=since, until=until): variant = game.get('variant', 'standard').title() @@ -333,7 +330,6 @@ async def _stats(self) -> None: stats[perf] = {'games': 0, 'wins': 0, 'draws': 0, 'losses': 0} stats[perf]['games'] += 1 total_games += 1 - # Rating diff player = None if 'players' in game: if 'white' in game['players'] and game['players']['white']['user']['id'] == user_id: @@ -343,7 +339,6 @@ async def _stats(self) -> None: if player and 'ratingDiff' in player: rating_diffs.setdefault(perf, 0) rating_diffs[perf] += player['ratingDiff'] - # Results result = game.get('winner') if result: if game['players'][result]['user']['id'] == user_id: @@ -362,7 +357,6 @@ async def _stats(self) -> None: details.append(f"{data['draws']} draw{'s' if data['draws'] != 1 else ''}") if data['losses']: details.append(f"{data['losses']} loss{'es' if data['losses'] != 1 else ''}") - # Add rating diff diff = rating_diffs.get(perf, 0) if diff > 0: details.append(f"+{diff} rating") @@ -376,12 +370,10 @@ async def _stats(self) -> None: except Exception as e: print(f"Could not retrieve games today: {e}") - # Memory usage process = psutil.Process() memory_mb = process.memory_info().rss / 1024 / 1024 print(f"\nMemory Usage: {memory_mb:.1f} MB") - # CPU usage cpu_percent = process.cpu_percent(interval=0.1) print(f"CPU Usage: {cpu_percent:.1f}%") From ddaf3cd56bf6bc0a6f047721a304534aebaac201 Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Fri, 3 Oct 2025 16:51:53 +0100 Subject: [PATCH 07/11] Align colons when showing stats --- user_interface.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/user_interface.py b/user_interface.py index e4650c57..2de63ee1 100755 --- a/user_interface.py +++ b/user_interface.py @@ -348,8 +348,10 @@ async def _stats(self) -> None: elif game.get('status') == 'draw': stats[perf]['draws'] += 1 print(f"\nTotal Games Today: {total_games}") + lines = [] + max_prefix_len = 0 for perf, data in stats.items(): - line = f"Played {data['games']} {perf} games" + prefix = f"Played {data['games']} {perf} games" details = [] if data['wins']: details.append(f"{data['wins']} win{'s' if data['wins'] != 1 else ''}") @@ -365,8 +367,11 @@ async def _stats(self) -> None: else: details.append("no rating change") if details: - line += ": " + " ".join(details) - print(line) + lines.append((prefix, details)) + max_prefix_len = max(max_prefix_len, len(prefix) + 2) + + for prefix, details in lines: + print(f"{prefix:<{max_prefix_len}}: {' '.join(details)}") except Exception as e: print(f"Could not retrieve games today: {e}") From a05a0bfef124e4e396c1065d58a89f85fb9e6420 Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Sun, 5 Oct 2025 09:46:45 +0100 Subject: [PATCH 08/11] Use efficient api endpoint for stats --- api.py | 7 ++- user_interface.py | 131 +++++++++++++++++++++------------------------- 2 files changed, 67 insertions(+), 71 deletions(-) diff --git a/api.py b/api.py index 8f56449e..71baae2e 100644 --- a/api.py +++ b/api.py @@ -35,7 +35,6 @@ } STREAM_TIMEOUT = aiohttp.ClientTimeout(sock_connect=5.0, sock_read=9.0) - class API: def __init__(self, config: Config) -> None: self.lichess_session = aiohttp.ClientSession( @@ -314,6 +313,12 @@ async def handle_takeback(self, game_id: str, accept: bool) -> bool: return False return True + @retry(**JSON_RETRY_CONDITIONS) + async def get_user_activity(self, username: str) -> dict[str, Any]: + """Get the user's recent activity""" + async with self.lichess_session.get(f"/api/user/{username}/activity") as response: + return await response.json() + @retry(**JSON_RETRY_CONDITIONS) async def join_team(self, team: str, password: str | None) -> bool: data = {"password": password} if password else None diff --git a/user_interface.py b/user_interface.py index 2de63ee1..bc12cb25 100755 --- a/user_interface.py +++ b/user_interface.py @@ -1,6 +1,5 @@ import argparse import asyncio -import datetime import logging import os import signal @@ -296,89 +295,81 @@ async def _stats(self) -> None: """Display bot statistics and performance information.""" print("\n=== Bot Statistics ===") - account = None try: account = await self.api.get_account() print(f"👤 Username: {account['username']}") perfs = account.get('perfs', {}) if perfs: - print("\nRatings:") + print("\nCurrent Ratings:") for perf_type, perf_data in perfs.items(): if 'rating' in perf_data: rating = perf_data['rating'] provisional = "?" if perf_data.get('provisional', False) else "" - print(f" {perf_type.title()}: {rating}{provisional}") + print(f" {perf_type.title():<15}: {rating}{provisional}") + + activity = await self.api.get_user_activity(account['username']) + if activity and isinstance(activity, list) and len(activity) > 0: + today_activity = activity[0] + + if 'games' in today_activity: + games = today_activity['games'] + print("\nToday's Games:") + + total_games = 0 + try: + for variant_data in games.values(): + if isinstance(variant_data, dict): + total_games += sum(count for count in variant_data.values() if isinstance(count, int)) + except Exception: + total_games = 0 + + print(f"Total Games: {total_games}") + + for variant, results in games.items(): + if not isinstance(results, dict): + continue + + wins = results.get('win', 0) if isinstance(results.get('win'), int) else 0 + losses = results.get('loss', 0) if isinstance(results.get('loss'), int) else 0 + draws = results.get('draw', 0) if isinstance(results.get('draw'), int) else 0 + variant_total = wins + losses + draws + + if variant_total > 0: + win_rate = (wins + 0.5 * draws) / variant_total * 100 if variant_total > 0 else 0 + + rating_diff = 0 + if isinstance(results.get('rp'), dict): + rp_data = results['rp'] + before = rp_data.get('before', 0) + after = rp_data.get('after', 0) + rating_diff = after - before + + results_parts = [] + if wins > 0: + results_parts.append(f"{wins} win{'s' if wins != 1 else ''}") + if draws > 0: + results_parts.append(f"{draws} draw{'s' if draws != 1 else ''}") + if losses > 0: + results_parts.append(f"{losses} loss{'es' if losses != 1 else ''}") + + results_str = ", ".join(results_parts) + + diff_str = f"+{rating_diff}" if rating_diff > 0 else str(rating_diff) + + variant_name = variant.title() + print(f"{variant_name:<15}: ({variant_total} game{'s' if variant_total != 1 else ''}: {results_str} • {win_rate:.1f}% win rate • {diff_str})") + + else: + print("\nNo games played today.") + except Exception as e: - print(f"Could not retrieve account info: {e}") - - if account: - try: - today = datetime.datetime.now(datetime.UTC).date() - start_of_day = datetime.datetime.combine(today, datetime.datetime.min.time(), tzinfo=datetime.UTC) - end_of_day = start_of_day + datetime.timedelta(days=1) - since = int(start_of_day.timestamp() * 1000) - until = int(end_of_day.timestamp() * 1000) - stats = {} - total_games = 0 - rating_diffs = {} - async for game in self.api.get_user_games(account['username'], since=since, until=until): - variant = game.get('variant', 'standard').title() - perf = game.get('perf', variant) - user_id = account['id'] - if perf not in stats: - stats[perf] = {'games': 0, 'wins': 0, 'draws': 0, 'losses': 0} - stats[perf]['games'] += 1 - total_games += 1 - player = None - if 'players' in game: - if 'white' in game['players'] and game['players']['white']['user']['id'] == user_id: - player = game['players']['white'] - elif 'black' in game['players'] and game['players']['black']['user']['id'] == user_id: - player = game['players']['black'] - if player and 'ratingDiff' in player: - rating_diffs.setdefault(perf, 0) - rating_diffs[perf] += player['ratingDiff'] - result = game.get('winner') - if result: - if game['players'][result]['user']['id'] == user_id: - stats[perf]['wins'] += 1 - else: - stats[perf]['losses'] += 1 - elif game.get('status') == 'draw': - stats[perf]['draws'] += 1 - print(f"\nTotal Games Today: {total_games}") - lines = [] - max_prefix_len = 0 - for perf, data in stats.items(): - prefix = f"Played {data['games']} {perf} games" - details = [] - if data['wins']: - details.append(f"{data['wins']} win{'s' if data['wins'] != 1 else ''}") - if data['draws']: - details.append(f"{data['draws']} draw{'s' if data['draws'] != 1 else ''}") - if data['losses']: - details.append(f"{data['losses']} loss{'es' if data['losses'] != 1 else ''}") - diff = rating_diffs.get(perf, 0) - if diff > 0: - details.append(f"+{diff} rating") - elif diff < 0: - details.append(f"{diff} rating") - else: - details.append("no rating change") - if details: - lines.append((prefix, details)) - max_prefix_len = max(max_prefix_len, len(prefix) + 2) - - for prefix, details in lines: - print(f"{prefix:<{max_prefix_len}}: {' '.join(details)}") - except Exception as e: - print(f"Could not retrieve games today: {e}") + print(f"Could not retrieve stats: {e}") + print("\nSystem Information:") process = psutil.Process() memory_mb = process.memory_info().rss / 1024 / 1024 - print(f"\nMemory Usage: {memory_mb:.1f} MB") - + print(f"Memory Usage: {memory_mb:.1f} MB") cpu_percent = process.cpu_percent(interval=0.1) print(f"CPU Usage: {cpu_percent:.1f}%") From a08ae64f6f19a368ba14d6942e047b73afe1caf7 Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Sun, 5 Oct 2025 09:53:56 +0100 Subject: [PATCH 09/11] Fix ruff errors --- user_interface.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/user_interface.py b/user_interface.py index bc12cb25..fc36ab8b 100755 --- a/user_interface.py +++ b/user_interface.py @@ -337,14 +337,14 @@ async def _stats(self) -> None: if variant_total > 0: win_rate = (wins + 0.5 * draws) / variant_total * 100 if variant_total > 0 else 0 - + rating_diff = 0 if isinstance(results.get('rp'), dict): rp_data = results['rp'] before = rp_data.get('before', 0) after = rp_data.get('after', 0) rating_diff = after - before - + results_parts = [] if wins > 0: results_parts.append(f"{wins} win{'s' if wins != 1 else ''}") @@ -352,13 +352,13 @@ async def _stats(self) -> None: results_parts.append(f"{draws} draw{'s' if draws != 1 else ''}") if losses > 0: results_parts.append(f"{losses} loss{'es' if losses != 1 else ''}") - + results_str = ", ".join(results_parts) - diff_str = f"+{rating_diff}" if rating_diff > 0 else str(rating_diff) - variant_name = variant.title() - print(f"{variant_name:<15}: ({variant_total} game{'s' if variant_total != 1 else ''}: {results_str} • {win_rate:.1f}% win rate • {diff_str})") + print(f"{variant_name:<15}: (" + f"{variant_total} game{'s' if variant_total != 1 else ''}: " + f"{results_str} • {win_rate:.1f}% win rate • {diff_str})") else: print("\nNo games played today.") From 603617de5e24124af61316814839315cd5c592d1 Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Sun, 5 Oct 2025 10:10:35 +0100 Subject: [PATCH 10/11] remove unused function --- api.py | 23 ----------------------- 1 file changed, 23 deletions(-) diff --git a/api.py b/api.py index 71baae2e..42906713 100644 --- a/api.py +++ b/api.py @@ -174,29 +174,6 @@ async def get_account(self) -> dict[str, Any]: raise RuntimeError(f"Account error: {json_response['error']}") return json_response - async def get_user_games(self, username: str, - since: int | None = None, - until: int | None = None, - max_games: int | None = None) -> AsyncGenerator[dict[str, Any], None]: - params = {} - if since is not None: - params['since'] = since - if until is not None: - params['until'] = until - if max_games is not None: - params['max'] = max_games - params['finished'] = 'true' - async with self.lichess_session.get( - f"/api/games/user/{username}", - params=params, - headers={'Accept': 'application/x-ndjson'} - ) as response: - response.raise_for_status() - async for line in response.content: - decoded_line = line.decode('utf-8').strip() - if decoded_line: - yield json.loads(decoded_line) - async def get_chessdb_eval(self, fen: str, best_move: bool, timeout: int) -> dict[str, Any] | None: try: async with self.external_session.get( From f11238173673c9b07ef7f1f111af63952df1d9dc Mon Sep 17 00:00:00 2001 From: mike-koala-bear <55577@pm.me> Date: Sun, 5 Oct 2025 10:13:46 +0100 Subject: [PATCH 11/11] remove unused import --- api.py | 1 - 1 file changed, 1 deletion(-) diff --git a/api.py b/api.py index 42906713..25d3deb8 100644 --- a/api.py +++ b/api.py @@ -2,7 +2,6 @@ import json import logging import time -from collections.abc import AsyncGenerator from typing import Any import aiohttp