548 lines
16 KiB
Python
548 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mice Game Score API
|
|
FastAPI application with SQLite database for user and score management
|
|
"""
|
|
|
|
from fastapi import FastAPI, HTTPException, Depends
|
|
from fastapi.responses import JSONResponse
|
|
import sqlite3
|
|
import os
|
|
from datetime import datetime
|
|
from typing import List, Optional
|
|
from pydantic import BaseModel
|
|
import hashlib
|
|
import uuid
|
|
|
|
|
|
# Pydantic models for request/response
|
|
class User(BaseModel):
|
|
user_id: str
|
|
device_id: str
|
|
created_at: Optional[str] = None
|
|
last_active: Optional[str] = None
|
|
|
|
class Score(BaseModel):
|
|
user_id: str
|
|
device_id: str
|
|
score: int
|
|
game_completed: bool = True
|
|
timestamp: Optional[str] = None
|
|
|
|
class ScoreResponse(BaseModel):
|
|
id: int
|
|
user_id: str
|
|
device_id: str
|
|
score: int
|
|
game_completed: bool
|
|
timestamp: str
|
|
|
|
class UserResponse(BaseModel):
|
|
user_id: str
|
|
device_id: str
|
|
created_at: str
|
|
last_active: str
|
|
total_scores: int
|
|
best_score: int
|
|
|
|
|
|
# Database setup
|
|
DATABASE_FILE = "mice_game.db"
|
|
|
|
def init_database():
|
|
"""Initialize the SQLite database with required tables"""
|
|
conn = sqlite3.connect(DATABASE_FILE)
|
|
cursor = conn.cursor()
|
|
|
|
# Create users table
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id TEXT NOT NULL,
|
|
device_id TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
last_active TEXT NOT NULL,
|
|
UNIQUE(user_id, device_id)
|
|
)
|
|
''')
|
|
|
|
# Create scores table
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS scores (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id TEXT NOT NULL,
|
|
device_id TEXT NOT NULL,
|
|
score INTEGER NOT NULL,
|
|
game_completed BOOLEAN NOT NULL DEFAULT 1,
|
|
timestamp TEXT NOT NULL,
|
|
FOREIGN KEY (user_id, device_id) REFERENCES users (user_id, device_id)
|
|
)
|
|
''')
|
|
|
|
# Create indexes for better performance
|
|
cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_device ON users(device_id)')
|
|
cursor.execute('CREATE INDEX IF NOT EXISTS idx_scores_user_device ON scores(user_id, device_id)')
|
|
cursor.execute('CREATE INDEX IF NOT EXISTS idx_scores_timestamp ON scores(timestamp)')
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_db_connection():
|
|
"""Get database connection"""
|
|
if not os.path.exists(DATABASE_FILE):
|
|
init_database()
|
|
|
|
conn = sqlite3.connect(DATABASE_FILE)
|
|
conn.row_factory = sqlite3.Row # Enable column access by name
|
|
return conn
|
|
|
|
def close_db_connection(conn):
|
|
"""Close database connection"""
|
|
conn.close()
|
|
|
|
|
|
# FastAPI app
|
|
app = FastAPI(
|
|
title="Mice Game Score API",
|
|
description="API for managing users and scores in the Mice game",
|
|
version="1.0.0"
|
|
)
|
|
|
|
# Initialize database on startup
|
|
@app.on_event("startup")
|
|
def startup_event():
|
|
init_database()
|
|
|
|
|
|
# Utility functions
|
|
def validate_device_id(device_id: str) -> bool:
|
|
"""Validate device ID format"""
|
|
return device_id.startswith("DEV-") and len(device_id) == 12
|
|
|
|
def validate_user_id(user_id: str) -> bool:
|
|
"""Validate user ID format"""
|
|
return len(user_id.strip()) > 0 and len(user_id) <= 50
|
|
|
|
def user_exists(conn, user_id: str, device_id: str) -> bool:
|
|
"""Check if user exists in database"""
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT 1 FROM users WHERE user_id = ? AND device_id = ?",
|
|
(user_id, device_id)
|
|
)
|
|
return cursor.fetchone() is not None
|
|
|
|
def user_id_unique_globally(conn, user_id: str) -> bool:
|
|
"""Check if user_id is globally unique (across all devices)"""
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT 1 FROM users WHERE user_id = ?", (user_id,))
|
|
return cursor.fetchone() is None
|
|
|
|
|
|
# API Endpoints
|
|
|
|
@app.get("/")
|
|
def read_root():
|
|
"""API root endpoint"""
|
|
return {
|
|
"message": "Mice Game Score API",
|
|
"version": "1.0.0",
|
|
"endpoints": [
|
|
"GET /users/{device_id} - Get all users for device",
|
|
"POST /signup/{device_id}/{user_id} - Register new user",
|
|
"POST /score/{device_id}/{user_id} - Submit score"
|
|
]
|
|
}
|
|
|
|
@app.post("/signup/{device_id}/{user_id}")
|
|
def signup_user(device_id: str, user_id: str):
|
|
"""
|
|
Register a new user for a device
|
|
- device_id: Device identifier (format: DEV-XXXXXXXX)
|
|
- user_id: Unique user identifier
|
|
"""
|
|
# Validate inputs
|
|
if not validate_device_id(device_id):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Invalid device_id format. Must be 'DEV-XXXXXXXX'"
|
|
)
|
|
|
|
if not validate_user_id(user_id):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Invalid user_id. Must be 1-50 characters long"
|
|
)
|
|
|
|
conn = get_db_connection()
|
|
|
|
try:
|
|
# Check if user_id is globally unique
|
|
if not user_id_unique_globally(conn, user_id):
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"User ID '{user_id}' already exists. Choose a different user ID."
|
|
)
|
|
|
|
# Check if user already exists for this device
|
|
if user_exists(conn, user_id, device_id):
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"User '{user_id}' already registered for device '{device_id}'"
|
|
)
|
|
|
|
# Register the user
|
|
cursor = conn.cursor()
|
|
now = datetime.now().isoformat()
|
|
|
|
cursor.execute(
|
|
"INSERT INTO users (user_id, device_id, created_at, last_active) VALUES (?, ?, ?, ?)",
|
|
(user_id, device_id, now, now)
|
|
)
|
|
|
|
conn.commit()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": f"User '{user_id}' successfully registered for device '{device_id}'",
|
|
"user": {
|
|
"user_id": user_id,
|
|
"device_id": device_id,
|
|
"created_at": now
|
|
}
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
@app.post("/score/{device_id}/{user_id}")
|
|
def submit_score(device_id: str, user_id: str, score_data: Score):
|
|
"""
|
|
Submit a score for a registered user
|
|
- device_id: Device identifier
|
|
- user_id: User identifier
|
|
- score_data: Score information (score, game_completed)
|
|
"""
|
|
# Validate inputs
|
|
if not validate_device_id(device_id):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Invalid device_id format. Must be 'DEV-XXXXXXXX'"
|
|
)
|
|
|
|
if not validate_user_id(user_id):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Invalid user_id"
|
|
)
|
|
|
|
if score_data.score < 0:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Score must be non-negative"
|
|
)
|
|
|
|
conn = get_db_connection()
|
|
|
|
try:
|
|
# Check if user exists and is registered for this device
|
|
if not user_exists(conn, user_id, device_id):
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"User '{user_id}' not registered for device '{device_id}'. Please signup first."
|
|
)
|
|
|
|
# Add the score
|
|
cursor = conn.cursor()
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
cursor.execute(
|
|
"INSERT INTO scores (user_id, device_id, score, game_completed, timestamp) VALUES (?, ?, ?, ?, ?)",
|
|
(user_id, device_id, score_data.score, score_data.game_completed, timestamp)
|
|
)
|
|
|
|
# Update user's last_active timestamp
|
|
cursor.execute(
|
|
"UPDATE users SET last_active = ? WHERE user_id = ? AND device_id = ?",
|
|
(timestamp, user_id, device_id)
|
|
)
|
|
|
|
conn.commit()
|
|
|
|
# Get user's statistics
|
|
cursor.execute(
|
|
"""
|
|
SELECT
|
|
COUNT(*) as total_games,
|
|
MAX(score) as best_score,
|
|
AVG(score) as avg_score,
|
|
SUM(score) as total_score
|
|
FROM scores
|
|
WHERE user_id = ? AND device_id = ?
|
|
""",
|
|
(user_id, device_id)
|
|
)
|
|
|
|
stats = cursor.fetchone()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": f"Score {score_data.score} submitted successfully",
|
|
"score_id": cursor.lastrowid,
|
|
"user_stats": {
|
|
"user_id": user_id,
|
|
"device_id": device_id,
|
|
"total_games": stats["total_games"],
|
|
"best_score": stats["best_score"],
|
|
"average_score": round(stats["avg_score"], 2) if stats["avg_score"] else 0,
|
|
"total_score": stats["total_score"]
|
|
}
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
@app.get("/users/{device_id}")
|
|
def get_device_users(device_id: str) -> List[UserResponse]:
|
|
"""
|
|
Get all registered users for a specific device
|
|
- device_id: Device identifier
|
|
"""
|
|
# Validate device_id
|
|
if not validate_device_id(device_id):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Invalid device_id format. Must be 'DEV-XXXXXXXX'"
|
|
)
|
|
|
|
conn = get_db_connection()
|
|
|
|
try:
|
|
cursor = conn.cursor()
|
|
|
|
# Get users with their statistics
|
|
cursor.execute(
|
|
"""
|
|
SELECT
|
|
u.user_id,
|
|
u.device_id,
|
|
u.created_at,
|
|
u.last_active,
|
|
COUNT(s.id) as total_scores,
|
|
COALESCE(MAX(s.score), 0) as best_score
|
|
FROM users u
|
|
LEFT JOIN scores s ON u.user_id = s.user_id AND u.device_id = s.device_id
|
|
WHERE u.device_id = ?
|
|
GROUP BY u.user_id, u.device_id, u.created_at, u.last_active
|
|
ORDER BY u.last_active DESC
|
|
""",
|
|
(device_id,)
|
|
)
|
|
|
|
users = cursor.fetchall()
|
|
|
|
if not users:
|
|
return []
|
|
|
|
return [
|
|
UserResponse(
|
|
user_id=user["user_id"],
|
|
device_id=user["device_id"],
|
|
created_at=user["created_at"],
|
|
last_active=user["last_active"],
|
|
total_scores=user["total_scores"],
|
|
best_score=user["best_score"]
|
|
)
|
|
for user in users
|
|
]
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
# Additional useful endpoints
|
|
|
|
@app.get("/scores/{device_id}/{user_id}")
|
|
def get_user_scores(device_id: str, user_id: str, limit: int = 10) -> List[ScoreResponse]:
|
|
"""
|
|
Get recent scores for a user
|
|
- device_id: Device identifier
|
|
- user_id: User identifier
|
|
- limit: Maximum number of scores to return (default: 10)
|
|
"""
|
|
if not validate_device_id(device_id) or not validate_user_id(user_id):
|
|
raise HTTPException(status_code=400, detail="Invalid device_id or user_id")
|
|
|
|
if limit < 1 or limit > 100:
|
|
raise HTTPException(status_code=400, detail="Limit must be between 1 and 100")
|
|
|
|
conn = get_db_connection()
|
|
|
|
try:
|
|
# Check if user exists
|
|
if not user_exists(conn, user_id, device_id):
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"User '{user_id}' not found for device '{device_id}'"
|
|
)
|
|
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""
|
|
SELECT id, user_id, device_id, score, game_completed, timestamp
|
|
FROM scores
|
|
WHERE user_id = ? AND device_id = ?
|
|
ORDER BY timestamp DESC
|
|
LIMIT ?
|
|
""",
|
|
(user_id, device_id, limit)
|
|
)
|
|
|
|
scores = cursor.fetchall()
|
|
|
|
return [
|
|
ScoreResponse(
|
|
id=score["id"],
|
|
user_id=score["user_id"],
|
|
device_id=score["device_id"],
|
|
score=score["score"],
|
|
game_completed=bool(score["game_completed"]),
|
|
timestamp=score["timestamp"]
|
|
)
|
|
for score in scores
|
|
]
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
@app.get("/leaderboard/{device_id}")
|
|
def get_device_leaderboard(device_id: str, limit: int = 10):
|
|
"""
|
|
Get leaderboard for a device (top scores)
|
|
- device_id: Device identifier
|
|
- limit: Maximum number of entries to return (default: 10)
|
|
"""
|
|
if not validate_device_id(device_id):
|
|
raise HTTPException(status_code=400, detail="Invalid device_id")
|
|
|
|
if limit < 1 or limit > 100:
|
|
raise HTTPException(status_code=400, detail="Limit must be between 1 and 100")
|
|
|
|
conn = get_db_connection()
|
|
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""
|
|
SELECT
|
|
user_id,
|
|
MAX(score) as best_score,
|
|
COUNT(*) as total_games,
|
|
MAX(timestamp) as last_play
|
|
FROM scores
|
|
WHERE device_id = ?
|
|
GROUP BY user_id
|
|
ORDER BY best_score DESC, last_play DESC
|
|
LIMIT ?
|
|
""",
|
|
(device_id, limit)
|
|
)
|
|
|
|
leaderboard = cursor.fetchall()
|
|
|
|
return [
|
|
{
|
|
"rank": idx + 1,
|
|
"user_id": entry["user_id"],
|
|
"best_score": entry["best_score"],
|
|
"total_games": entry["total_games"],
|
|
"last_play": entry["last_play"]
|
|
}
|
|
for idx, entry in enumerate(leaderboard)
|
|
]
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
|
|
@app.get("/leaderboard/global/top")
|
|
def get_global_leaderboard(limit: int = 10):
|
|
"""
|
|
Get global leaderboard across all devices (absolute top scores)
|
|
- limit: Maximum number of entries to return (default: 10)
|
|
"""
|
|
if limit < 1 or limit > 100:
|
|
raise HTTPException(status_code=400, detail="Limit must be between 1 and 100")
|
|
|
|
conn = get_db_connection()
|
|
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"""
|
|
SELECT
|
|
user_id,
|
|
device_id,
|
|
MAX(score) as best_score,
|
|
COUNT(*) as total_games,
|
|
MAX(timestamp) as last_play
|
|
FROM scores
|
|
GROUP BY user_id, device_id
|
|
ORDER BY best_score DESC, last_play DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,)
|
|
)
|
|
|
|
leaderboard = cursor.fetchall()
|
|
|
|
return [
|
|
{
|
|
"rank": idx + 1,
|
|
"user_id": entry["user_id"],
|
|
"device_id": entry["device_id"],
|
|
"best_score": entry["best_score"],
|
|
"total_games": entry["total_games"],
|
|
"last_play": entry["last_play"]
|
|
}
|
|
for idx, entry in enumerate(leaderboard)
|
|
]
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
print("Starting Mice Game Score API...")
|
|
print("Available endpoints:")
|
|
print(" POST /signup/{device_id}/{user_id} - Register new user")
|
|
print(" POST /score/{device_id}/{user_id} - Submit score")
|
|
print(" GET /users/{device_id} - Get all users for device")
|
|
print(" GET /scores/{device_id}/{user_id} - Get user scores")
|
|
print(" GET /leaderboard/{device_id} - Get device leaderboard")
|
|
print()
|
|
|
|
# Initialize database
|
|
init_database()
|
|
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|