230 lines
9.5 KiB
Python
230 lines
9.5 KiB
Python
#!/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()
|
||
|