add per-game accuracy and outcome table to stats/notifications
All checks were successful
CI/CD Pipeline / build-and-deploy (push) Successful in 31s

Games-of-period responses can now include a per-game breakdown
(include_games) with Lichess post-analysis accuracy, plus which side
the tracked player was on and their rating change for that game.

- lichess_client.py requests accuracy=true from Lichess; stats_service
  computes per-mode average accuracy and builds GameRow entries
  (tracked_is_white, tracked_rating_diff) for blitz/rapid/classical
- /games/{username}/period gained an include_games query param
- formatters.py renders a column-aligned monospace table per game:
  outcome circle (win/loss/draw relative to the tracked player),
  rating change, accuracy, names/ratings, result — for today/yesterday
  and periodic notifications; week keeps an aggregate accuracy line
- usernames are Markdown-escaped before formatting since messages are
  now sent with parse_mode='Markdown'
This commit is contained in:
vrubelroman 2026-07-03 09:38:51 +00:00
parent 4a783225af
commit d479e14bf9
8 changed files with 364 additions and 75 deletions

View file

@ -1070,8 +1070,40 @@ class LichessBot:
# Only send response if there's activity
if has_activity:
formatted_response = StatsFormatter.format_stats_response(data, username, period, lang)
await update.message.reply_text(formatted_response)
# Дополнительный, параллельный запрос за реальными партиями периода —
# только чтобы добавить точность (accuracy) и, для today/yesterday,
# построчный разбор партий. Шапку (данные из /activity) не трогаем:
# при любой ошибке здесь просто не покажем точность.
accuracy_extra = None
try:
now_dt = datetime.now()
if period == "today":
since_dt = now_dt.replace(hour=0, minute=0, second=0, microsecond=0)
until_dt = now_dt
want_rows = True
elif period == "yesterday":
today_midnight = now_dt.replace(hour=0, minute=0, second=0, microsecond=0)
since_dt = today_midnight - timedelta(days=1)
until_dt = today_midnight
want_rows = True
else: # week — то же окно (today-6 .. now), что использует _is_date_in_range(days_back=7)
since_dt = now_dt.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=6)
until_dt = now_dt
want_rows = False
accuracy_extra = await self.lichess_api.get_games_period(
username,
int(since_dt.timestamp() * 1000),
int(until_dt.timestamp() * 1000),
rated_only=True,
include_games=want_rows
)
except Exception as e:
logger.warning(f"⚠️ Supplementary accuracy fetch failed for {username}/{period}: {e}")
accuracy_extra = None
formatted_response = StatsFormatter.format_stats_response(data, username, period, lang, accuracy_extra=accuracy_extra)
await update.message.reply_text(formatted_response, parse_mode='Markdown')
has_any_activity = True
else:
logger.info(f" No activity found for {username}, skipping response")
@ -1159,7 +1191,7 @@ class LichessBot:
if games_count > 0:
# Format and send immediately
text = StatsFormatter.format_last_year_or_1000(data, username, lang)
await update.message.reply_text(text)
await update.message.reply_text(text, parse_mode='Markdown')
has_any_activity = True
# Wait 3 seconds before next request (except after the last one)
@ -1585,7 +1617,8 @@ class LichessBot:
logger.info(f"📥 Adding games request to queue for {gamer['username']}")
games_data = await self.request_queue.add_request(
self.lichess_api.get_games_period,
gamer['username'], since_timestamp, until_timestamp_approx
gamer['username'], since_timestamp, until_timestamp_approx,
include_games=True
)
if games_data is None:
raise RuntimeError("Games period API returned no data")
@ -1695,7 +1728,8 @@ class LichessBot:
await self.application.bot.send_message(
chat_id=user_id,
text=notification
text=notification,
parse_mode='Markdown'
)
logger.info(f"✅ Sent periodic notification for {gamer['username']} to user {user_id}")
# Increment periodic notification counter

View file

