app.py - task2 - tonmalanin/bd

app.py
414 строк12 KB
Антон Маланьин
task2 done
01 апр 2026, 22:20
01 апр 2026, 22:20c98d986
import os
from contextlib import asynccontextmanager
from typing import Optional, List, Dict, Any, Tuple
from datetime import datetime

from fastapi import FastAPI, HTTPException, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, validator, field_validator
from valkey.cluster import ValkeyCluster, ClusterNode
from valkey.exceptions import ConnectionError, TimeoutError

VALKEY_NODES = os.getenv("VALKEY_NODES", "localhost:6379,localhost:6380,localhost:6381")
API_PORT = 8000

class PlayerRegistration(BaseModel):
    player_id: str = Field(
        ..., 
        min_length=1, 
        max_length=100,
        description="Уникальный идентификатор игрока",
        example="warrior_123"
    )
    initial_score: int = Field(
        0,
        ge=0,
        le=1000000,
        description="Начальный счет игрока (от 0 до 1,000,000)",
        example=1500
    )
    
    @field_validator("player_id")
    @classmethod
    def validate_player_id(cls, v: str) -> str:
        v = v.strip()
        if " " in v:
            raise ValueError("player_id не может содержать пробелы")
        return v


class PlayerScore(BaseModel):
    player_id: str = Field(
        ...,
        min_length=1,
        max_length=100,
        description="ID игрока",
        example="warrior_123"
    )
    score: int = Field(
        ...,
        ge=0,
        le=1000000,
        description="Новый счет игрока",
        example=2450
    )
    
    @field_validator("player_id")
    @classmethod
    def validate_player_id(cls, v: str) -> str:
        v = v.strip()
        if " " in v:
            raise ValueError("player_id не может содержать пробелы")
        return v


class PlayerInfo(BaseModel):
    player_id: str = Field(..., description="ID игрока")
    score: int = Field(..., description="Текущий счет")
    rank: Optional[int] = Field(None, description="Позиция в рейтинге (1-индексация)")


class LeaderboardResponse(BaseModel):
    top_players: List[PlayerInfo] = Field(..., description="Список топ игроков")
    total_players: int = Field(..., description="Общее количество игроков в лидерборде")
    timestamp: str = Field(..., description="Время запроса")


class PlayerRankResponse(BaseModel):
    player_id: str = Field(..., description="ID игрока")
    score: int = Field(..., description="Текущий счет")
    rank: int = Field(..., description="Позиция в рейтинге (1-индексация)")
    total_players: int = Field(..., description="Общее количество игроков")
    better_than_percent: float = Field(..., description="Процент игроков, которых опережает")


class ScoreUpdateResponse(BaseModel):
    status: str = Field(..., description="Статус операции")
    player_id: str = Field(..., description="ID игрока")
    old_score: Optional[int] = Field(None, description="Предыдущий счет")
    new_score: int = Field(..., description="Новый счет")
    rank: int = Field(..., description="Новая позиция в рейтинге")
    message: str = Field(..., description="Сообщение")


class ErrorResponse(BaseModel):
    error: str = Field(..., description="Сообщение об ошибке")
    detail: Optional[str] = Field(None, description="Детали ошибки")
    timestamp: str = Field(..., description="Время ошибки")

class LeaderboardService:

    LEADERBOARD_KEY = "arena:leaderboard" 
    
    def __init__(self, redis_client: ValkeyCluster):
        self.redis = redis_client
    
    async def register_or_update(self, player_id: str, score: int) -> Dict[str, Any]:
        old_score = self.redis.zscore(self.LEADERBOARD_KEY, player_id)
        
        self.redis.zadd(self.LEADERBOARD_KEY, {player_id: score})
        
        rank = self.get_player_rank(player_id)
        
        return {
            "status": "success",
            "player_id": player_id,
            "old_score": int(old_score) if old_score else None,
            "new_score": score,
            "rank": rank,
            "message": f"Игрок {player_id} обновлен до {score} очков"
        }
    
    async def get_top_players(self, limit: int) -> List[Tuple[str, int]]:
        top = self.redis.zrevrange(
            self.LEADERBOARD_KEY, 
            0, 
            limit - 1, 
            withscores=True
        )
        return [(player, int(score)) for player, score in top]
    
    def get_player_rank(self, player_id: str) -> int:
        rank = self.redis.zrevrank(self.LEADERBOARD_KEY, player_id)
        if rank is not None:
            return rank + 1 
        return 0
    
    async def get_player_info(self, player_id: str) -> Optional[Dict[str, Any]]:
        score = self.redis.zscore(self.LEADERBOARD_KEY, player_id)
        if score is None:
            return None
        
        rank = self.get_player_rank(player_id)
        total_players = self.redis.zcard(self.LEADERBOARD_KEY)
        
        better_than_percent = 0
        if total_players > 1:
            better_than_percent = (total_players - rank) / (total_players - 1) * 100
        
        return {
            "player_id": player_id,
            "score": int(score),
            "rank": rank,
            "total_players": total_players,
            "better_than_percent": round(better_than_percent, 2)
        }
    
    async def get_total_players(self) -> int:
        return self.redis.zcard(self.LEADERBOARD_KEY)
    
    async def delete_player(self, player_id: str) -> bool:
        removed = self.redis.zrem(self.LEADERBOARD_KEY, player_id)
        return bool(removed)

class ValkeyClusterManager:
	
	def __init__(self):
		self.client: Optional[ValkeyCluster] = None
		self.startup_nodes: List[ClusterNode] = []
		for node in VALKEY_NODES.split(","):
			host, port = node.split(":")
			self.startup_nodes.append(ClusterNode(host, int(port)))
	
	def connect(self) -> ValkeyCluster: 
		self.client = ValkeyCluster(
			startup_nodes=self.startup_nodes,
			decode_responses=True,           
			max_connections=32,              
			max_connections_per_node=10,     
			connection_timeout=10,           
			socket_timeout=10,               
			socket_connect_timeout=10,        
			retry_on_timeout=True,           
			retry_on_error=[ConnectionError, TimeoutError],  
			health_check_interval=30,       
		)

		self.client.ping()
		
		return self.client

	def get_client(self) -> ValkeyCluster:
		return self.client
	
	def close(self):
		if self.client:
			self.client.close()
		self.client = None

valkey_manager = ValkeyClusterManager()
leaderboard_service = None

@asynccontextmanager
async def lifespan(app: FastAPI):
	global leaderboard_service

	client = valkey_manager.connect()
	leaderboard_service = LeaderboardService(client)
	
	yield  
	
	valkey_manager.close()

app = FastAPI(
	title="Arena Legends Leaderboard API",
	description="Система лидерборда для игры Арена Легенд",
	lifespan=lifespan
)

@app.get("/", tags=["Health"])
async def health_check():
	try:
		client = valkey_manager.get_client()
		
		client.ping()
		
		cluster_info = client.cluster_info()
		
		return {
			"status": "healthy",
			"service": "Arena Legends Leaderboard",
			"valkey_cluster": {
				"state": cluster_info.get("cluster_state"),
				"known_nodes": int(cluster_info.get("cluster_known_nodes", 0)),
				"slots_assigned": int(cluster_info.get("cluster_slots_assigned", 0)),
				"slots_ok": int(cluster_info.get("cluster_slots_ok", 0))
			}
		}
		
	except Exception as e:
		return {
			"status": "unhealthy",
			"error": str(e)
		}


@app.get("/cluster/nodes", tags=["Health"])
async def cluster_nodes():
    try:
        client = valkey_manager.get_client()
        
        nodes_info = client.cluster_nodes()
        
        cluster_info = client.cluster_info()
        
        return {
            "total_nodes": len(nodes_info),
            "cluster_state": cluster_info.get("cluster_state"),
            "cluster_size": int(cluster_info.get("cluster_size", 0)),
            "nodes": nodes_info
        }
        
    except Exception as e:
        return {
            "error": str(e),
            "type": type(e).__name__
        }

@app.post(
    "/players",
    response_model=ScoreUpdateResponse,
    status_code=status.HTTP_201_CREATED,
    tags=["Players"]
)
async def register_player(registration: PlayerRegistration):
    try:
        result = await leaderboard_service.register_or_update(
            registration.player_id,
            registration.initial_score
        )
        return result
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Ошибка регистрации: {str(e)}"
        )


@app.put(
    "/players/{player_id}/score",
    response_model=ScoreUpdateResponse,
    tags=["Players"]
)
async def update_player_score(player_id: str, score_data: PlayerScore):
    if player_id != score_data.player_id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="player_id в URL и в теле запроса не совпадают"
        )
    
    try:
        result = await leaderboard_service.register_or_update(
            player_id,
            score_data.score
        )
        return result
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Ошибка обновления: {str(e)}"
        )


@app.get(
    "/leaderboard",
    response_model=LeaderboardResponse,
    tags=["Leaderboard"]
)
async def get_leaderboard(limit: int = 10):
    if limit < 1:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="limit должен быть >= 1"
        )
    if limit > 1000:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="limit не может превышать 1000"
        )
    
    try:
        top_players = await leaderboard_service.get_top_players(limit)
        total_players = await leaderboard_service.get_total_players()
        
        players_info = []
        for idx, (player_id, score) in enumerate(top_players, 1):
            players_info.append(PlayerInfo(
                player_id=player_id,
                score=score,
                rank=idx
            ))
        
        return LeaderboardResponse(
            top_players=players_info,
            total_players=total_players,
            timestamp=datetime.now().isoformat()
        )
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Ошибка получения лидерборда: {str(e)}"
        )


@app.get(
    "/players/{player_id}/rank",
    response_model=PlayerRankResponse,
    tags=["Players"]
)
async def get_player_rank(player_id: str):
    try:
        player_info = await leaderboard_service.get_player_info(player_id)
        
        if player_info is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"Игрок {player_id} не найден"
            )
        
        return PlayerRankResponse(**player_info)
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Ошибка получения информации: {str(e)}"
        )


@app.delete(
    "/players/{player_id}",
    tags=["Players"]
)
async def delete_player(player_id: str):
    try:
        deleted = await leaderboard_service.delete_player(player_id)
        
        if not deleted:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"Игрок {player_id} не найден"
            )
        
        return {
            "status": "success",
            "player_id": player_id,
            "message": f"Игрок {player_id} удален"
        }
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Ошибка удаления: {str(e)}"
        )

if __name__ == "__main__":
	import uvicorn
	
	uvicorn.run(
		"app:app",
		host="0.0.0.0",
		port=API_PORT,
		reload=True,
	)