|
|
@@ -0,0 +1,355 @@
|
|
|
+import re
|
|
|
+import random
|
|
|
+import struct
|
|
|
+import operator
|
|
|
+
|
|
|
+import aiohttp
|
|
|
+
|
|
|
+
|
|
|
+async def _pyrandom(min_val, max_val, is_dice=False):
|
|
|
+ if is_dice:
|
|
|
+ return [random.randint(1, max_val) for _ in range(min_val)]
|
|
|
+ return random.randint(min_val, max_val)
|
|
|
+
|
|
|
+
|
|
|
+async def _get(url):
|
|
|
+ async with aiohttp.ClientSession() as session:
|
|
|
+ async with session.get(url) as resp:
|
|
|
+ return await resp.read()
|
|
|
+
|
|
|
+
|
|
|
+async def generate_unbiased_numbers(data, min_val, max_val, count):
|
|
|
+ mod = max_val - min_val + 1
|
|
|
+ max_acceptable = (1 << 64) // mod * mod
|
|
|
+ numbers = []
|
|
|
+ data_len = len(data)
|
|
|
+ bytes_needed = count * 8
|
|
|
+
|
|
|
+ if data_len < bytes_needed:
|
|
|
+ return None
|
|
|
+
|
|
|
+ for i in range(count):
|
|
|
+ chunk = data[i * 8 : (i + 1) * 8]
|
|
|
+ number = struct.unpack("<Q", chunk)[0]
|
|
|
+
|
|
|
+ while number >= max_acceptable:
|
|
|
+ number = number >> 1
|
|
|
+
|
|
|
+ numbers.append((number % mod) + min_val)
|
|
|
+
|
|
|
+ return numbers
|
|
|
+
|
|
|
+
|
|
|
+async def generate(source, min_val, max_val, count=1):
|
|
|
+ total_bytes = count * 8
|
|
|
+ try:
|
|
|
+ data = await _get(f"{source}{total_bytes}")
|
|
|
+ except Exception:
|
|
|
+ data = b""
|
|
|
+
|
|
|
+ if len(data) >= total_bytes:
|
|
|
+ result = await generate_unbiased_numbers(data, min_val, max_val, count)
|
|
|
+ if result is not None:
|
|
|
+ return result
|
|
|
+
|
|
|
+ if count == 1:
|
|
|
+ return [await _pyrandom(min_val, max_val)]
|
|
|
+ return await _pyrandom(min_val, max_val, is_dice=True)
|
|
|
+
|
|
|
+
|
|
|
+async def _trng_yebisu(min_val, max_val, is_dice=False):
|
|
|
+ count = min_val if is_dice else 1
|
|
|
+ actual_min = 1 if is_dice else min_val
|
|
|
+ actual_max = max_val if is_dice else max_val
|
|
|
+
|
|
|
+ numbers = await generate(
|
|
|
+ "https://yebi.su/api/pool?count=", actual_min, actual_max, count
|
|
|
+ )
|
|
|
+
|
|
|
+ return numbers if is_dice else numbers[0]
|
|
|
+
|
|
|
+
|
|
|
+async def rolldices(count, sides):
|
|
|
+ try:
|
|
|
+ return await _trng_yebisu(count, sides, is_dice=True)
|
|
|
+ except Exception:
|
|
|
+ return await _pyrandom(count, sides, is_dice=True)
|
|
|
+
|
|
|
+
|
|
|
+async def _roll(count, sides):
|
|
|
+ if count <= 0:
|
|
|
+ raise ValueError("Количество костей должно быть больше нуля.")
|
|
|
+
|
|
|
+ if sides <= 0:
|
|
|
+ raise ValueError("Количество сторон должно быть больше нуля.")
|
|
|
+
|
|
|
+ return await rolldices(count, sides)
|
|
|
+
|
|
|
+
|
|
|
+OPS = {
|
|
|
+ "+": operator.add,
|
|
|
+ "-": operator.sub,
|
|
|
+ "*": operator.mul,
|
|
|
+ "/": operator.truediv,
|
|
|
+ "%": operator.mod,
|
|
|
+ "^": operator.pow,
|
|
|
+}
|
|
|
+OPS_KEYS = "".join(OPS.keys()).replace("-", r"\-")
|
|
|
+
|
|
|
+T_NAME = re.compile(r"([abce-zа-ге-йл-яA-ZА-Я]+)")
|
|
|
+T_COLON = re.compile(r"(:)")
|
|
|
+T_DICE = re.compile(r"(d|д|к)")
|
|
|
+T_MINUS = re.compile(r"(-)")
|
|
|
+T_OP = re.compile(f"([{OPS_KEYS}])")
|
|
|
+T_DIGIT = re.compile(r"(\d+)")
|
|
|
+T_OPEN_PAREN = re.compile(r"(\()")
|
|
|
+T_CLOSE_PAREN = re.compile(r"(\))")
|
|
|
+T_WS = re.compile(r"([ \t\r\n]+)")
|
|
|
+
|
|
|
+TOKEN_NAMES = {
|
|
|
+ T_NAME: "имя",
|
|
|
+ T_COLON: "двоеточие",
|
|
|
+ T_DICE: "кость",
|
|
|
+ T_OP: "оператор",
|
|
|
+ T_DIGIT: "число",
|
|
|
+ T_OPEN_PAREN: "открывающая скобка",
|
|
|
+ T_CLOSE_PAREN: "закрывающая скобка",
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+class Value:
|
|
|
+ def __init__(self, value):
|
|
|
+ if not isinstance(value, list):
|
|
|
+ value = [int(value)]
|
|
|
+
|
|
|
+ self.value = value
|
|
|
+
|
|
|
+ def __int__(self):
|
|
|
+ return sum(map(int, self.value))
|
|
|
+
|
|
|
+ def __repr__(self):
|
|
|
+ return str(self.value[0] if len(self.value) == 1 else self.value)
|
|
|
+
|
|
|
+ def __iter__(self):
|
|
|
+ return iter(self.value)
|
|
|
+
|
|
|
+ def __next__(self):
|
|
|
+ return next(self.value)
|
|
|
+
|
|
|
+ def __index__(self, index):
|
|
|
+ return self.value[index]
|
|
|
+
|
|
|
+ def __len__(self):
|
|
|
+ return len(self.value)
|
|
|
+
|
|
|
+ def __eq__(self, other):
|
|
|
+ if not isinstance(other, Value):
|
|
|
+ return False
|
|
|
+
|
|
|
+ return self.value == other.value
|
|
|
+
|
|
|
+ def apply(self, what, *args):
|
|
|
+ args = list(map(int, args))
|
|
|
+
|
|
|
+ return Value(what(int(self), *args))
|
|
|
+
|
|
|
+
|
|
|
+class Dices:
|
|
|
+ def __init__(self, text):
|
|
|
+ self.text = text.strip()
|
|
|
+ self.position = 0
|
|
|
+
|
|
|
+ self.names = {}
|
|
|
+ self.rolls = []
|
|
|
+ self.result = None
|
|
|
+
|
|
|
+ self._rolls = []
|
|
|
+
|
|
|
+ def __repr__(self):
|
|
|
+ if self.result:
|
|
|
+ buffer = ""
|
|
|
+
|
|
|
+ for count, sides, roll in self._rolls:
|
|
|
+ buffer += f"{'' if count == 1 else count}d{sides}: {roll}\n"
|
|
|
+
|
|
|
+ for roll, result in zip(self.rolls, self.result):
|
|
|
+ count, sides, roll = roll
|
|
|
+
|
|
|
+ buffer += f"{'' if count == 1 else count}d{sides}: "
|
|
|
+
|
|
|
+ roll_sum = int(roll)
|
|
|
+ result_sum = int(result)
|
|
|
+ if result_sum == roll_sum:
|
|
|
+ results = ", ".join(map(str, result))
|
|
|
+
|
|
|
+ if "," in results:
|
|
|
+ results = f"[{results}]"
|
|
|
+
|
|
|
+ buffer += results
|
|
|
+
|
|
|
+ if "," in results:
|
|
|
+ buffer += f" ({result_sum})"
|
|
|
+
|
|
|
+ buffer += "\n"
|
|
|
+ else:
|
|
|
+ difference = result_sum - roll_sum
|
|
|
+ results = f"{'' if len(roll) == 1 else f'{roll_sum} -> '}{roll_sum}{'' if difference < 0 else '+'}{difference}"
|
|
|
+
|
|
|
+ buffer += f"{roll} -> {results} ({int(result)})\n"
|
|
|
+
|
|
|
+ return f"```\n{self.text}\n{buffer}= {int(self.result)}```"
|
|
|
+
|
|
|
+ return self.text
|
|
|
+
|
|
|
+ def _skip_ws(self):
|
|
|
+ match = T_WS.match(self.text, self.position)
|
|
|
+ if match:
|
|
|
+ self.position += len(match.group(0))
|
|
|
+
|
|
|
+ def _done(self):
|
|
|
+ self._skip_ws()
|
|
|
+
|
|
|
+ return self.position >= len(self.text)
|
|
|
+
|
|
|
+ def _match(self, what, skip_ws=True):
|
|
|
+ if skip_ws:
|
|
|
+ self._skip_ws()
|
|
|
+
|
|
|
+ match = what.match(self.text, self.position)
|
|
|
+ if match:
|
|
|
+ self.position += len(match.group(0))
|
|
|
+
|
|
|
+ return match.groups()
|
|
|
+
|
|
|
+ def _expected(self, expected):
|
|
|
+ raise SyntaxError(
|
|
|
+ f"Неожиданный ввод на позиции `#{self.position + 1}`: ожидалось: `{expected}`."
|
|
|
+ )
|
|
|
+
|
|
|
+ def _expect(self, what):
|
|
|
+ match = self._match(what)
|
|
|
+ if not match:
|
|
|
+ self._expected(TOKEN_NAMES[what])
|
|
|
+
|
|
|
+ return match
|
|
|
+
|
|
|
+ async def _parse_dice(self, left=1):
|
|
|
+ if self._match(T_OPEN_PAREN):
|
|
|
+ right = await self._parse_expr()
|
|
|
+
|
|
|
+ self._expect(T_CLOSE_PAREN)
|
|
|
+ else:
|
|
|
+ right = await self._parse_atom()
|
|
|
+
|
|
|
+ left = int(left)
|
|
|
+ right = int(right)
|
|
|
+
|
|
|
+ if left > 1000 or right > 1000:
|
|
|
+ raise SyntaxError("Слишком длинное число.")
|
|
|
+
|
|
|
+ roll = Value(await _roll(left, right))
|
|
|
+
|
|
|
+ self._rolls.append((left, right, roll))
|
|
|
+
|
|
|
+ return roll
|
|
|
+
|
|
|
+ async def _parse_atom(self):
|
|
|
+ if self._match(T_OPEN_PAREN):
|
|
|
+ expr = await self._parse_expr()
|
|
|
+
|
|
|
+ self._expect(T_CLOSE_PAREN)
|
|
|
+
|
|
|
+ if self._match(T_DICE, skip_ws=False):
|
|
|
+ return await self._parse_dice(expr)
|
|
|
+
|
|
|
+ return expr
|
|
|
+ elif self._match(T_MINUS):
|
|
|
+ value = await self._parse_atom()
|
|
|
+
|
|
|
+ return value.apply(operator.neg)
|
|
|
+ elif match := self._match(T_DIGIT):
|
|
|
+ try:
|
|
|
+ left = int(match[0])
|
|
|
+ except ValueError:
|
|
|
+ raise SyntaxError("Слишком длинное число.")
|
|
|
+
|
|
|
+ if match := self._match(T_DICE, skip_ws=False):
|
|
|
+ return await self._parse_dice(left)
|
|
|
+
|
|
|
+ return Value(left)
|
|
|
+ elif self._match(T_DICE):
|
|
|
+ return await self._parse_dice()
|
|
|
+ elif match := self._match(T_NAME):
|
|
|
+ name = match[0].upper()
|
|
|
+
|
|
|
+ if name not in self.names:
|
|
|
+ raise NameError(f"Неизвестная переменная: `{match[0]}`.")
|
|
|
+
|
|
|
+ expr = self.names[name]
|
|
|
+
|
|
|
+ if self._match(T_DICE, skip_ws=False):
|
|
|
+ return await self._parse_dice(expr)
|
|
|
+
|
|
|
+ return expr
|
|
|
+
|
|
|
+ self._expected("число, кость или переменная")
|
|
|
+
|
|
|
+ async def _parse_expr(self):
|
|
|
+ left = await self._parse_atom()
|
|
|
+
|
|
|
+ if op := self._match(T_OP):
|
|
|
+ op = OPS[op[0]]
|
|
|
+ right = await self._parse_expr()
|
|
|
+
|
|
|
+ left = left.apply(op, right)
|
|
|
+ elif self._match(T_COLON):
|
|
|
+ right = self._expect(T_NAME)[0].upper()
|
|
|
+
|
|
|
+ self.names[right] = left
|
|
|
+
|
|
|
+ return left
|
|
|
+
|
|
|
+ async def _parse_exprs(self):
|
|
|
+ exprs = []
|
|
|
+ while not self._done():
|
|
|
+ if len(exprs) >= 10:
|
|
|
+ raise SyntaxError("Слишком длинное число.")
|
|
|
+
|
|
|
+ rolls_count = len(self._rolls)
|
|
|
+
|
|
|
+ expr = await self._parse_expr()
|
|
|
+
|
|
|
+ if len(self._rolls) == rolls_count:
|
|
|
+ raise SyntaxError("Выражение не содержит бросков.")
|
|
|
+
|
|
|
+ self.rolls.append(self._rolls.pop(-1))
|
|
|
+
|
|
|
+ exprs.append(expr)
|
|
|
+
|
|
|
+ if not exprs:
|
|
|
+ raise SyntaxError("Выражение не должно быть пустым.")
|
|
|
+
|
|
|
+ return Value(exprs)
|
|
|
+
|
|
|
+ async def roll(self, vars={}):
|
|
|
+ self.names = {str(k).upper(): Value(vars[k]) for k in vars}
|
|
|
+ self.position = 0
|
|
|
+ self.rolls = []
|
|
|
+ self._rolls = []
|
|
|
+
|
|
|
+ self.result = await self._parse_exprs()
|
|
|
+
|
|
|
+ return self
|
|
|
+
|
|
|
+
|
|
|
+async def roll_dices(dices, vars={}):
|
|
|
+ dices = Dices(dices)
|
|
|
+
|
|
|
+ try:
|
|
|
+ await dices.roll(vars=vars)
|
|
|
+ except (ValueError, SyntaxError, NameError) as e:
|
|
|
+ return str(e)
|
|
|
+ except ZeroDivisionError:
|
|
|
+ raise "Попытка деления на ноль."
|
|
|
+
|
|
|
+ return str(dices)
|
