#!/usr/bin/env python3 """ Скрипт для проверки активности всех отслеживаемых игроков за сегодня. Делает запросы к Lichess API с задержкой 200 мс между запросами. """ import sqlite3 import logging import requests import json import time from datetime import datetime, timezone from typing import List, Dict, Any, Optional import os # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def get_all_tracked_players(db_path: str) -> List[str]: """Получить список всех уникальных отслеживаемых игроков из базы данных""" try: with sqlite3.connect(db_path) as conn: cursor = conn.cursor() # Получаем все уникальные username из таблицы gamers cursor.execute("SELECT DISTINCT username FROM gamers ORDER BY username") players = [row[0] for row in cursor.fetchall()] return players except Exception as e: logger.error(f"Ошибка при получении списка игроков из базы: {e}") return [] def get_today_timestamps() -> tuple[int, int]: """Получить начало и конец сегодняшнего дня в миллисекундах""" now = datetime.now(timezone.utc) # Начало сегодняшнего дня (00:00:00 UTC) today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) # Конец сегодняшнего дня (23:59:59.999 UTC) today_end = now.replace(hour=23, minute=59, second=59, microsecond=999000) # Конвертируем в миллисекунды (Unix timestamp * 1000) since_ms = int(today_start.timestamp() * 1000) until_ms = int(today_end.timestamp() * 1000) return since_ms, until_ms class RateLimiter: """Простой rate limiter для синхронных запросов""" def __init__(self, min_delay: float = 1.0): self.min_delay = min_delay self.last_request_time = None def wait_before_request(self): """Ждать перед запросом если нужно соблюсти минимальную задержку""" if self.last_request_time is not None: elapsed = time.time() - self.last_request_time if elapsed < self.min_delay: wait_time = self.min_delay - elapsed logger.debug(f"Ждем {wait_time:.2f} сек перед следующим запросом") time.sleep(wait_time) def wait_after_response(self): """Ждать после получения ответа чтобы соблюсти минимальную задержку между запросами""" self.last_request_time = time.time() # Всегда ждем 1 секунду после получения ответа перед следующим запросом time.sleep(self.min_delay) def get_games_of_period(rate_limiter: RateLimiter, username: str, since_ms: int, until_ms: int, rated_only: bool = False) -> Optional[List[Dict[str, Any]]]: """Получить игры пользователя за период напрямую из Lichess API""" # Ждем перед запросом если нужно (для первого запроса не ждем) rate_limiter.wait_before_request() url = f"https://lichess.org/api/games/user/{username}" params = { 'since': since_ms, 'until': until_ms, 'max': 1000 } if rated_only: params['rated'] = 'true' headers = { 'Accept': 'application/x-ndjson' } try: # Делаем запрос response = requests.get(url, params=params, headers=headers, timeout=30) # Получаем ответ и парсим if response.status_code == 404: logger.warning(f"Пользователь {username} не найден (404)") rate_limiter.wait_after_response() # Ждем 1 сек после ответа return None elif response.status_code != 200: logger.error(f"Ошибка API для {username}: {response.status_code} - {response.text[:200]}") rate_limiter.wait_after_response() # Ждем 1 сек после ответа return None # Парсим NDJSON games = [] content = response.text.strip() if content: for line in content.split('\n'): if line.strip(): try: game = json.loads(line) games.append(game) except json.JSONDecodeError as e: logger.warning(f"Ошибка парсинга JSON для {username}: {e}") continue # Ждем 1 секунду после получения ответа перед следующим запросом rate_limiter.wait_after_response() return games except Exception as e: logger.error(f"Ошибка при запросе игр для {username}: {e}") rate_limiter.wait_after_response() # Ждем 1 сек даже при ошибке return None def check_player_activity(rate_limiter: RateLimiter, username: str, since_ms: int, until_ms: int) -> bool: """Проверить, играл ли игрок сегодня""" try: games = get_games_of_period(rate_limiter, username, since_ms, until_ms, rated_only=False) if games is None: # Пользователь не найден или ошибка API return False return len(games) > 0 except Exception as e: logger.error(f"Ошибка при проверке игрока {username}: {e}") return False def main(): """Основная функция скрипта""" # Определяем путь к базе данных db_path = os.getenv("DATABASE_PATH") if not db_path: # Пробуем найти базу в стандартных местах possible_paths = [ "LichessClientTG_bot/data/lichess_bot.db", "data/lichess_bot.db", "/app/data/lichess_bot.db" ] for path in possible_paths: if os.path.exists(path): db_path = path break if not db_path: logger.error("Не удалось найти базу данных. Укажите DATABASE_PATH в переменных окружения.") return logger.info(f"Используется база данных: {db_path}") # Получаем список всех игроков players = get_all_tracked_players(db_path) total_players = len(players) if total_players == 0: logger.warning("В базе данных не найдено отслеживаемых игроков") return logger.info(f"Найдено отслеживаемых игроков: {total_players}") # Получаем временные метки для сегодняшнего дня since_ms, until_ms = get_today_timestamps() today_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") logger.info(f"Проверяем активность за {today_str} (с {since_ms} по {until_ms})") # Создаем rate limiter (10 секунд) rate_limiter = RateLimiter(min_delay=10.0) # Проверяем каждого игрока active_players = [] inactive_players = [] error_players = [] logger.info("Начинаем проверку игроков...") for i, username in enumerate(players, 1): logger.info(f"[{i}/{total_players}] Проверяем {username}...") try: is_active = check_player_activity(rate_limiter, username, since_ms, until_ms) if is_active: active_players.append(username) logger.info(f" ✓ {username} играл сегодня") else: inactive_players.append(username) logger.info(f" ✗ {username} не играл сегодня") except Exception as e: error_players.append((username, str(e))) logger.error(f" ✗ Ошибка при проверке {username}: {e}") # Выводим итоговую статистику print("\n" + "="*60) print("ИТОГОВАЯ СТАТИСТИКА") print("="*60) print(f"Всего отслеживаемых игроков: {total_players}") print(f"Играли сегодня: {len(active_players)}") print(f"Не играли сегодня: {len(inactive_players)}") if error_players: print(f"Ошибки при проверке: {len(error_players)}") print("="*60) if active_players: print(f"\nИгроки, которые играли сегодня ({len(active_players)}):") for username in active_players: print(f" • {username}") if error_players: print(f"\nИгроки с ошибками ({len(error_players)}):") for username, error in error_players: print(f" • {username}: {error}") if __name__ == "__main__": main()