mirror of
https://github.com/aiogram/aiogram.git
synced 2026-04-08 16:37:47 +00:00
Update memory.py
This commit is contained in:
Bokoval69
GitHub
parent
358ecc7821
commit
7e1fac8558
1 changed files with 32 additions and 102 deletions
|
|
@ -1,112 +1,42 @@
|
|||
import copy
|
||||
import typing
|
||||
from typing import List, Tuple
|
||||
|
||||
from ...dispatcher.storage import BaseStorage
|
||||
# 0 1 2 3
|
||||
casino = ["BAR", "виноград", "лимон", "семь"]
|
||||
|
||||
|
||||
class MemoryStorage(BaseStorage):
|
||||
def is_winning_combo(combo) -> Tuple[bool, int]:
|
||||
"""
|
||||
In-memory based states storage.
|
||||
Проверка на выигрышную комбинацию
|
||||
|
||||
This type of storage is not recommended for usage in bots, because you will lost all states after restarting.
|
||||
:param combo: массив значений дайса (см. перем. casino)
|
||||
:return: пара ("есть_выигрыш?", "изменение счёта игрока")
|
||||
"""
|
||||
|
||||
async def wait_closed(self):
|
||||
pass
|
||||
# Все комбинации из трёх одинаковых оцениваем в 10 или 7 очков
|
||||
if combo[0] == combo[1] == combo[2]:
|
||||
if combo[0] == "семь":
|
||||
return True, 10
|
||||
return True, 7
|
||||
# Две семёрки + что угодно = 5 очков
|
||||
elif combo[0] == combo[1] == "семь":
|
||||
return True, 5
|
||||
# Всё остальное -- минус одно очко
|
||||
else:
|
||||
return False, -1
|
||||
|
||||
async def close(self):
|
||||
self.data.clear()
|
||||
|
||||
def __init__(self):
|
||||
self.data = {}
|
||||
def get_casino_values(dice_value) -> List:
|
||||
"""
|
||||
Возвращает то, что было на конкретном дайсе-казино
|
||||
:param dice_value: Число, которое вернул Bot API
|
||||
:return: строка, содержащая все выпавшие элементы
|
||||
|
||||
def resolve_address(self, chat, user):
|
||||
chat_id, user_id = map(str, self.check_address(chat=chat, user=user))
|
||||
|
||||
if chat_id not in self.data:
|
||||
self.data[chat_id] = {}
|
||||
if user_id not in self.data[chat_id]:
|
||||
self.data[chat_id][user_id] = {'state': None, 'data': {}, 'bucket': {}}
|
||||
|
||||
return chat_id, user_id
|
||||
|
||||
async def get_state(self, *,
|
||||
chat: typing.Union[str, int, None] = None,
|
||||
user: typing.Union[str, int, None] = None,
|
||||
default: typing.Optional[str] = None) -> typing.Optional[str]:
|
||||
chat, user = self.resolve_address(chat=chat, user=user)
|
||||
return self.data[chat][user].get("state", self.resolve_state(default))
|
||||
|
||||
async def get_data(self, *,
|
||||
chat: typing.Union[str, int, None] = None,
|
||||
user: typing.Union[str, int, None] = None,
|
||||
default: typing.Optional[str] = None) -> typing.Dict:
|
||||
chat, user = self.resolve_address(chat=chat, user=user)
|
||||
return copy.deepcopy(self.data[chat][user]['data'])
|
||||
|
||||
async def update_data(self, *,
|
||||
chat: typing.Union[str, int, None] = None,
|
||||
user: typing.Union[str, int, None] = None,
|
||||
data: typing.Dict = None, **kwargs):
|
||||
if data is None:
|
||||
data = {}
|
||||
chat, user = self.resolve_address(chat=chat, user=user)
|
||||
self.data[chat][user]['data'].update(data, **kwargs)
|
||||
|
||||
async def set_state(self, *,
|
||||
chat: typing.Union[str, int, None] = None,
|
||||
user: typing.Union[str, int, None] = None,
|
||||
state: typing.AnyStr = None):
|
||||
chat, user = self.resolve_address(chat=chat, user=user)
|
||||
self.data[chat][user]['state'] = self.resolve_state(state)
|
||||
|
||||
async def set_data(self, *,
|
||||
chat: typing.Union[str, int, None] = None,
|
||||
user: typing.Union[str, int, None] = None,
|
||||
data: typing.Dict = None):
|
||||
chat, user = self.resolve_address(chat=chat, user=user)
|
||||
self.data[chat][user]['data'] = copy.deepcopy(data)
|
||||
self._cleanup(chat, user)
|
||||
|
||||
async def reset_state(self, *,
|
||||
chat: typing.Union[str, int, None] = None,
|
||||
user: typing.Union[str, int, None] = None,
|
||||
with_data: typing.Optional[bool] = True):
|
||||
await self.set_state(chat=chat, user=user, state=None)
|
||||
if with_data:
|
||||
await self.set_data(chat=chat, user=user, data={})
|
||||
self._cleanup(chat, user)
|
||||
|
||||
def has_bucket(self):
|
||||
return True
|
||||
|
||||
async def get_bucket(self, *,
|
||||
chat: typing.Union[str, int, None] = None,
|
||||
user: typing.Union[str, int, None] = None,
|
||||
default: typing.Optional[dict] = None) -> typing.Dict:
|
||||
chat, user = self.resolve_address(chat=chat, user=user)
|
||||
return copy.deepcopy(self.data[chat][user]['bucket'])
|
||||
|
||||
async def set_bucket(self, *,
|
||||
chat: typing.Union[str, int, None] = None,
|
||||
user: typing.Union[str, int, None] = None,
|
||||
bucket: typing.Dict = None):
|
||||
chat, user = self.resolve_address(chat=chat, user=user)
|
||||
self.data[chat][user]['bucket'] = copy.deepcopy(bucket)
|
||||
self._cleanup(chat, user)
|
||||
|
||||
async def update_bucket(self, *,
|
||||
chat: typing.Union[str, int, None] = None,
|
||||
user: typing.Union[str, int, None] = None,
|
||||
bucket: typing.Dict = None, **kwargs):
|
||||
if bucket is None:
|
||||
bucket = {}
|
||||
chat, user = self.resolve_address(chat=chat, user=user)
|
||||
self.data[chat][user]['bucket'].update(bucket, **kwargs)
|
||||
|
||||
def _cleanup(self, chat, user):
|
||||
chat, user = self.resolve_address(chat=chat, user=user)
|
||||
if self.data[chat][user] == {'state': None, 'data': {}, 'bucket': {}}:
|
||||
del self.data[chat][user]
|
||||
if not self.data[chat]:
|
||||
del self.data[chat]
|
||||
Альтернативный вариант (ещё раз спасибо t.me/svinerus):
|
||||
return [casino[(dice_value - 1) // i % 4]for i in (1, 4, 16)]
|
||||
"""
|
||||
dice_value -= 1
|
||||
result = []
|
||||
for _ in range(3):
|
||||
result.append(casino[dice_value % 4])
|
||||
dice_value //= 4
|
||||
return result
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue