Files
bottohelp/services/game_service.py
T

874 lines
32 KiB
Python

"""
Сервис управления играми.
Отвечает за бизнес-логику игровых механик.
"""
import logging
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass
from datetime import datetime, timedelta
from core.exceptions import ValidationError
from utils.validators import InputValidator
import random
@dataclass
class GameSession:
"""Игровая сессия"""
game_id: str
game_type: str
player_id: int
chat_id: int
status: str = "active"
created_at: datetime = None
data: Dict[str, Any] = None
def __post_init__(self):
if self.created_at is None:
self.created_at = datetime.now()
if self.data is None:
self.data = {}
class GameService:
"""
Сервис для управления игровыми механиками.
Отвечает за:
- Управление игровыми сессиями
- Логику игр (камень-ножницы-бумага, викторина, etc.)
- Расчет очков за игры
- Статистику игровых достижений
"""
def __init__(self, user_repository, score_repository):
"""
Инициализация сервиса.
Args:
user_repository: Репозиторий пользователей
score_repository: Репозиторий очков и рейтингов
"""
self.user_repo = user_repository
self.score_repo = score_repository
self.logger = logging.getLogger(__name__)
# Настройки игр
self.game_settings = {
'rock_paper_scissors': {
'points_win': 5,
'points_draw': 1
},
'tic_tac_toe': {
'points_win': 15,
'points_draw': 3
},
'quiz': {
'points_correct': 5,
'max_questions': 10
},
'battleship': {
'points_win': 20,
'board_size': 5,
'ships': [3, 2, 1] # размеры кораблей
},
'game_2048': {
'points_win': 20,
'board_size': 4,
'win_value': 2048
},
'tetris': {
'points_per_line': 10,
'board_width': 10,
'board_height': 20
},
'snake': {
'points_per_food': 10,
'board_size': 10,
'initial_length': 3
}
}
# Вопросы для викторины
self.quiz_questions = [
{
'question': 'Столица Франции?',
'answers': ['Париж', 'Лондон', 'Берлин', 'Мадрид'],
'correct': 0
},
{
'question': 'Сколько континентов на Земле?',
'answers': ['5', '6', '7', '8'],
'correct': 2
},
{
'question': 'Какой газ составляет большую часть атмосферы?',
'answers': ['Кислород', 'Азот', 'Углекислый газ', 'Водород'],
'correct': 1
},
{
'question': 'В каком году был запущен первый искусственный спутник?',
'answers': ['1955', '1957', '1959', '1961'],
'correct': 1
}
]
# Активные игровые сессии
self.active_sessions = {}
def create_game_session(self, game_type: str, player_id: int, chat_id: int) -> GameSession:
"""
Создание новой игровой сессии.
Args:
game_type: Тип игры
player_id: ID игрока
chat_id: ID чата
Returns:
Игровая сессия
"""
if game_type not in self.game_settings:
raise ValidationError(f"Неизвестный тип игры: {game_type}")
import random
game_id = f"{game_type[:3]}_{random.randint(100000, 999999)}"
# Проверяем, что game_id не пустой
if not game_id:
raise ValidationError(f"Не удалось сгенерировать game_id для игры {game_type}")
self.logger.info(f"Создан game_id: {game_id} для игры {game_type}")
session = GameSession(
game_id=game_id,
game_type=game_type,
player_id=player_id,
chat_id=chat_id,
status="active"
)
# Инициализируем данные игры в зависимости от типа
if game_type == 'rock_paper_scissors':
session.data = {
'player_choice': None,
'bot_choice': None,
'result': None
}
elif game_type == 'tic_tac_toe':
session.data = {
'board': [' '] * 9,
'current_player': 'X',
'moves': 0,
'winner': None
}
elif game_type == 'quiz':
session.data = {
'current_question': 0,
'score': 0,
'questions': random.sample(self.quiz_questions, 3)
}
elif game_type == 'battleship':
session.data = {
'player_board': [['~'] * 5 for _ in range(5)],
'bot_board': [['~'] * 5 for _ in range(5)],
'bot_ships': self._place_battleship_ships(),
'player_shots': 0,
'player_hits': 0,
'game_status': 'active'
}
elif game_type == 'game_2048':
session.data = {
'board': [[0] * 4 for _ in range(4)],
'score': 0,
'moves': 0,
'game_status': 'active'
}
# Добавляем две начальные плитки
self._add_random_2048_tile(session.data['board'])
self._add_random_2048_tile(session.data['board'])
elif game_type == 'tetris':
session.data = {
'board': [[0] * 10 for _ in range(20)],
'current_piece': self._get_random_tetris_piece(),
'score': 0,
'lines_cleared': 0,
'level': 1,
'game_status': 'active'
}
elif game_type == 'snake':
session.data = {
'board': [[0] * 10 for _ in range(10)],
'snake': [(5, 5)],
'food': None,
'direction': 'right',
'score': 0,
'game_status': 'active'
}
# Размещаем первую еду
session.data['food'] = self._place_snake_food(session.data['board'], session.data['snake'])
self.active_sessions[game_id] = session
self.logger.info(f"Сессия сохранена в active_sessions: {repr(game_id)}, активные сессии: {list(self.active_sessions.keys())}")
return session
def get_game_session(self, game_id: str) -> Optional[GameSession]:
"""Получение игровой сессии по ID"""
self.logger.info(f"Поиск сессии для game_id: {repr(game_id)}, активные сессии: {list(self.active_sessions.keys())}")
session = self.active_sessions.get(game_id)
if not session:
# self.logger.error(f"Сессия не найдена для game_id: {repr(game_id)}, активные сессии: {list(self.active_sessions.keys())}")
return None
if session.status != "active":
self.logger.warning(f"Сессия {repr(game_id)} не активна (status: {session.status})")
return None
return session
def end_game_session(self, game_id: str) -> bool:
"""Завершение игровой сессии"""
self.logger.info(f"Завершение сессии для game_id: {repr(game_id)}, активные сессии: {list(self.active_sessions.keys())}")
if game_id in self.active_sessions:
self.active_sessions[game_id].status = "completed"
del self.active_sessions[game_id]
self.logger.info(f"Сессия {repr(game_id)} завершена и удалена")
return True
self.logger.warning(f"Сессия {repr(game_id)} не найдена для завершения")
return False
def play_rock_paper_scissors(self, game_id: str, player_choice: str) -> Dict[str, Any]:
"""
Игра в камень-ножницы-бумага.
Args:
game_id: ID игровой сессии
player_choice: Выбор игрока (rock, paper, scissors, или на русском: камень, ножницы, бумага)
Returns:
Результат игры
"""
session = self.get_game_session(game_id)
if not session or session.game_type != 'rock_paper_scissors':
raise ValidationError("Игра не найдена или неверный тип игры")
# Mapping для поддержки русского ввода
choice_mapping = {
'камень': 'rock',
'ножницы': 'scissors',
'бумага': 'paper',
'rock': 'rock',
'paper': 'paper',
'scissors': 'scissors'
}
normalized_choice = choice_mapping.get(player_choice.lower(), player_choice)
valid_choices = ['rock', 'paper', 'scissors']
if normalized_choice not in valid_choices:
raise ValidationError(f"Неверный выбор: {player_choice}. Допустимые: камень, ножницы, бумага или rock, paper, scissors")
# Определяем выбор бота
bot_choice = random.choice(valid_choices)
# Определяем результат
if normalized_choice == bot_choice:
result = 'draw'
points = self.game_settings['rock_paper_scissors']['points_draw']
elif (normalized_choice == 'rock' and bot_choice == 'scissors') or \
(normalized_choice == 'paper' and bot_choice == 'rock') or \
(normalized_choice == 'scissors' and bot_choice == 'paper'):
result = 'win'
points = self.game_settings['rock_paper_scissors']['points_win']
else:
result = 'lose'
points = 0
# Обновляем данные сессии
session.data.update({
'player_choice': normalized_choice,
'bot_choice': bot_choice,
'result': result
})
# Начисляем очки игроку
if points > 0:
# Здесь будет вызов score_repository для начисления очков
pass
return {
'result': result,
'player_choice': normalized_choice,
'bot_choice': bot_choice,
'points': points
}
def make_tic_tac_toe_move(self, game_id: str, position: int) -> Dict[str, Any]:
"""
Ход в крестики-нолики.
Args:
game_id: ID игровой сессии
position: Позиция хода (0-8)
Returns:
Результат хода
"""
session = self.get_game_session(game_id)
if not session or session.game_type != 'tic_tac_toe':
raise ValidationError("Игра не найдена или неверный тип игры")
if not (0 <= position <= 8):
raise ValidationError("Неверная позиция хода")
board = session.data['board']
if board[position] != ' ':
# Клетка уже занята, возвращаем текущее состояние
return {
'status': 'already_occupied',
'board': board,
'points': 0
}
# Ход игрока
board[position] = 'X'
session.data['moves'] += 1
session.data['board'] = board
# Проверяем победу игрока
winner = self._check_tic_tac_toe_winner(board)
if winner:
session.data['winner'] = 'player'
points = self.game_settings['tic_tac_toe']['points_win']
return {
'status': 'player_win',
'board': board,
'winner': 'player',
'points': points
}
# Проверяем ничью
if session.data['moves'] >= 9:
session.data['winner'] = 'draw'
points = self.game_settings['tic_tac_toe']['points_draw']
return {
'status': 'draw',
'board': board,
'winner': 'draw',
'points': points
}
# Ход бота (простая логика - случайный ход)
available_moves = [i for i, cell in enumerate(board) if cell == ' ']
if available_moves:
bot_position = random.choice(available_moves)
board[bot_position] = 'O'
session.data['moves'] += 1
session.data['board'] = board
# Проверяем победу бота
winner = self._check_tic_tac_toe_winner(board)
if winner:
session.data['winner'] = 'bot'
return {
'status': 'bot_win',
'board': board,
'winner': 'bot',
'points': 0
}
return {
'status': 'continue',
'board': board,
'points': 0
}
def answer_quiz_question(self, game_id: str, answer_index: int) -> Dict[str, Any]:
"""
Ответ на вопрос викторины.
Args:
game_id: ID игровой сессии
answer_index: Индекс ответа
Returns:
Результат ответа
"""
session = self.get_game_session(game_id)
if not session or session.game_type != 'quiz':
raise ValidationError("Игра не найдена или неверный тип игры")
current_q = session.data['current_question']
questions = session.data['questions']
if current_q >= len(questions):
raise ValidationError("Все вопросы отвечены")
question = questions[current_q]
if not (0 <= answer_index < len(question['answers'])):
raise ValidationError("Неверный индекс ответа")
# Проверяем ответ
is_correct = (answer_index == question['correct'])
if is_correct:
session.data['score'] += self.game_settings['quiz']['points_correct']
# Переходим к следующему вопросу
session.data['current_question'] += 1
current_q += 1
# Проверяем завершение игры
if current_q >= len(questions):
final_score = session.data['score']
return {
'status': 'game_over',
'correct': is_correct,
'final_score': final_score,
'points': final_score
}
return {
'status': 'continue',
'correct': is_correct,
'current_score': session.data['score']
}
def _check_tic_tac_toe_winner(self, board: List[str]) -> Optional[str]:
"""Проверка победителя в крестики-нолики"""
# Проверяем строки
for i in range(0, 9, 3):
if board[i] == board[i+1] == board[i+2] != ' ':
return board[i]
# Проверяем столбцы
for i in range(3):
if board[i] == board[i+3] == board[i+6] != ' ':
return board[i]
# Проверяем диагонали
if board[0] == board[4] == board[8] != ' ':
return board[0]
if board[2] == board[4] == board[6] != ' ':
return board[2]
return None
def get_game_statistics(self, user_id: int) -> Dict[str, Any]:
"""Получение игровой статистики пользователя"""
# Здесь будет получение статистики из базы данных
return {
'games_played': 0,
'games_won': 0,
'games_lost': 0,
'total_points': 0,
'favorite_game': None
}
# ===== МОРСКОЙ БОЙ =====
def make_battleship_shot(self, game_id: str, row: int, col: int) -> Dict[str, Any]:
"""Ход в морском бое"""
session = self.get_game_session(game_id)
if not session or session.game_type != 'battleship':
raise ValidationError("Игра не найдена или неверный тип игры")
data = session.data
bot_board = data['bot_board']
bot_ships = data['bot_ships']
if not (0 <= row < 5 and 0 <= col < 5):
raise ValidationError("Неверные координаты")
if bot_board[row][col] != '~':
# Уже стреляли, возвращаем текущее состояние
return {
'status': 'already_shot',
'board': bot_board,
'hit': False,
'shots': data['player_shots'],
'points': 0
}
# Определяем попадание
hit = False
for ship in bot_ships:
if (row, col) in ship:
hit = True
bot_board[row][col] = '💥'
data['player_hits'] += 1
break
if not hit:
bot_board[row][col] = '💧'
data['player_shots'] += 1
session.data = data
# Проверяем победу
total_ship_cells = sum(len(ship) for ship in bot_ships)
if data['player_hits'] >= total_ship_cells:
data['game_status'] = 'won'
return {
'status': 'win',
'board': bot_board,
'hit': hit,
'shots': data['player_shots'],
'points': self.game_settings['battleship']['points_win']
}
return {
'status': 'continue',
'board': bot_board,
'hit': hit,
'shots': data['player_shots'],
'points': 0
}
def _place_battleship_ships(self) -> List[List[Tuple[int, int]]]:
"""Размещение кораблей бота"""
ships = []
ship_sizes = self.game_settings['battleship']['ships'].copy()
for size in ship_sizes:
placed = False
attempts = 0
while not placed and attempts < 100:
direction = random.randint(0, 1) # 0 - горизонтально, 1 - вертикально
if direction == 0:
row = random.randint(0, 4)
col = random.randint(0, 4 - size)
ship_coords = [(row, col + i) for i in range(size)]
else:
row = random.randint(0, 4 - size)
col = random.randint(0, 4)
ship_coords = [(row + i, col) for i in range(size)]
# Проверяем пересечения
if not any(coord in [c for ship in ships for c in ship] for coord in ship_coords):
ships.append(ship_coords)
placed = True
attempts += 1
return ships
# ===== ИГРА 2048 =====
def make_2048_move(self, game_id: str, direction: str) -> Dict[str, Any]:
"""Ход в игре 2048"""
session = self.get_game_session(game_id)
if not session or session.game_type != 'game_2048':
raise ValidationError("Игра не найдена или неверный тип игры")
data = session.data
board = data['board']
valid_directions = ['left', 'right', 'up', 'down']
if direction not in valid_directions:
raise ValidationError(f"Неверное направление: {direction}")
# Сохраняем доску для сравнения
old_board = [row[:] for row in board]
# Делаем ход
moved, score_gained = self._move_2048_board(board, direction)
data['score'] += score_gained
data['moves'] += 1
if moved:
self._add_random_2048_tile(board)
# Проверяем победу
if self._check_2048_win(board):
data['game_status'] = 'won'
return {
'status': 'win',
'board': board,
'score': data['score'],
'moved': moved,
'points': self.game_settings['game_2048']['points_win']
}
# Проверяем поражение
if not self._can_move_2048(board):
data['game_status'] = 'lost'
return {
'status': 'lose',
'board': board,
'score': data['score'],
'moved': moved,
'points': 0
}
return {
'status': 'continue',
'board': board,
'score': data['score'],
'moved': moved,
'moves': data['moves'],
'points': 0
}
def _move_2048_board(self, board: List[List[int]], direction: str) -> Tuple[bool, int]:
"""Движение доски 2048"""
def move_row_left(row):
new_row = [num for num in row if num != 0]
score = 0
for i in range(len(new_row) - 1):
if new_row[i] == new_row[i + 1]:
new_row[i] *= 2
score += new_row[i]
new_row[i + 1] = 0
new_row = [num for num in new_row if num != 0]
return new_row + [0] * (4 - len(new_row)), score
moved = False
total_score = 0
if direction == 'left':
for i in range(4):
old_row = board[i][:]
board[i], score = move_row_left(board[i])
if board[i] != old_row:
moved = True
total_score += score
elif direction == 'right':
for i in range(4):
old_row = board[i][::-1]
new_row, score = move_row_left(old_row)
board[i] = new_row[::-1]
if board[i] != old_row[::-1]:
moved = True
total_score += score
elif direction == 'up':
for j in range(4):
col = [board[i][j] for i in range(4)]
old_col = col[:]
new_col, score = move_row_left(col)
for i in range(4):
board[i][j] = new_col[i]
if new_col != old_col:
moved = True
total_score += score
elif direction == 'down':
for j in range(4):
col = [board[i][j] for i in range(3, -1, -1)]
old_col = col[:]
new_col, score = move_row_left(col)
for i in range(3, -1, -1):
board[3 - i][j] = new_col[i]
if new_col != old_col:
moved = True
total_score += score
return moved, total_score
def _add_random_2048_tile(self, board: List[List[int]]):
"""Добавление случайной плитки"""
empty_cells = [(i, j) for i in range(4) for j in range(4) if board[i][j] == 0]
if empty_cells:
i, j = random.choice(empty_cells)
board[i][j] = 2 if random.random() < 0.9 else 4
def _check_2048_win(self, board: List[List[int]]) -> bool:
"""Проверка победы в 2048"""
return any(2048 in row for row in board)
def _can_move_2048(self, board: List[List[int]]) -> bool:
"""Проверка возможности хода"""
for i in range(4):
for j in range(4):
if board[i][j] == 0:
return True
if i > 0 and board[i][j] == board[i-1][j]:
return True
if j > 0 and board[i][j] == board[i][j-1]:
return True
return False
# ===== ТЕТРИС =====
def make_tetris_move(self, game_id: str, action: str) -> Dict[str, Any]:
"""Ход в тетрисе"""
session = self.get_game_session(game_id)
if not session or session.game_type != 'tetris':
raise ValidationError("Игра не найдена или неверный тип игры")
data = session.data
board = data['board']
piece = data['current_piece']
if action == 'left' and piece['x'] > 0:
piece['x'] -= 1
if not self._is_valid_tetris_position(board, piece):
piece['x'] += 1
elif action == 'right' and piece['x'] < 10 - len(piece['shape'][0]):
piece['x'] += 1
if not self._is_valid_tetris_position(board, piece):
piece['x'] -= 1
elif action == 'down':
piece['y'] += 1
if not self._is_valid_tetris_position(board, piece):
piece['y'] -= 1
# Фиксируем фигуру
self._place_tetris_piece(board, piece)
lines_cleared = self._clear_tetris_lines(board)
data['score'] += lines_cleared * self.game_settings['tetris']['points_per_line']
data['lines_cleared'] += lines_cleared
# Создаем новую фигуру
piece = self._get_random_tetris_piece()
if not self._is_valid_tetris_position(board, piece):
data['game_status'] = 'game_over'
return {
'status': 'game_over',
'board': board,
'score': data['score'],
'lines_cleared': data['lines_cleared'],
'points': 0
}
data['current_piece'] = piece
elif action == 'rotate':
old_shape = piece['shape'][:]
piece['shape'] = list(zip(*piece['shape'][::-1]))
if not self._is_valid_tetris_position(board, piece):
piece['shape'] = old_shape
return {
'status': 'continue',
'board': board,
'piece': piece,
'score': data['score'],
'lines_cleared': data['lines_cleared'],
'level': data['level'],
'points': 0
}
def _is_valid_tetris_position(self, board: List[List[int]], piece: Dict) -> bool:
"""Проверка валидности позиции тетрис-фигуры"""
for i, row in enumerate(piece['shape']):
for j, cell in enumerate(row):
if cell:
x, y = piece['x'] + j, piece['y'] + i
if x < 0 or x >= 10 or y >= 20 or (y >= 0 and board[y][x] != 0):
return False
return True
def _place_tetris_piece(self, board: List[List[int]], piece: Dict):
"""Размещение тетрис-фигуры на доске"""
for i, row in enumerate(piece['shape']):
for j, cell in enumerate(row):
if cell:
x, y = piece['x'] + j, piece['y'] + i
if 0 <= y < 20 and 0 <= x < 10:
board[y][x] = 1
def _clear_tetris_lines(self, board: List[List[int]]) -> int:
"""Очистка заполненных линий"""
lines_cleared = 0
y = 19
while y >= 0:
if all(board[y]):
del board[y]
board.insert(0, [0] * 10)
lines_cleared += 1
else:
y -= 1
return lines_cleared
def _get_random_tetris_piece(self) -> Dict:
"""Получение случайной тетрис-фигуры"""
pieces = [
{'shape': [[1, 1, 1, 1]], 'x': 3, 'y': 0}, # I
{'shape': [[1, 1], [1, 1]], 'x': 4, 'y': 0}, # O
{'shape': [[1, 0, 0], [1, 1, 1]], 'x': 3, 'y': 0}, # J
{'shape': [[0, 0, 1], [1, 1, 1]], 'x': 3, 'y': 0}, # L
{'shape': [[0, 1, 1], [1, 1, 0]], 'x': 3, 'y': 0}, # S
{'shape': [[1, 1, 0], [0, 1, 1]], 'x': 3, 'y': 0}, # Z
{'shape': [[0, 1, 0], [1, 1, 1]], 'x': 3, 'y': 0}, # T
]
return random.choice(pieces)
# ===== ЗМЕЙКА =====
def make_snake_move(self, game_id: str, direction: str) -> Dict[str, Any]:
"""Ход в змейке"""
session = self.get_game_session(game_id)
if not session or session.game_type != 'snake':
raise ValidationError("Игра не найдена или неверный тип игры")
data = session.data
snake = data['snake']
food = data['food']
current_direction = data['direction']
# Проверяем допустимость направления
opposites = {'up': 'down', 'down': 'up', 'left': 'right', 'right': 'left'}
if direction == opposites.get(current_direction):
raise ValidationError("Нельзя двигаться в противоположном направлении")
# Вычисляем новую голову
head = snake[0]
if direction == 'up':
new_head = (head[0] - 1, head[1])
elif direction == 'down':
new_head = (head[0] + 1, head[1])
elif direction == 'left':
new_head = (head[0], head[1] - 1)
elif direction == 'right':
new_head = (head[0], head[1] + 1)
else:
raise ValidationError(f"Неверное направление: {direction}")
# Проверяем столкновения
if (new_head[0] < 0 or new_head[0] >= 10 or
new_head[1] < 0 or new_head[1] >= 10 or
new_head in snake):
data['game_status'] = 'game_over'
return {
'status': 'game_over',
'snake': snake,
'food': food,
'score': data['score'],
'length': len(snake),
'points': 0
}
# Добавляем новую голову
snake.insert(0, new_head)
data['direction'] = direction
# Проверяем еду
if new_head == food:
data['score'] += self.game_settings['snake']['points_per_food']
food = self._place_snake_food(data['board'], snake)
data['food'] = food
else:
# Удаляем хвост если не съели еду
snake.pop()
return {
'status': 'continue',
'snake': snake,
'food': food,
'score': data['score'],
'length': len(snake),
'points': 0
}
def _place_snake_food(self, board: List[List[int]], snake: List[Tuple[int, int]]) -> Tuple[int, int]:
"""Размещение еды для змейки"""
empty_cells = [(i, j) for i in range(10) for j in range(10) if (i, j) not in snake]
if empty_cells:
return random.choice(empty_cells)
return None
def cleanup_old_sessions(self, max_age_minutes: int = 30):
"""Очистка старых игровых сессий"""
current_time = datetime.now()
cutoff_time = current_time - timedelta(minutes=max_age_minutes)
to_remove = []
for game_id, session in self.active_sessions.items():
if session.created_at < cutoff_time:
to_remove.append(game_id)
for game_id in to_remove:
del self.active_sessions[game_id]