""" Lichess Statistics API - Основной модуль FastAPI приложения Этот модуль содержит все API эндпоинты для получения статистики игроков Lichess. Включает в себя: - Статистику игр за разные периоды (сегодня, вчера, неделя) - Детальную статистику игр за произвольный период - Статистику решения задач (пазлов) за период - Health check и информационные эндпоинты Автор: Lichess Web Services Team Версия: 1.0.0 """ from fastapi import FastAPI, HTTPException, Path, Query, Header from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager import logging from datetime import datetime from stats_service import StatsService from models import ActivityResponse, ErrorResponse, HealthResponse, GamesOfPeriodResponse, PuzzleOfPeriodResponse # ============================================================================= # НАСТРОЙКА ЛОГИРОВАНИЯ # ============================================================================= # Настройка базового логирования для всего приложения # Уровень INFO позволяет видеть все важные события и ошибки logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ============================================================================= # ГЛОБАЛЬНЫЕ ПЕРЕМЕННЫЕ # ============================================================================= # Глобальный экземпляр сервиса статистики # Инициализируется при запуске приложения и используется во всех эндпоинтах stats_service = None # ============================================================================= # УПРАВЛЕНИЕ ЖИЗНЕННЫМ ЦИКЛОМ ПРИЛОЖЕНИЯ # ============================================================================= @asynccontextmanager async def lifespan(app: FastAPI): """ Контекстный менеджер для управления жизненным циклом FastAPI приложения. Выполняется при запуске и остановке приложения: - При запуске: инициализирует сервис статистики - При остановке: корректно закрывает все соединения Args: app: Экземпляр FastAPI приложения """ # ========== STARTUP (Запуск приложения) ========== global stats_service # Создаем экземпляр сервиса статистики, который будет использоваться во всех эндпоинтах stats_service = StatsService() logger.info("Lichess API сервис запущен") # Передаем управление приложению yield # ========== SHUTDOWN (Остановка приложения) ========== # Корректно закрываем все соединения и освобождаем ресурсы if stats_service: await stats_service.close() logger.info("Lichess API сервис остановлен") # ============================================================================= # СОЗДАНИЕ FASTAPI ПРИЛОЖЕНИЯ # ============================================================================= app = FastAPI( title="Lichess Statistics API", description=""" ## Lichess Statistics API REST API для получения детальной статистики игроков платформы Lichess. ### Возможности: * 📊 Получение статистики игр по режимам (Bullet, Blitz, Rapid) * 🧩 Статистика решения задач (пазлов) * 📅 Статистика за разные периоды (сегодня, вчера, неделя) * 🎯 Отслеживание изменений рейтинга * 📈 Подробная аналитика результатов игр ### Режимы игр: - **Bullet**: Быстрые игры (1-3 минуты) - **Blitz**: Блиц игры (3-10 минут) - **Rapid**: Рапид игры (10+ минут) ### Примеры использования: - Получить статистику за сегодня: `GET /stats/{username}/today` - Получить статистику за вчера: `GET /stats/{username}/yesterday` - Получить статистику за неделю: `GET /stats/{username}/week` """, version="1.0.0", contact={ "name": "Lichess Statistics API Support", "url": "https://github.com/vrubelroman/LichessWebServices", }, license_info={ "name": "MIT", }, lifespan=lifespan, openapi_tags=[ { "name": "health", "description": "Проверка состояния сервиса" }, { "name": "statistics", "description": "Получение статистики игроков Lichess" }, { "name": "info", "description": "Информация об API" } ] ) # ============================================================================= # НАСТРОЙКА CORS (Cross-Origin Resource Sharing) # ============================================================================= # CORS middleware позволяет веб-приложениям делать запросы к API с других доменов app.add_middleware( CORSMiddleware, allow_origins=["*"], # В продакшене следует ограничить домены для безопасности allow_credentials=True, # Разрешаем отправку cookies и авторизационных заголовков allow_methods=["*"], # Разрешаем все HTTP методы (GET, POST, PUT, DELETE и т.д.) allow_headers=["*"], # Разрешаем все заголовки ) # ============================================================================= # API ЭНДПОИНТЫ # ============================================================================= @app.get("/", tags=["info"]) async def root(): """ ## Корневой endpoint Возвращает основную информацию об API и доступных эндпоинтах. Используется для получения базовой информации о сервисе. ### Возвращает: - Название API - Версию - Список доступных эндпоинтов - Ссылки на документацию """ return { "message": "Lichess Statistics API", "version": "1.0.0", "description": "REST API для получения статистики игроков Lichess", "endpoints": { "today": "/stats/{username}/today", # Статистика за сегодня "yesterday": "/stats/{username}/yesterday", # Статистика за вчера "week": "/stats/{username}/week", # Статистика за неделю "games_period": "/games/{username}/period?since={timestamp}&until={timestamp}", # Статистика игр за период "puzzle_period": "/puzzle/period?since={timestamp}&until={timestamp}&max={max}" # Статистика задач за период }, "documentation": "/docs", # Swagger UI документация "openapi_schema": "/openapi.json" # OpenAPI схема } @app.get("/health", response_model=HealthResponse, tags=["health"], summary="Health Check", description="Проверка состояния сервиса") async def health_check(): """ ## Проверка здоровья сервиса Простой endpoint для проверки работоспособности API. Используется для мониторинга и health checks в production среде. ### Возвращает: - Статус сервиса (healthy/unhealthy) - Время проверки в ISO формате - Название сервиса ### Использование: - Мониторинг системы - Load balancer health checks - Kubernetes liveness/readiness probes """ return HealthResponse( status="healthy", # Всегда возвращает healthy, если сервис запущен timestamp=datetime.now().isoformat(), # Текущее время в ISO формате service="Lichess Statistics API" ) @app.get("/stats/{username}/today", response_model=ActivityResponse, tags=["statistics"], summary="Статистика за сегодня", description="Получает детальную статистику игрока за сегодняшний день", responses={ 200: { "description": "Статистика успешно получена", "content": { "application/json": { "example": { "message": "Статистика за сегодняшний день", "data": { "username": "magnus", "tasks": { "total": 15, "solved": 12, "unsolved": 3 }, "games": { "bullet": { "games_played": 8, "rating_change": 15, "final_rating": 2850, "wins": 5, "losses": 2, "draws": 1 }, "blitz": { "games_played": 3, "rating_change": -5, "final_rating": 2750, "wins": 1, "losses": 2, "draws": 0 }, "rapid": { "games_played": 0, "rating_change": 0, "final_rating": 0, "wins": 0, "losses": 0, "draws": 0 } } } } } } }, 404: { "description": "Пользователь не найден", "content": { "application/json": { "example": { "message": "Пользователь magnus не найден или неактивен" } } } }, 500: { "description": "Внутренняя ошибка сервера", "content": { "application/json": { "example": { "detail": "Внутренняя ошибка сервера: Connection timeout" } } } } }) async def get_today_stats( username: str = Path(..., description="Имя пользователя на Lichess", example="magnus", min_length=1, max_length=50) ): """ ## Статистика за сегодняшний день Получает детальную статистику игрока за сегодняшний день, включая: ### Статистика игр: - **Bullet**: Быстрые игры (1-3 минуты) - **Blitz**: Блиц игры (3-10 минут) - **Rapid**: Рапид игры (10+ минут) ### Для каждого режима: - Количество сыгранных игр - Изменение рейтинга - Текущий рейтинг - Количество побед, поражений, ничьих ### Статистика задач: - Общее количество решенных задач - Количество решенных задач - Количество нерешенных задач ### Параметры: - **username**: Имя пользователя на Lichess (обязательно) ### Возможные ошибки: - **404**: Пользователь не найден или неактивен - **500**: Внутренняя ошибка сервера """ # Проверяем, что сервис статистики инициализирован if not stats_service: raise HTTPException(status_code=500, detail="Сервис не инициализирован") try: # Получаем статистику за сегодняшний день через сервис result = await stats_service.get_today_stats(username) return result except Exception as e: # Логируем ошибку и возвращаем HTTP 500 logger.error(f"Ошибка в endpoint get_today_stats: {e}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") @app.get("/stats/{username}/yesterday", response_model=ActivityResponse, tags=["statistics"], summary="Статистика за вчера", description="Получает детальную статистику игрока за вчерашний день") async def get_yesterday_stats( username: str = Path(..., description="Имя пользователя на Lichess", example="magnus", min_length=1, max_length=50) ): """ ## Статистика за вчерашний день Получает детальную статистику игрока за вчерашний день. ### Возвращает: - Статистику игр по всем режимам (Bullet, Blitz, Rapid) - Статистику решения задач (пазлов) - Изменения рейтинга - Результаты игр (победы, поражения, ничьи) ### Параметры: - **username**: Имя пользователя на Lichess (обязательно) """ if not stats_service: raise HTTPException(status_code=500, detail="Сервис не инициализирован") try: result = await stats_service.get_yesterday_stats(username) return result except Exception as e: logger.error(f"Ошибка в endpoint get_yesterday_stats: {e}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") @app.get("/stats/{username}/week", response_model=ActivityResponse, tags=["statistics"], summary="Статистика за неделю", description="Получает агрегированную статистику игрока за последние 7 дней") async def get_week_stats( username: str = Path(..., description="Имя пользователя на Lichess", example="magnus", min_length=1, max_length=50) ): """ ## Статистика за последние 7 дней Получает агрегированную статистику игрока за последние 7 дней. ### Особенности: - Суммирует все игры и задачи за неделю - Показывает общее изменение рейтинга - Отображает финальный рейтинг на конец периода ### Возвращает: - Общую статистику игр по всем режимам - Суммарную статистику решения задач - Агрегированные изменения рейтинга ### Параметры: - **username**: Имя пользователя на Lichess (обязательно) """ if not stats_service: raise HTTPException(status_code=500, detail="Сервис не инициализирован") try: result = await stats_service.get_week_stats(username) return result except Exception as e: logger.error(f"Ошибка в endpoint get_week_stats: {e}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") @app.get("/games/{username}/period", response_model=GamesOfPeriodResponse, tags=["statistics"], summary="Статистика игр за период", description="Получает детальную статистику игр пользователя за указанный период", responses={ 200: { "description": "Статистика игр успешно получена", "content": { "application/json": { "example": { "message": "Статистика игр за период", "username": "magnus", "period_start": 1640995200000, "period_end": 1641081600000, "games_count": 25, "data": { "bullet": { "games_played": 10, "wins": 6, "losses": 3, "draws": 1, "rating_change": 15, "rating": 2850 }, "blitz": { "games_played": 8, "wins": 5, "losses": 2, "draws": 1, "rating_change": 12, "rating": 2750 }, "rapid": { "games_played": 5, "wins": 3, "losses": 1, "draws": 1, "rating_change": 8, "rating": 2600 }, "classical": { "games_played": 2, "wins": 1, "losses": 1, "draws": 0, "rating_change": 0, "rating": 2400 }, "correspondence": { "games_played": 0, "wins": 0, "losses": 0, "draws": 0, "rating_change": 0, "rating": None }, "total": { "games_played": 25, "wins": 15, "losses": 7, "draws": 3, "rating_change": 35, "rating": 2850 } } } } } }, 400: { "description": "Некорректные параметры запроса", "content": { "application/json": { "example": { "detail": "Параметр 'since' должен быть меньше 'until'" } } } }, 404: { "description": "Пользователь не найден", "content": { "application/json": { "example": { "message": "Пользователь magnus не найден" } } } }, 500: { "description": "Внутренняя ошибка сервера", "content": { "application/json": { "example": { "detail": "Внутренняя ошибка сервера: Connection timeout" } } } } }) async def get_games_of_period( username: str = Path(..., description="Имя пользователя на Lichess", example="magnus", min_length=1, max_length=50), since: int = Query(..., description="Начало периода (Unix timestamp в миллисекундах)", example=1640995200000), until: int = Query(..., description="Конец периода (Unix timestamp в миллисекундах)", example=1641081600000), rated_only: bool = Query(True, description="Только рейтинговые игры (по умолчанию true - рекомендуется)", example=True) ): """ ## Статистика игр за период Получает детальную статистику игр пользователя за указанный период времени. Этот эндпоинт позволяет получить подробную аналитику по играм за любой период. ### Возможности: - **Фильтрация по времени**: точный период с Unix timestamp - **Типы игр**: Bullet, Blitz, Rapid, Classical, Correspondence - **Статистика результатов**: победы, поражения, ничьи - **Рейтинговые изменения**: суммарные изменения рейтинга - **Итоговый рейтинг**: рейтинг после последней игры в каждом режиме - **Фильтр рейтинговых игр**: только рейтинговые или все игры ### Параметры: - **username**: Имя пользователя на Lichess (обязательно) - **since**: Начало периода в Unix timestamp (миллисекунды) (обязательно) - **until**: Конец периода в Unix timestamp (миллисекунды) (обязательно) - **rated_only**: Только рейтинговые игры (по умолчанию true - рекомендуется) ### Примеры использования: - Статистика за последние 7 дней: `since=1640995200000&until=1641081600000` (по умолчанию только рейтинговые) - Только рейтинговые игры: `rated_only=true` (рекомендуется) - Все игры: `rated_only=false` ### Возможные ошибки: - **400**: Некорректные параметры (since >= until) - **404**: Пользователь не найден - **500**: Внутренняя ошибка сервера """ # Проверяем, что сервис статистики инициализирован if not stats_service: raise HTTPException(status_code=500, detail="Сервис не инициализирован") # Валидация параметров времени if since >= until: raise HTTPException( status_code=400, detail="Параметр 'since' должен быть меньше 'until'" ) # Проверяем разумность периода (не более 1 года в миллисекундах) if until - since > 365 * 24 * 3600 * 1000: raise HTTPException( status_code=400, detail="Период не может превышать 1 год" ) try: # Конвертируем миллисекунды в секунды для внутренней логики since_seconds = since // 1000 until_seconds = until // 1000 result = await stats_service.get_games_of_period(username, since_seconds, until_seconds, rated_only) return result except Exception as e: logger.error(f"Ошибка в endpoint get_games_of_period: {e}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") @app.get("/puzzle/period", response_model=PuzzleOfPeriodResponse, tags=["statistics"], summary="Статистика решения задач за период", description="Получает статистику решения задач (пазлов) за указанный период времени. Требует авторизации через Bearer токен.", responses={ 200: { "description": "Статистика решения задач успешно получена", "content": { "application/json": { "example": { "message": "Статистика решения задач за период", "period_start": 1640995200000, "period_end": 1641081600000, "max_puzzles": 50, "puzzles_in_period": 15, "data": { "total_attempts": 15, "solved": 12, "failed": 3, "success_rate": 80.0 } } } } }, 400: { "description": "Некорректные параметры запроса", "content": { "application/json": { "example": { "detail": "Параметр 'since' должен быть меньше 'until'" } } } }, 401: { "description": "Неверный токен авторизации", "content": { "application/json": { "example": { "message": "Неверный токен авторизации или доступ запрещен" } } } }, 500: { "description": "Внутренняя ошибка сервера", "content": { "application/json": { "example": { "detail": "Внутренняя ошибка сервера: Connection timeout" } } } } }) async def get_puzzle_of_period( since: int = Query(..., description="Начало периода (Unix timestamp в миллисекундах)", example=1640995200000), until: int = Query(..., description="Конец периода (Unix timestamp в миллисекундах)", example=1641081600000), max: int = Query(50, description="Максимальное количество задач для получения от Lichess API. Внимание: если в периоде было больше задач, чем указано в max, будут показаны только последние N активностей", example=50, ge=1, le=1000), authorization: str = Header(..., description="Bearer токен авторизации", example="Bearer your_token_here") ): """ ## Статистика решения задач за период Получает статистику решения задач (пазлов) за указанный период времени. Требует авторизации через Bearer токен от Lichess. ### Возможности: - **Фильтрация по времени**: точный период с Unix timestamp в миллисекундах - **Статистика решений**: количество решенных и нерешенных задач - **Процент успеха**: автоматический расчет процента успешных решений - **Настраиваемый лимит**: максимальное количество задач для анализа ### Параметры: - **since**: Начало периода в Unix timestamp (миллисекунды) (обязательно) - **until**: Конец периода в Unix timestamp (миллисекунды) (обязательно) - **max**: Максимальное количество задач (по умолчанию 50, максимум 1000) - **Authorization**: Bearer токен в заголовке (обязательно) ### Примеры использования: - Статистика за последние 7 дней: `since=1640995200000&until=1641081600000` - Больше задач: `max=100` - Заголовок: `Authorization: Bearer your_token_here` ### Получение токена: 1. Зайдите на https://lichess.org/account/oauth/token/create 2. Создайте новый токен с правами на чтение активности 3. Используйте токен в заголовке Authorization ### Возможные ошибки: - **400**: Некорректные параметры (since >= until, неверный max) - **401**: Неверный токен авторизации - **500**: Внутренняя ошибка сервера """ # Проверяем, что сервис статистики инициализирован if not stats_service: raise HTTPException(status_code=500, detail="Сервис не инициализирован") # Валидация параметров if since >= until: raise HTTPException( status_code=400, detail="Параметр 'since' должен быть меньше 'until'" ) # Проверяем разумность периода (не более 1 года в миллисекундах) if until - since > 365 * 24 * 3600 * 1000: raise HTTPException( status_code=400, detail="Период не может превышать 1 год" ) # Извлекаем токен из заголовка Authorization if not authorization.startswith("Bearer "): raise HTTPException( status_code=401, detail="Неверный формат токена. Используйте 'Bearer your_token'" ) token = authorization[7:] # Убираем "Bearer " из начала try: result = await stats_service.get_puzzle_of_period(token, since, until, max) return result except Exception as e: logger.error(f"Ошибка в endpoint get_puzzle_of_period: {e}") raise HTTPException(status_code=500, detail=f"Внутренняя ошибка сервера: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)