@ -12,9 +12,98 @@ class StatsFormatter:
return f"🔴 {rating_change}"
else:
return "⚪ 0"
@staticmethod
def format_stats_response(data: Dict[str, Any], username: str, period: str, lang: str = 'en') -> str:
def _escape_md(text: str) -> str:
"""Escape literal Markdown-special chars in dynamic text (e.g. usernames) for legacy Telegram Markdown"""
if not text:
return text
for ch in ('_', '*', '`', '['):
text = text.replace(ch, '\\' + ch)
return text
@staticmethod
def _format_accuracy(accuracy: Optional[float]) -> str:
"""Format accuracy percentage, or '-' if not available (game not analyzed)"""
if accuracy is None:
return "-"
return f"{accuracy:.0f}%"
@staticmethod
def _format_result(result: str) -> str:
"""Format a raw '1-0'/'0-1'/'1/2-1/2' result string for display"""
return "½-½" if result == "1/2-1/2" else result
@staticmethod
def _format_row_outcome_circle(row: dict) -> str:
"""
Colored circle for the tracked user's outcome in this row:
🟢 win, 🔴 loss, draw (or unknown side).
"""
tracked_is_white = row.get('tracked_is_white')
result = row.get('result', '')
if tracked_is_white is None or result == "1/2-1/2":
return ""
tracked_won = (result == "1-0" and tracked_is_white) or (result == "0-1" and not tracked_is_white)
return "🟢" if tracked_won else "🔴"
@staticmethod
def _format_rating_diff_paren(diff: Optional[int]) -> str:
"""Format the tracked user's rating change for this game, e.g. '(+5)' / '(-7)' / '(0)'"""
diff = diff or 0
if diff > 0:
return f"(+{diff})"
elif diff < 0:
return f"({diff})"
else:
return "(0)"
@staticmethod
def _format_game_rows_block(rows: list) -> str:
"""
Format a list of individual game rows as a column-aligned, monospace table:
outcome | rating_diff | accuracy1 | White | rating1 | result | Black | rating2 | accuracy2
Column widths are computed from the actual data so every column lines up
character-for-character. Caller is expected to wrap the result in a
Markdown ``` code block ``` so Telegram renders it with a monospace font.
"""
if not rows:
return ""
NAME_WIDTH = 7
prepared = []
for row in rows:
circle = StatsFormatter._format_row_outcome_circle(row)
diff_str = StatsFormatter._format_rating_diff_paren(row.get('tracked_rating_diff'))
white = (row.get('white_name') or '?')[:NAME_WIDTH]
black = (row.get('black_name') or '?')[:NAME_WIDTH]
wr = row.get('white_rating')
br = row.get('black_rating')
wr_str = str(wr) if wr is not None else "-"
br_str = str(br) if br is not None else "-"
wa = StatsFormatter._format_accuracy(row.get('white_accuracy'))
ba = StatsFormatter._format_accuracy(row.get('black_accuracy'))
result = StatsFormatter._format_result(row.get('result', ''))
prepared.append((circle, diff_str, wa, white, wr_str, result, black, br_str, ba))
diff_width = max(len(p[1]) for p in prepared)
acc_width = max(max(len(p[2]), len(p[8])) for p in prepared)
name_width = NAME_WIDTH
rating_width = max(max(len(p[4]), len(p[7])) for p in prepared)
result_width = max(len(p[5]) for p in prepared)
lines = []
for circle, diff_str, wa, white, wr_str, result, black, br_str, ba in prepared:
lines.append(
f"{circle} {diff_str:>{diff_width}} {wa:>{acc_width}} {white:<{name_width}} {wr_str:>{rating_width}} "
f"{result:^{result_width}} {black:<{name_width}} {br_str:>{rating_width}} {ba:>{acc_width}}"
)
return "\n".join(lines)
@staticmethod
def format_stats_response(data: Dict[str, Any], username: str, period: str, lang: str = 'en', accuracy_extra: Optional[Dict[str, Any]] = None) -> str:
"""Format statistics response according to the template"""
if not data or data.get('data') is None:
message = data.get('message', t('no_data', lang)) if data else t('no_data', lang)
@ -39,45 +128,78 @@ class StatsFormatter:
unsolved = tasks.get('unsolved', 0)
task_text = t('puzzles_section', lang, total=total_tasks, solved=solved, unsolved=unsolved)
# Supplementary per-mode accuracy (+ per-game rows for today/yesterday), from a parallel
# games-of-period fetch. Purely additive: absent/None just means no accuracy shown.
accuracy_by_mode = (accuracy_extra.get('data') or {}) if accuracy_extra else {}
# Format games section
games_text = ""
if games:
for game_type, game_data in games.items():
if not game_data or game_data.get('games_played', 0) == 0:
continue
# Get game type emoji
emoji = StatsFormatter._get_game_type_emoji(game_type)
games_count = game_data.get('games_played', 0)
rating_change = game_data.get('rating_change', 0)
rating = game_data.get('final_rating', 0)
wins = game_data.get('wins', 0)
losses = game_data.get('losses', 0)
draws = game_data.get('draws', 0)
# Format rating change
rating_change_str = StatsFormatter._format_rating_change(rating_change)
# Get game type name (capitalize first letter)
game_type_name = game_type.title()
games_text += t('games_section', lang,
emoji=emoji,
game_type=game_type_name,
games_count=games_count,
rating_change=rating_change_str,
rating=rating,
wins=wins,
losses=losses,
draws=draws
)
# Точность в шапке имеет смысл только там, где нет построчного разбора партий
# ниже (week) — для today/yesterday она была бы избыточна рядом со строками партий.
if game_type in ('blitz', 'rapid', 'classical') and period == "week":
mode_accuracy = (accuracy_by_mode.get(game_type) or {}).get('accuracy')
games_text += t('games_section_with_accuracy', lang,
emoji=emoji,
game_type=game_type_name,
games_count=games_count,
rating_change=rating_change_str,
rating=rating,
wins=wins,
losses=losses,
draws=draws,
accuracy=StatsFormatter._format_accuracy(mode_accuracy)
)
else:
games_text += t('games_section', lang,
emoji=emoji,
game_type=game_type_name,
games_count=games_count,
rating_change=rating_change_str,
rating=rating,
wins=wins,
losses=losses,
draws=draws
)
# Per-game row breakdown (today/yesterday only — week/lastYear stay aggregate-only)
rows_text = ""
if period in ("today", "yesterday"):
for mode in ("blitz", "rapid", "classical"):
rows = (accuracy_by_mode.get(mode) or {}).get('games') or []
if not rows:
continue
emoji = StatsFormatter._get_game_type_emoji(mode)
rows_text += t('game_rows_heading', lang, emoji=emoji, game_type=mode.title())
rows_text += "```\n" + StatsFormatter._format_game_rows_block(rows) + "\n```\n\n"
# Combine all parts
result = t('stats_title', lang, username=username, date_range=date_range)
result = t('stats_title', lang, username=StatsFormatter._escape_md(username), date_range=date_range)
result += task_text
result += games_text.rstrip()
if rows_text:
result += "\n\n" + rows_text.rstrip()
return result
@staticmethod
@ -123,7 +245,7 @@ class StatsFormatter:
else:
period_text = t('period_minutes_text', lang, period=period_minutes)
result = t('period_notification_title', lang, username=username, period_text=period_text)
result = t('period_notification_title', lang, username=StatsFormatter._escape_md(username), period_text=period_text)
# Format puzzles first (if available and there's actual activity)
has_puzzles_data = False
@ -157,6 +279,7 @@ class StatsFormatter:
effective_games_count = top_level_games_count if top_level_games_count > 0 else total_games
# Show details for each game type if there were games
rows_text = ""
if effective_games_count > 0:
for game_type, game_data in games_info.items():
if game_type != 'total' and game_data and game_data.get('games_played', 0) > 0:
@ -168,25 +291,45 @@ class StatsFormatter:
wins = game_data.get('wins', 0)
losses = game_data.get('losses', 0)
draws = game_data.get('draws', 0)
rating_change_str = StatsFormatter._format_rating_change(rating_change)
game_type_name = game_type.title()
result += t('period_games_section', lang,
emoji=emoji,
game_type=game_type_name,
games_count=games_count,
rating_change=rating_change_str,
rating=rating,
wins=wins,
losses=losses,
draws=draws
)
if game_type in ('blitz', 'rapid', 'classical'):
# Точность в шапку не выводим — она всегда дублируется построчным
# разбором партий ниже (периодическая проверка — короткий период).
result += t('period_games_section', lang,
emoji=emoji,
game_type=game_type_name,
games_count=games_count,
rating_change=rating_change_str,
rating=rating,
wins=wins,
losses=losses,
draws=draws
)
rows = game_data.get('games') or []
if rows:
rows_text += t('game_rows_heading', lang, emoji=emoji, game_type=game_type_name)
rows_text += "```\n" + StatsFormatter._format_game_rows_block(rows) + "\n```\n\n"
else:
result += t('period_games_section', lang,
emoji=emoji,
game_type=game_type_name,
games_count=games_count,
rating_change=rating_change_str,
rating=rating,
wins=wins,
losses=losses,
draws=draws
)
if rows_text:
result += rows_text
# If no activity at all
if not has_games_data and not has_puzzles_data:
result += t('no_activity', lang)
return result.rstrip()
@staticmethod
@ -202,14 +345,15 @@ class StatsFormatter:
period_end = data.get('period_end')
stats = (data.get('data') or {})
# Title and subheader
escaped_username = StatsFormatter._escape_md(username)
if games_count >= 1000:
header = f"📈 {username}: last 1000 rated games"
header = f"📈 {escaped_username}: last 1000 rated games"
earliest_ts = data.get('earliest_game_ts')
if isinstance(earliest_ts, int):
earliest = datetime.fromtimestamp(earliest_ts).strftime("%d.%m.%Y")
header += f"\n\n\nStart of these 1000 games: {earliest}"
else:
header = f"📈 {username}: last year (rated), games: {games_count}"
header = f"📈 {escaped_username}: last year (rated), games: {games_count}"
# Use earliest actual game date instead of naive 'year ago'
earliest_ts = data.get('earliest_game_ts', period_start)
if isinstance(earliest_ts, int) and isinstance(period_end, int):
@ -233,9 +377,10 @@ class StatsFormatter:
rating_change_str = StatsFormatter._format_rating_change(rating_change)
rating = mode_stats.get('rating')
rating_str = rating if rating is not None else ""
lines.append(
f"{emoji} {mode.title()}: {games_played} Δ {rating_change_str} R {rating_str}{wins}{losses} 🤝 {draws}"
)
line = f"{emoji} {mode.title()}: {games_played} Δ {rating_change_str} R {rating_str}{wins}{losses} 🤝 {draws}"
if mode in ("blitz", "rapid", "classical"):
line += f" 🎯 {StatsFormatter._format_accuracy(mode_stats.get('accuracy'))}"
lines.append(line)
# Join lines with newlines between each mode
# Between regular modes: one empty line (\n\n)
# Before last mode: two empty lines (\n\n\n)

View file

@ -93,7 +93,9 @@ TRANSLATIONS = {
'stats_title': "📊 Statistics {username}{date_range}\n\n",
'puzzles_section': "🧩 Puzzles: {total} (✅ {solved} - ❌ {unsolved})\n\n",
'games_section': "{emoji} {game_type}{games_count} games • {rating_change}\nRating: {rating}\n✅ Wins: {wins}\n❌ Losses: {losses}\n🤝 Draws: {draws}\n\n",
'games_section_with_accuracy': "{emoji} {game_type}{games_count} games • {rating_change}\nRating: {rating}\n✅ Wins: {wins}\n❌ Losses: {losses}\n🤝 Draws: {draws}\n🎯 Accuracy: {accuracy}\n\n",
'game_rows_heading': "{emoji} {game_type} games:\n",
# Set period
'select_gamer_for_period': "⏱️ <b>Select player to set notification period:</b>\n\n",
'select_period': "⏱️ Select period for player {username}:\n📱 Notifications will be sent to personal messages",
@ -110,7 +112,7 @@ TRANSLATIONS = {
'period_puzzles_section': "🧩 Puzzles: {total} (✅ {solved} - ❌ {failed})\n\n",
'period_games_section': "{emoji} {game_type}{games_count} games • {rating_change}\nRating: {rating}\n✅ Wins: {wins}\n❌ Losses: {losses}\n🤝 Draws: {draws}\n\n",
'no_activity': "📭 No activity for this period",
# Last year or 1000 games
'last_year_1000_processing': "⏳ Processing request... This may take a while as requests are very slow.",
'last_year_1000_player_processing': "🔄 Requesting data for player <b>{username}</b>...",
@ -228,7 +230,9 @@ TRANSLATIONS = {
'stats_title': "📊 Статистика {username}{date_range}\n\n",
'puzzles_section': "🧩 Пазлы: {total} (✅ {solved} - ❌ {unsolved})\n\n",
'games_section': "{emoji} {game_type}{games_count} игр • {rating_change}\nРейтинг: {rating}\n✅ Побед: {wins}\n❌ Поражений: {losses}\n🤝 Ничьих: {draws}\n\n",
'games_section_with_accuracy': "{emoji} {game_type}{games_count} игр • {rating_change}\nРейтинг: {rating}\n✅ Побед: {wins}\n❌ Поражений: {losses}\n🤝 Ничьих: {draws}\n🎯 Точность: {accuracy}\n\n",
'game_rows_heading': "{emoji} Партии {game_type}:\n",
# Set period
'select_gamer_for_period': "⏱️ <b>Выберите игрока для установки периода уведомлений:</b>\n\n",
'select_period': "⏱️ Выберите период для игрока {username}:\n📱 Уведомления будут отправляться в личные сообщения",
@ -245,7 +249,7 @@ TRANSLATIONS = {
'period_puzzles_section': "🧩 Пазлы: {total} (✅ {solved} - ❌ {failed})\n\n",
'period_games_section': "{emoji} {game_type}{games_count} игр • {rating_change}\nРейтинг: {rating}\n✅ Побед: {wins}\n❌ Поражений: {losses}\n🤝 Ничьих: {draws}\n\n",
'no_activity': "📭 Нет активности за этот период",
# Last year or 1000 games
'last_year_1000_processing': "⏳ Обработка запроса... Это может занять некоторое время, так как запросы очень медленные.",
'last_year_1000_player_processing': "🔄 Запрос данных для игрока <b>{username}</b>...",

View file

@ -98,7 +98,7 @@ class LichessAPI:
logger.error(f"Error getting week stats: {e}")
return None
async def get_games_period(self, username: str, since: int, until: int, rated_only: Optional[bool] = None) -> Optional[Dict[str, Any]]:
async def get_games_period(self, username: str, since: int, until: int, rated_only: Optional[bool] = None, include_games: bool = False) -> Optional[Dict[str, Any]]:
"""Get games for a specific period"""
await self.rate_limiter.wait_if_needed()
try:
@ -106,6 +106,8 @@ class LichessAPI:
params = {"since": since, "until": until}
if rated_only is not None:
params["rated_only"] = "true" if rated_only else "false"
if include_games:
params["include_games"] = "true"
logger.info(f"🔍 LichessAPI.get_games_period: URL={url}, params={params}")
async with aiohttp.ClientSession() as session: