b0173950b8
All checks were successful
CI/CD Pipeline / build-and-deploy (push) Successful in 13s
Persist user_data (PicklePersistence) so an in-flight /addgamer username prompt survives a bot restart instead of being silently swallowed by handle_username when the in-memory awaiting flag is gone. Collapse periodic-check backlog into a single request spanning the whole missed gap instead of replaying it one period_minutes window at a time — with enough tracked gamers sharing one RequestQueue, per-window replay could never catch up and the checkpoint fell further behind indefinitely. Notification period label now reflects the actual queried span (minutes/ hours/days) instead of the configured interval, so a weeks-old catch-up no longer gets mislabeled as "for 15 minutes".
463 lines
21 KiB
Python
463 lines
21 KiB
Python
from typing import Dict, Any, Optional
|
||
from datetime import datetime
|
||
import html
|
||
from i18n import t
|
||
|
||
class StatsFormatter:
|
||
@staticmethod
|
||
def _format_rating_change(rating_change: int) -> str:
|
||
"""Format rating change with colored circles"""
|
||
if rating_change > 0:
|
||
return f"🟢 +{rating_change}"
|
||
elif rating_change < 0:
|
||
return f"🔴 {rating_change}"
|
||
else:
|
||
return "⚪ 0"
|
||
|
||
@staticmethod
|
||
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_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_game_rows_block(rows: list) -> str:
|
||
"""
|
||
Format a list of individual game rows as a column-aligned, monospace table:
|
||
outcome | accuracy(tracked) | name(tracked)+color | rating(tracked) | vs | rating(opponent) | accuracy(opponent)
|
||
|
||
Only the tracked player is named — the opponent is shown by rating/accuracy
|
||
only, keeping the line short enough for mobile screens. A small circle right
|
||
after the tracked player's name shows which color they played (⚪ white / ⚫ black).
|
||
|
||
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)
|
||
tracked_is_white = row.get('tracked_is_white') is not False
|
||
color_marker = "⚪" if tracked_is_white else "⚫"
|
||
if tracked_is_white:
|
||
tracked_name = row.get('white_name') or '?'
|
||
tracked_rating = row.get('white_rating')
|
||
tracked_accuracy = row.get('white_accuracy')
|
||
opp_rating = row.get('black_rating')
|
||
opp_accuracy = row.get('black_accuracy')
|
||
else:
|
||
tracked_name = row.get('black_name') or '?'
|
||
tracked_rating = row.get('black_rating')
|
||
tracked_accuracy = row.get('black_accuracy')
|
||
opp_rating = row.get('white_rating')
|
||
opp_accuracy = row.get('white_accuracy')
|
||
name_field = f"{tracked_name[:NAME_WIDTH]}{color_marker}"
|
||
tr_str = str(tracked_rating) if tracked_rating is not None else "-"
|
||
or_str = str(opp_rating) if opp_rating is not None else "-"
|
||
ta = StatsFormatter._format_accuracy(tracked_accuracy)
|
||
oa = StatsFormatter._format_accuracy(opp_accuracy)
|
||
prepared.append((circle, ta, name_field, tr_str, or_str, oa))
|
||
|
||
acc_width = max(max(len(p[1]), len(p[5])) for p in prepared)
|
||
name_width = NAME_WIDTH + 1 # +1 for the color marker glued to the name
|
||
rating_width = max(max(len(p[3]), len(p[4])) for p in prepared)
|
||
|
||
lines = []
|
||
for circle, ta, name_field, tr_str, or_str, oa in prepared:
|
||
lines.append(
|
||
f"{circle} {ta:>{acc_width}} {name_field:<{name_width}} {tr_str:>{rating_width}} "
|
||
f"- {or_str:>{rating_width}} {oa:>{acc_width}}"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
@staticmethod
|
||
def _format_mode_stats_block(rating, wins, losses, draws, lang: str) -> str:
|
||
"""
|
||
Column-aligned monospace block for one game type's rating/wins/losses/draws,
|
||
shown under the header line. Caller is expected to wrap the result in a
|
||
Markdown ``` code block ``` for guaranteed monospace rendering.
|
||
|
||
No accuracy row here: an average accuracy across several games isn't a
|
||
meaningful stat (unlike the per-game accuracy shown in the game-rows table).
|
||
|
||
Every label carries exactly one emoji prefix — Telegram's monospace font
|
||
renders emoji wider than plain text of the same length() count, so a label
|
||
with no emoji (unlike the other three) would visibly end up one column off
|
||
despite matching character counts.
|
||
"""
|
||
rows = [
|
||
(t('stat_label_rating', lang), str(rating)),
|
||
(t('stat_label_wins', lang), str(wins)),
|
||
(t('stat_label_losses', lang), str(losses)),
|
||
(t('stat_label_draws', lang), str(draws)),
|
||
]
|
||
|
||
label_width = max(len(label) for label, _ in rows)
|
||
value_width = max(len(value) for _, value in rows)
|
||
return "\n".join(f"{label:<{label_width}} {value:>{value_width}}" for label, value in rows)
|
||
|
||
@staticmethod
|
||
def format_gamers_table(gamers_data: list) -> str:
|
||
"""
|
||
Column-aligned monospace table for /getgamers: username, then bullet/blitz/rapid
|
||
ratings, then the periodic-notification period suffix (e.g. "· 60m").
|
||
|
||
Column widths are computed from the actual data so every column lines up
|
||
character-for-character. Caller is expected to wrap the result in an HTML
|
||
<pre> block so Telegram renders it with a monospace font (username can't be
|
||
bolded there — pre/code entities can't contain other entities).
|
||
"""
|
||
if not gamers_data:
|
||
return ""
|
||
|
||
rows = []
|
||
for g in gamers_data:
|
||
username = html.escape(str(g['username']))
|
||
bullet = str(g['bullet'])
|
||
blitz = str(g['blitz'])
|
||
rapid = str(g['rapid'])
|
||
period = (g.get('period') or '').strip()
|
||
rows.append((username, bullet, blitz, rapid, period))
|
||
|
||
name_width = max(len(r[0]) for r in rows)
|
||
bullet_width = max(len(r[1]) for r in rows)
|
||
blitz_width = max(len(r[2]) for r in rows)
|
||
rapid_width = max(len(r[3]) for r in rows)
|
||
|
||
lines = []
|
||
for username, bullet, blitz, rapid, period in rows:
|
||
line = (
|
||
f"{username:<{name_width}} "
|
||
f"⚡{bullet:>{bullet_width}} 🔥{blitz:>{blitz_width}} 🐇{rapid:>{rapid_width}}"
|
||
)
|
||
if period:
|
||
line += f" {period}"
|
||
lines.append(line)
|
||
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)
|
||
# Filter out old "No active player" messages - this functionality is deprecated
|
||
if 'No active player' in message or 'Нет активного игрока' in message or 'active player' in message.lower() or 'активного игрока' in message.lower():
|
||
return t('no_data', lang)
|
||
return f"📭 {message}"
|
||
|
||
# Extract data from API response
|
||
api_data = data.get('data', {})
|
||
tasks = api_data.get('tasks', {})
|
||
games = api_data.get('games', {})
|
||
|
||
# Format date range
|
||
date_range = StatsFormatter._get_date_range(period, lang)
|
||
|
||
# Format tasks section
|
||
task_text = ""
|
||
if tasks and tasks.get('total', 0) > 0:
|
||
total_tasks = tasks.get('total', 0)
|
||
solved = tasks.get('solved', 0)
|
||
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('game_type_header', lang,
|
||
emoji=emoji,
|
||
game_type=game_type_name,
|
||
games_count=games_count,
|
||
rating_change=rating_change_str
|
||
)
|
||
|
||
games_text += StatsFormatter._format_mode_stats_block(
|
||
rating, wins, losses, draws, lang
|
||
) + "\n\n"
|
||
|
||
# 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 += StatsFormatter._format_game_rows_block(rows) + "\n\n"
|
||
|
||
# Whole body goes into a single monospace code block so the message doesn't
|
||
# alternate between plain-text headers and separately-fenced tables.
|
||
body = t('stats_title', lang, username=username, date_range=date_range)
|
||
body += task_text
|
||
body += games_text.rstrip()
|
||
if rows_text:
|
||
body += "\n\n" + rows_text.rstrip()
|
||
|
||
return "```\n" + body + "\n```"
|
||
|
||
@staticmethod
|
||
def _get_date_range(period: str, lang: str = 'en') -> str:
|
||
"""Get date range string for the period"""
|
||
from datetime import datetime, timedelta
|
||
|
||
today = datetime.now()
|
||
|
||
if period == "today":
|
||
return today.strftime("%d.%m.%Y")
|
||
elif period == "yesterday":
|
||
yesterday = today - timedelta(days=1)
|
||
return yesterday.strftime("%d.%m.%Y")
|
||
elif period == "week":
|
||
week_ago = today - timedelta(days=7)
|
||
return f"{week_ago.strftime('%d.%m.%Y')}–{today.strftime('%d.%m.%Y')}"
|
||
else:
|
||
return today.strftime("%d.%m.%Y")
|
||
|
||
@staticmethod
|
||
def _get_game_type_emoji(game_type: str) -> str:
|
||
"""Get emoji for game type"""
|
||
emoji_map = {
|
||
'bullet': '⚡️',
|
||
'blitz': '🔥',
|
||
'rapid': '🐇',
|
||
'classical': '♟️',
|
||
'correspondence': '📮'
|
||
}
|
||
return emoji_map.get(game_type.lower(), '🎯')
|
||
|
||
@staticmethod
|
||
def format_period_notification(username: str, games_data: Optional[Dict], puzzles_data: Optional[Dict], period_minutes: int, lang: str = 'en') -> str:
|
||
"""Format notification for periodic checks"""
|
||
from datetime import datetime
|
||
|
||
# Format period text. period_minutes here is the actual queried span, which
|
||
# can be much larger than the configured check interval after a backlog
|
||
# collapse (see periodic_check in bot.py) — fall back to hours/days so a
|
||
# weeks-old catch-up doesn't get mislabeled as "for 15 minutes".
|
||
if period_minutes < 60:
|
||
if period_minutes == 1:
|
||
period_text = t('period_1_minute', lang)
|
||
elif period_minutes in [2, 3, 4]:
|
||
period_text = t('period_2_3_4_minutes', lang, period=period_minutes)
|
||
else:
|
||
period_text = t('period_minutes_text', lang, period=period_minutes)
|
||
elif period_minutes < 1440:
|
||
hours = round(period_minutes / 60)
|
||
period_text = t('period_hours_text', lang, hours=hours)
|
||
else:
|
||
days = round(period_minutes / 1440)
|
||
period_text = t('period_days_text', lang, days=days)
|
||
|
||
result = t('period_notification_title', lang, username=username, period_text=period_text)
|
||
|
||
# Format puzzles first (if available and there's actual activity)
|
||
has_puzzles_data = False
|
||
if puzzles_data:
|
||
# Check puzzles_in_period on top level first (priority)
|
||
top_level_puzzles = puzzles_data.get('puzzles_in_period', 0)
|
||
# Also check data.total_attempts
|
||
if puzzles_data.get('data'):
|
||
puzzles_info = puzzles_data['data']
|
||
total_puzzles = puzzles_info.get('total_attempts', 0)
|
||
solved = puzzles_info.get('solved', 0)
|
||
failed = puzzles_info.get('failed', 0)
|
||
|
||
effective_puzzles = top_level_puzzles if top_level_puzzles > 0 else total_puzzles
|
||
|
||
# Only show tasks section if there's actual activity (not all zeros)
|
||
if effective_puzzles > 0 or solved > 0 or failed > 0:
|
||
has_puzzles_data = True
|
||
result += t('period_puzzles_section', lang, total=effective_puzzles, solved=solved, failed=failed)
|
||
|
||
# Format games
|
||
has_games_data = False
|
||
if games_data and games_data.get('data'):
|
||
games_info = games_data['data']
|
||
# Check games_count on top level first (priority)
|
||
top_level_games_count = games_data.get('games_count', 0)
|
||
# Also check data.total.games_played
|
||
total_games = games_info.get('total', {}).get('games_played', 0)
|
||
|
||
# Use top-level games_count if available, otherwise use total.games_played
|
||
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:
|
||
has_games_data = True # Only set to True if we actually add game data
|
||
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', game_data.get('rating', 0))
|
||
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('game_type_header', lang,
|
||
emoji=emoji,
|
||
game_type=game_type_name,
|
||
games_count=games_count,
|
||
rating_change=rating_change_str
|
||
)
|
||
result += StatsFormatter._format_mode_stats_block(rating, wins, losses, draws, lang) + "\n\n"
|
||
|
||
if game_type in ('blitz', 'rapid', 'classical'):
|
||
rows = game_data.get('games') or []
|
||
if rows:
|
||
rows_text += t('game_rows_heading', lang, emoji=emoji, game_type=game_type_name)
|
||
rows_text += StatsFormatter._format_game_rows_block(rows) + "\n\n"
|
||
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)
|
||
|
||
# Whole body goes into a single monospace code block so the message doesn't
|
||
# alternate between plain-text headers and separately-fenced tables.
|
||
return "```\n" + result.rstrip() + "\n```"
|
||
|
||
@staticmethod
|
||
def format_last_year_or_1000(data: Dict[str, Any], username: str, lang: str = 'en') -> str:
|
||
"""
|
||
Format response for last year or last 1000 games.
|
||
Expects GamesOfPeriodResponse-like payload.
|
||
"""
|
||
if not data:
|
||
return "📭 No data"
|
||
games_count = data.get('games_count', 0)
|
||
period_start = data.get('period_start')
|
||
period_end = data.get('period_end')
|
||
stats = (data.get('data') or {})
|
||
# Title and subheader
|
||
if games_count >= 1000:
|
||
header = f"📈 {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\nStart of these 1000 games: {earliest}"
|
||
else:
|
||
header = f"📈 {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):
|
||
start_str = datetime.fromtimestamp(earliest_ts).strftime("%d.%m.%Y")
|
||
end_str = datetime.fromtimestamp(period_end).strftime("%d.%m.%Y")
|
||
header += f"\n\nPeriod: {start_str}–{end_str}"
|
||
|
||
# Collect per-mode rows
|
||
rows = []
|
||
for mode in ["bullet", "blitz", "rapid", "classical", "correspondence"]:
|
||
mode_stats = stats.get(mode)
|
||
if not mode_stats:
|
||
continue
|
||
games_played = mode_stats.get('games_played', 0)
|
||
if games_played == 0:
|
||
continue
|
||
# Strip the variation selector some emoji include (e.g. bullet/classical) —
|
||
# it's invisible but counts as an extra character, which would throw off
|
||
# the column padding below even though it doesn't add any visual width.
|
||
emoji = StatsFormatter._get_game_type_emoji(mode).replace('️', '')
|
||
wins = mode_stats.get('wins', 0)
|
||
losses = mode_stats.get('losses', 0)
|
||
draws = mode_stats.get('draws', 0)
|
||
rating_change = mode_stats.get('rating_change', 0)
|
||
rating = mode_stats.get('rating')
|
||
accuracy_str = "-"
|
||
if mode in ("blitz", "rapid", "classical"):
|
||
accuracy_str = StatsFormatter._format_accuracy(mode_stats.get('accuracy'))
|
||
rows.append((
|
||
f"{emoji} {mode.title()}",
|
||
str(games_played),
|
||
StatsFormatter._format_rating_change(rating_change),
|
||
str(rating) if rating is not None else "—",
|
||
str(wins),
|
||
str(losses),
|
||
str(draws),
|
||
accuracy_str,
|
||
))
|
||
|
||
if not rows:
|
||
return f"{header}\n\n📭 No data"
|
||
|
||
label_w = max(len(r[0]) for r in rows)
|
||
games_w = max(len(r[1]) for r in rows)
|
||
change_w = max(len(r[2]) for r in rows)
|
||
rating_w = max(len(r[3]) for r in rows)
|
||
wins_w = max(len(r[4]) for r in rows)
|
||
losses_w = max(len(r[5]) for r in rows)
|
||
draws_w = max(len(r[6]) for r in rows)
|
||
acc_w = max((len(r[7]) for r in rows), default=0)
|
||
|
||
lines = []
|
||
for label, games, change, rating, wins, losses, draws, acc in rows:
|
||
line = (
|
||
f"{label:<{label_w}} {games:>{games_w}} {change:<{change_w}} "
|
||
f"R{rating:>{rating_w}} ✅{wins:>{wins_w}} ❌{losses:>{losses_w}} 🤝{draws:>{draws_w}}"
|
||
)
|
||
if acc_w:
|
||
line += f" 🎯{acc:>{acc_w}}"
|
||
lines.append(line)
|
||
|
||
body = "\n".join(lines)
|
||
return f"{header}\n\n```\n{body}\n```"
|