331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""
|
|
Player tracking: FIDE profile lookup, subscription management,
|
|
background polling for new results and pairings.
|
|
"""
|
|
import re
|
|
import os
|
|
import sys
|
|
import sqlite3
|
|
import requests
|
|
from pathlib import Path
|
|
from datetime import date
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
DB_PATH = os.environ.get('STATS_DB', '/app/data/tournaments.db')
|
|
FIDE_URL = 'https://ratings.fide.com/profile/{fide_id}'
|
|
HEADERS = {
|
|
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36'
|
|
}
|
|
|
|
RESULT_LOCALE = {
|
|
'ru': {
|
|
'win': 'победа', 'draw': 'ничья', 'loss': 'поражение',
|
|
'white': 'белыми', 'black': 'чёрными',
|
|
'points': 'Очков', 'round': 'тур', 'vs': 'против',
|
|
},
|
|
'en': {
|
|
'win': 'win', 'draw': 'draw', 'loss': 'loss',
|
|
'white': 'White', 'black': 'Black',
|
|
'points': 'Points', 'round': 'round', 'vs': 'vs',
|
|
},
|
|
}
|
|
|
|
|
|
def _get_conn():
|
|
p = Path(DB_PATH)
|
|
try:
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
except PermissionError:
|
|
fallback = Path(__file__).parent.parent / 'data' / 'tournaments.db'
|
|
fallback.parent.mkdir(parents=True, exist_ok=True)
|
|
p = fallback
|
|
conn = sqlite3.connect(str(p))
|
|
conn.execute('PRAGMA journal_mode=WAL')
|
|
conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS subscriptions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
fide_id INTEGER NOT NULL,
|
|
player_name TEXT NOT NULL,
|
|
rating INTEGER DEFAULT 0,
|
|
fed TEXT DEFAULT '',
|
|
lang TEXT DEFAULT 'en',
|
|
tournament_url TEXT NOT NULL,
|
|
tournament_name TEXT DEFAULT '',
|
|
player_sno INTEGER DEFAULT 0,
|
|
last_results_count INTEGER DEFAULT 0,
|
|
last_points REAL DEFAULT 0.0,
|
|
last_round_done INTEGER DEFAULT 0,
|
|
active INTEGER DEFAULT 1,
|
|
UNIQUE(user_id, fide_id, tournament_url)
|
|
)
|
|
''')
|
|
conn.commit()
|
|
return conn
|
|
|
|
|
|
# ═══ FIDE profile scraper ═══
|
|
|
|
def fetch_fide_player(fide_id: int) -> dict:
|
|
url = FIDE_URL.format(fide_id=fide_id)
|
|
resp = requests.get(url, headers=HEADERS, timeout=20)
|
|
if resp.status_code != 200:
|
|
raise RuntimeError(f'FIDE profile not found (HTTP {resp.status_code})')
|
|
html = resp.text
|
|
|
|
# Check for "No record found"
|
|
if 'No record found' in html:
|
|
raise RuntimeError(f'Player with FIDE ID {fide_id} not found')
|
|
|
|
# Name: <h1 class="player-title">Carlsen, Magnus</h1>
|
|
name = ''
|
|
m = re.search(r'<h1\s+class="player-title"[^>]*>\s*(.*?)\s*</h1>', html)
|
|
if m:
|
|
name = m.group(1).strip()
|
|
name = re.sub(r'<[^>]+>', '', name).strip()
|
|
if not name:
|
|
raise RuntimeError('Could not extract player name from FIDE profile')
|
|
|
|
# Federation: "National Rank NOR"
|
|
fed = ''
|
|
m = re.search(r'National Rank\s+([A-Z]{3})', html)
|
|
if m:
|
|
fed = m.group(1)
|
|
|
|
# Rating (standard): <p>2841</p><p ...>STANDARD
|
|
rating = 0
|
|
m = re.search(r'<p>(\d{3,4})</p>\s*<p[^>]*>\s*STANDARD', html)
|
|
if m:
|
|
rating = int(m.group(1))
|
|
|
|
# Title: <div class="profile-info-title "><p>Grandmaster</p>
|
|
title = ''
|
|
m = re.search(r'<div\s+class="profile-info-title[^"]*"\s*>\s*<p>(.*?)</p>', html)
|
|
if m:
|
|
title = m.group(1).strip()
|
|
# Some players have multiple titles in adjacent <p> — take only the first one
|
|
title = re.sub(r'<[^>]+>', '', title).strip()
|
|
|
|
return {
|
|
'fide_id': fide_id, 'name': name, 'rating': rating,
|
|
'fed': fed, 'title': title,
|
|
}
|
|
|
|
|
|
# ═══ Subscription CRUD ═══
|
|
|
|
def add_subscription(user_id: int, player: dict, tournament_url: str,
|
|
tournament_name: str, player_sno: int,
|
|
last_results_count: int, last_points: float,
|
|
lang: str):
|
|
conn = _get_conn()
|
|
conn.execute('''
|
|
INSERT OR REPLACE INTO subscriptions
|
|
(user_id, fide_id, player_name, rating, fed, lang, tournament_url,
|
|
tournament_name, player_sno, last_results_count, last_points,
|
|
last_round_done, active)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 1)
|
|
''', (user_id, player['fide_id'], player['name'], player['rating'],
|
|
player.get('fed', ''), lang, tournament_url, tournament_name,
|
|
player_sno, last_results_count, last_points))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def get_user_subs(user_id: int) -> list:
|
|
conn = _get_conn()
|
|
rows = conn.execute(
|
|
'SELECT id, player_name, rating, fed, tournament_name, '
|
|
'last_points, last_round_done, active FROM subscriptions '
|
|
'WHERE user_id = ? AND active = 1 ORDER BY id',
|
|
(user_id,)).fetchall()
|
|
conn.close()
|
|
return [{'id': r[0], 'player_name': r[1], 'rating': r[2], 'fed': r[3],
|
|
'tournament_name': r[4], 'last_points': r[5],
|
|
'last_round_done': r[6], 'active': r[7]} for r in rows]
|
|
|
|
|
|
def get_active_subs() -> list:
|
|
conn = _get_conn()
|
|
rows = conn.execute(
|
|
'SELECT id, user_id, fide_id, player_name, rating, fed, lang, '
|
|
'tournament_url, tournament_name, player_sno, last_results_count, '
|
|
'last_points, last_round_done FROM subscriptions '
|
|
'WHERE active = 1').fetchall()
|
|
conn.close()
|
|
return [dict(zip(
|
|
['id', 'user_id', 'fide_id', 'player_name', 'rating', 'fed', 'lang',
|
|
'tournament_url', 'tournament_name', 'player_sno',
|
|
'last_results_count', 'last_points', 'last_round_done'], r))
|
|
for r in rows]
|
|
|
|
|
|
def remove_subscription(sub_id: int, user_id: int) -> bool:
|
|
conn = _get_conn()
|
|
cur = conn.execute(
|
|
'UPDATE subscriptions SET active = 0 WHERE id = ? AND user_id = ?',
|
|
(sub_id, user_id))
|
|
conn.commit()
|
|
deleted = cur.rowcount > 0
|
|
conn.close()
|
|
return deleted
|
|
|
|
|
|
def update_result(sub_id: int, results_count: int, points: float):
|
|
conn = _get_conn()
|
|
conn.execute(
|
|
'UPDATE subscriptions SET last_results_count = ?, last_points = ? '
|
|
'WHERE id = ?',
|
|
(results_count, points, sub_id))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def update_round(sub_id: int, round_num: int):
|
|
conn = _get_conn()
|
|
conn.execute(
|
|
'UPDATE subscriptions SET last_round_done = ? WHERE id = ?',
|
|
(round_num, sub_id))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
# ═══ Tournament player lookup ═══
|
|
|
|
def _normalize(s: str) -> str:
|
|
return re.sub(r'\s+', ' ', s.strip()).lower()
|
|
|
|
|
|
def find_player(tournament: dict, name: str, sno: int = 0) -> dict | None:
|
|
standings = tournament.get('standings', [])
|
|
|
|
# Try exact SNo match first
|
|
if sno:
|
|
for s in standings:
|
|
if s.get('starting_sno') == sno:
|
|
return s
|
|
|
|
# Try exact name match (normalized)
|
|
norm_target = _normalize(name)
|
|
for s in standings:
|
|
if _normalize(s['name']) == norm_target:
|
|
return s
|
|
|
|
# Fuzzy: split into parts, find best overlap
|
|
target_parts = set(norm_target.split())
|
|
best, best_overlap = None, 0
|
|
for s in standings:
|
|
s_parts = set(_normalize(s['name']).split())
|
|
overlap = len(target_parts & s_parts)
|
|
if overlap > best_overlap:
|
|
best_overlap = overlap
|
|
best = s
|
|
|
|
if best and best_overlap >= 1:
|
|
return best
|
|
|
|
return None
|
|
|
|
|
|
def find_opponent_name(tournament: dict, opp_sno: int) -> str:
|
|
if not opp_sno:
|
|
return 'BYE'
|
|
standings = tournament.get('standings', [])
|
|
for s in standings:
|
|
if s.get('starting_sno') == opp_sno:
|
|
return s['name']
|
|
return f'#{opp_sno}'
|
|
|
|
|
|
# ═══ Result formatting ═══
|
|
|
|
def _md_escape_tracker(text: str) -> str:
|
|
escape_chars = r'_*[]()~`>#+-=|{}.!'
|
|
result = []
|
|
for ch in text:
|
|
if ch in escape_chars:
|
|
result.append('\\' + ch)
|
|
else:
|
|
result.append(ch)
|
|
return ''.join(result)
|
|
|
|
|
|
def format_result(player_name: str, result: dict, opponent_name: str,
|
|
points: float, lang: str) -> str:
|
|
loc = RESULT_LOCALE.get(lang, RESULT_LOCALE['en'])
|
|
score = result.get('score', 0.0)
|
|
color = result.get('color', '')
|
|
rd = result.get('round', '?')
|
|
|
|
score_label = {1.0: loc['win'], 0.5: loc['draw'], 0.0: loc['loss']}
|
|
score_desc = score_label.get(score, f'{score}')
|
|
color_desc = loc['white'] if color == 'w' else loc['black']
|
|
|
|
safe_name = _md_escape_tracker(player_name)
|
|
safe_opp = _md_escape_tracker(opponent_name)
|
|
|
|
return (
|
|
f'♟ *{safe_name}* — {loc["round"]} {rd}\n'
|
|
f'{score_desc} \({color_desc}\) {loc["vs"]} {safe_opp}\n'
|
|
f'{loc["points"]}: {points:.1f}'
|
|
)
|
|
|
|
|
|
# ═══ Background check ═══
|
|
|
|
async def check_all_subscriptions(context):
|
|
from swiss_calc.parser import fetch_tournament
|
|
from swiss_calc.swiss import calculate_next_round
|
|
from bots.client_bot import format_pairings, _render_chunks
|
|
from telegram.constants import ParseMode
|
|
|
|
subs = get_active_subs()
|
|
if not subs:
|
|
return
|
|
|
|
today_str = date.today().isoformat()
|
|
for sub in subs:
|
|
try:
|
|
tournament = fetch_tournament(sub['tournament_url'])
|
|
except Exception:
|
|
continue
|
|
|
|
player = find_player(tournament, sub['player_name'], sub['player_sno'])
|
|
if player is None:
|
|
continue
|
|
|
|
results_count = len(player.get('results', []))
|
|
points = player.get('points', 0.0)
|
|
|
|
# New individual result detected
|
|
if results_count > sub['last_results_count']:
|
|
new_r = player['results'][-1]
|
|
opp_name = find_opponent_name(tournament, new_r.get('opponent', 0))
|
|
msg = format_result(
|
|
player['name'], new_r, opp_name, points, sub['lang'])
|
|
try:
|
|
await context.bot.send_message(
|
|
sub['user_id'], msg, parse_mode=ParseMode.MARKDOWN_V2)
|
|
except Exception:
|
|
pass
|
|
update_result(sub['id'], results_count, points)
|
|
|
|
# Round fully completed → calculate next round pairings
|
|
current_rd = tournament.get('current_round', 0)
|
|
if current_rd > sub['last_round_done'] and current_rd < tournament.get('num_rounds', 0):
|
|
try:
|
|
result = calculate_next_round(tournament)
|
|
fmt = format_pairings(result, tournament.get('name', ''), sub['lang'])
|
|
chunks = _render_chunks(fmt)
|
|
for chunk in chunks:
|
|
try:
|
|
await context.bot.send_message(
|
|
sub['user_id'], chunk,
|
|
parse_mode=ParseMode.MARKDOWN_V2,
|
|
disable_web_page_preview=True)
|
|
except Exception:
|
|
break
|
|
except Exception:
|
|
pass
|
|
update_round(sub['id'], current_rd)
|