107 lines
4.2 KiB
Python
107 lines
4.2 KiB
Python
import sqlite3
|
||
from datetime import datetime
|
||
import os
|
||
|
||
class Database:
|
||
def __init__(self):
|
||
self.db_path = '/app/data/nard_dice.db'
|
||
self._init_db()
|
||
|
||
def _init_db(self):
|
||
"""Инициализация базы данных"""
|
||
if not os.path.exists(self.db_path):
|
||
os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
|
||
with sqlite3.connect(self.db_path) as conn:
|
||
with open('bot/schema.sql', 'r') as f:
|
||
conn.executescript(f.read())
|
||
|
||
def _get_connection(self):
|
||
"""Получение соединения с базой данных"""
|
||
conn = sqlite3.connect(self.db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
def create_game(self, user_id: int, username: str) -> int:
|
||
"""Создание новой игры"""
|
||
with self._get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO games (user_id, username)
|
||
VALUES (?, ?)
|
||
''', (user_id, username))
|
||
conn.commit()
|
||
return cursor.lastrowid
|
||
|
||
def add_throw(self, game_id: int, dice1: int, dice2: int, unused_points: int = 0):
|
||
"""Добавление броска с учетом неиспользованных очков"""
|
||
with self._get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO throws (game_id, dice1, dice2, unused_points)
|
||
VALUES (?, ?, ?, ?)
|
||
''', (game_id, dice1, dice2, unused_points))
|
||
conn.commit()
|
||
|
||
def get_active_game(self, user_id: int):
|
||
"""Получение активной игры пользователя"""
|
||
with self._get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
game = cursor.execute('''
|
||
SELECT id, user_id, username, start_time
|
||
FROM games
|
||
WHERE user_id = ? AND status = 'active'
|
||
ORDER BY start_time DESC
|
||
LIMIT 1
|
||
''', (user_id,)).fetchone()
|
||
|
||
if game:
|
||
return dict(game)
|
||
return None
|
||
|
||
def end_game(self, game_id: int):
|
||
"""Завершение игры"""
|
||
with self._get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
UPDATE games
|
||
SET status = 'completed', end_time = CURRENT_TIMESTAMP
|
||
WHERE id = ?
|
||
''', (game_id,))
|
||
conn.commit()
|
||
|
||
def get_statistics(self, user_id: int):
|
||
"""Получение статистики пользователя с учетом неиспользованных очков"""
|
||
with self._get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
stats = cursor.execute('''
|
||
SELECT
|
||
COUNT(*) as total_throws,
|
||
SUM(CASE
|
||
WHEN dice1 = dice2 THEN (dice1 + dice2) * 2
|
||
ELSE dice1 + dice2
|
||
END) as total_base_sum,
|
||
SUM(CASE
|
||
WHEN dice1 = dice2 THEN (dice1 + dice2) * 2 - unused_points
|
||
ELSE dice1 + dice2 - unused_points
|
||
END) as total_final_sum,
|
||
COUNT(DISTINCT game_id) as total_games,
|
||
SUM(unused_points) as total_unused
|
||
FROM throws t
|
||
JOIN games g ON t.game_id = g.id
|
||
WHERE g.user_id = ?
|
||
''', (user_id,)).fetchone()
|
||
|
||
return dict(stats)
|
||
|
||
def get_game_throws(self, game_id: int):
|
||
"""Получение всех бросков игры с учетом неиспользованных очков"""
|
||
with self._get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
throws = cursor.execute('''
|
||
SELECT dice1, dice2, unused_points, throw_time
|
||
FROM throws
|
||
WHERE game_id = ?
|
||
ORDER BY throw_time DESC
|
||
''', (game_id,)).fetchall()
|
||
|
||
return [dict(throw) for throw in throws] |