mirror of
https://github.com/uhIgnacio/EmoteManager.git
synced 2024-08-15 02:23:13 +00:00
format
This commit is contained in:
igna
parent
1ff3e93938
commit
3f73309974
10 changed files with 584 additions and 523 deletions
120
utils/archive.py
120
utils/archive.py
|
|
@ -14,79 +14,85 @@ from . import errors
|
|||
|
||||
ArchiveInfo = collections.namedtuple('ArchiveInfo', 'filename content error')
|
||||
|
||||
|
||||
def extract(archive: typing.io.BinaryIO, *, size_limit=None) \
|
||||
-> Iterable[Tuple[str, Optional[bytes], Optional[BaseException]]]:
|
||||
"""
|
||||
extract a binary file-like object representing a zip or uncompressed tar archive, yielding filenames and contents.
|
||||
-> Iterable[Tuple[str, Optional[bytes], Optional[BaseException]]]:
|
||||
"""
|
||||
extract a binary file-like object representing a zip or uncompressed tar archive, yielding filenames and contents.
|
||||
|
||||
yields ArchiveInfo objects: (filename: str, content: typing.Optional[bytes], error: )
|
||||
if size_limit is not None and the size limit is exceeded, or for any other error, yield None for content
|
||||
on success, error will be None
|
||||
"""
|
||||
yields ArchiveInfo objects: (filename: str, content: typing.Optional[bytes], error: )
|
||||
if size_limit is not None and the size limit is exceeded, or for any other error, yield None for content
|
||||
on success, error will be None
|
||||
"""
|
||||
|
||||
try:
|
||||
yield from extract_zip(archive, size_limit=size_limit)
|
||||
return
|
||||
except zipfile.BadZipFile:
|
||||
pass
|
||||
finally:
|
||||
archive.seek(0)
|
||||
try:
|
||||
yield from extract_zip(archive, size_limit=size_limit)
|
||||
return
|
||||
except zipfile.BadZipFile:
|
||||
pass
|
||||
finally:
|
||||
archive.seek(0)
|
||||
|
||||
try:
|
||||
yield from extract_tar(archive, size_limit=size_limit)
|
||||
except tarfile.ReadError as exc:
|
||||
raise ValueError('not a valid zip or tar file') from exc
|
||||
finally:
|
||||
archive.seek(0)
|
||||
|
||||
try:
|
||||
yield from extract_tar(archive, size_limit=size_limit)
|
||||
except tarfile.ReadError as exc:
|
||||
raise ValueError('not a valid zip or tar file') from exc
|
||||
finally:
|
||||
archive.seek(0)
|
||||
|
||||
def extract_zip(archive, *, size_limit=None):
|
||||
with zipfile.ZipFile(archive) as zip:
|
||||
members = [m for m in zip.infolist() if not m.is_dir()]
|
||||
for member in members:
|
||||
if size_limit is not None and member.file_size >= size_limit:
|
||||
yield ArchiveInfo(
|
||||
filename=member.filename,
|
||||
content=None,
|
||||
error=errors.FileTooBigError(member.file_size, size_limit))
|
||||
continue
|
||||
with zipfile.ZipFile(archive) as zip:
|
||||
members = [m for m in zip.infolist() if not m.is_dir()]
|
||||
for member in members:
|
||||
if size_limit is not None and member.file_size >= size_limit:
|
||||
yield ArchiveInfo(
|
||||
filename=member.filename,
|
||||
content=None,
|
||||
error=errors.FileTooBigError(member.file_size, size_limit))
|
||||
continue
|
||||
|
||||
try:
|
||||
content = zip.open(member).read()
|
||||
except RuntimeError as exc: # why no specific exceptions smh
|
||||
yield ArchiveInfo(filename=member.filename, content=None, error=exc)
|
||||
else: # this else is required to avoid UnboundLocalError for some reason
|
||||
yield ArchiveInfo(filename=member.filename, content=content, error=None)
|
||||
|
||||
try:
|
||||
content = zip.open(member).read()
|
||||
except RuntimeError as exc: # why no specific exceptions smh
|
||||
yield ArchiveInfo(filename=member.filename, content=None, error=exc)
|
||||
else: # this else is required to avoid UnboundLocalError for some reason
|
||||
yield ArchiveInfo(filename=member.filename, content=content, error=None)
|
||||
|
||||
def extract_tar(archive, *, size_limit=None):
|
||||
with tarfile.open(fileobj=archive) as tar:
|
||||
members = [f for f in tar.getmembers() if f.isfile()]
|
||||
for member in members:
|
||||
if size_limit is not None and member.size >= size_limit:
|
||||
yield ArchiveInfo(
|
||||
filename=member.name,
|
||||
content=None,
|
||||
error=errors.FileTooBigError(member.size, size_limit))
|
||||
continue
|
||||
with tarfile.open(fileobj=archive) as tar:
|
||||
members = [f for f in tar.getmembers() if f.isfile()]
|
||||
for member in members:
|
||||
if size_limit is not None and member.size >= size_limit:
|
||||
yield ArchiveInfo(
|
||||
filename=member.name,
|
||||
content=None,
|
||||
error=errors.FileTooBigError(member.size, size_limit))
|
||||
continue
|
||||
|
||||
yield ArchiveInfo(member.name, content=tar.extractfile(member).read(), error=None)
|
||||
|
||||
yield ArchiveInfo(member.name, content=tar.extractfile(member).read(), error=None)
|
||||
|
||||
async def extract_async(archive: typing.io.BinaryIO, size_limit=None):
|
||||
for x in extract(archive, size_limit=size_limit):
|
||||
yield await asyncio.sleep(0, x)
|
||||
for x in extract(archive, size_limit=size_limit):
|
||||
yield await asyncio.sleep(0, x)
|
||||
|
||||
|
||||
def main():
|
||||
import io
|
||||
import sys
|
||||
import io
|
||||
import sys
|
||||
|
||||
import humanize
|
||||
import humanize
|
||||
|
||||
arc = io.BytesIO(sys.stdin.detach().read())
|
||||
for name, data, error in extract(arc):
|
||||
if error is not None:
|
||||
print(f'{name}: {error}')
|
||||
continue
|
||||
arc = io.BytesIO(sys.stdin.detach().read())
|
||||
for name, data, error in extract(arc):
|
||||
if error is not None:
|
||||
print(f'{name}: {error}')
|
||||
continue
|
||||
|
||||
print(f'{name}: {humanize.naturalsize(len(data)):>10}')
|
||||
|
||||
print(f'{name}: {humanize.naturalsize(len(data)):>10}')
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
main()
|
||||
|
|
|
|||
|
|
@ -2,25 +2,28 @@
|
|||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
import functools
|
||||
from discord.ext.commands import BadArgument
|
||||
from nextcord.ext.commands import BadArgument
|
||||
|
||||
_emote_type_predicates = {
|
||||
'all': lambda _: True,
|
||||
'static': lambda e: not e.animated,
|
||||
'animated': lambda e: e.animated}
|
||||
'all': lambda _: True,
|
||||
'static': lambda e: not e.animated,
|
||||
'animated': lambda e: e.animated}
|
||||
|
||||
# this is kind of a hack to ensure that the last argument is always converted, even if the default is used.
|
||||
|
||||
|
||||
def emote_type_filter_default(command):
|
||||
old_callback = command.callback
|
||||
old_callback = command.callback
|
||||
|
||||
@functools.wraps(old_callback)
|
||||
async def callback(self, ctx, *args):
|
||||
image_type = args[-1]
|
||||
try:
|
||||
image_type = _emote_type_predicates[image_type]
|
||||
except KeyError:
|
||||
raise BadArgument(f'Invalid emote type. Specify one of "all", "static", or "animated".')
|
||||
return await old_callback(self, ctx, *args[:-1], image_type)
|
||||
@functools.wraps(old_callback)
|
||||
async def callback(self, ctx, *args):
|
||||
image_type = args[-1]
|
||||
try:
|
||||
image_type = _emote_type_predicates[image_type]
|
||||
except KeyError:
|
||||
raise BadArgument(
|
||||
f'Invalid emote type. Specify one of "all", "static", or "animated".')
|
||||
return await old_callback(self, ctx, *args[:-1], image_type)
|
||||
|
||||
command.callback = callback
|
||||
return command
|
||||
command.callback = callback
|
||||
return command
|
||||
|
|
|
|||
|
|
@ -11,9 +11,11 @@ various utilities related to custom emotes
|
|||
RE_EMOTE = re.compile(r'(:|;)(?P<name>\w{2,32})\1|(?P<newline>\n)', re.ASCII)
|
||||
|
||||
"""Matches only custom server emotes."""
|
||||
RE_CUSTOM_EMOTE = re.compile(r'<(?P<animated>a?):(?P<name>\w{2,32}):(?P<id>\d{17,})>', re.ASCII)
|
||||
RE_CUSTOM_EMOTE = re.compile(
|
||||
r'<(?P<animated>a?):(?P<name>\w{2,32}):(?P<id>\d{17,})>', re.ASCII)
|
||||
|
||||
|
||||
def url(id, *, animated: bool = False):
|
||||
"""Convert an emote ID to the image URL for that emote."""
|
||||
extension = 'gif' if animated else 'png'
|
||||
return f'https://cdn.discordapp.com/emojis/{id}.{extension}?v=1'
|
||||
"""Convert an emote ID to the image URL for that emote."""
|
||||
extension = 'gif' if animated else 'png'
|
||||
return f'https://cdn.discordapp.com/emojis/{id}.{extension}?v=1'
|
||||
|
|
|
|||
|
|
@ -17,100 +17,105 @@ from nextcord import HTTPException, Forbidden, NotFound, DiscordServerError
|
|||
|
||||
GuildId = int
|
||||
|
||||
async def json_or_text(resp):
|
||||
text = await resp.text(encoding='utf-8')
|
||||
try:
|
||||
if resp.headers['content-type'] == 'application/json':
|
||||
return json.loads(text)
|
||||
except KeyError:
|
||||
# Thanks Cloudflare
|
||||
pass
|
||||
|
||||
return text
|
||||
async def json_or_text(resp):
|
||||
text = await resp.text(encoding='utf-8')
|
||||
try:
|
||||
if resp.headers['content-type'] == 'application/json':
|
||||
return json.loads(text)
|
||||
except KeyError:
|
||||
# Thanks Cloudflare
|
||||
pass
|
||||
|
||||
return text
|
||||
|
||||
|
||||
class EmoteClient:
|
||||
BASE_URL = 'https://discord.com/api/v9'
|
||||
HTTP_ERROR_CLASSES = {
|
||||
HTTPStatus.FORBIDDEN: Forbidden,
|
||||
HTTPStatus.NOT_FOUND: NotFound,
|
||||
HTTPStatus.SERVICE_UNAVAILABLE: DiscordServerError,
|
||||
}
|
||||
BASE_URL = 'https://discord.com/api/v9'
|
||||
HTTP_ERROR_CLASSES = {
|
||||
HTTPStatus.FORBIDDEN: Forbidden,
|
||||
HTTPStatus.NOT_FOUND: NotFound,
|
||||
HTTPStatus.SERVICE_UNAVAILABLE: DiscordServerError,
|
||||
}
|
||||
|
||||
def __init__(self, bot):
|
||||
self.guild_rls: Dict[GuildId, float] = {}
|
||||
self.http = aiohttp.ClientSession(headers={
|
||||
'User-Agent': bot.config['user_agent'] + ' ' + bot.http.user_agent,
|
||||
'Authorization': 'Bot ' + bot.config['tokens']['discord'],
|
||||
'X-Ratelimit-Precision': 'millisecond',
|
||||
})
|
||||
def __init__(self, bot):
|
||||
self.guild_rls: Dict[GuildId, float] = {}
|
||||
self.http = aiohttp.ClientSession(headers={
|
||||
'User-Agent': bot.config['user_agent'] + ' ' + bot.http.user_agent,
|
||||
'Authorization': 'Bot ' + bot.config['tokens']['discord'],
|
||||
'X-Ratelimit-Precision': 'millisecond',
|
||||
})
|
||||
|
||||
async def request(self, method, path, guild_id, **kwargs):
|
||||
self.check_rl(guild_id)
|
||||
async def request(self, method, path, guild_id, **kwargs):
|
||||
self.check_rl(guild_id)
|
||||
|
||||
headers = {}
|
||||
# Emote Manager shouldn't use walrus op until Debian adopts 3.8 :(
|
||||
reason = kwargs.pop('reason', None)
|
||||
if reason:
|
||||
headers['X-Audit-Log-Reason'] = urllib.parse.quote(reason, safe='/ ')
|
||||
kwargs['headers'] = headers
|
||||
headers = {}
|
||||
# Emote Manager shouldn't use walrus op until Debian adopts 3.8 :(
|
||||
reason = kwargs.pop('reason', None)
|
||||
if reason:
|
||||
headers['X-Audit-Log-Reason'] = urllib.parse.quote(
|
||||
reason, safe='/ ')
|
||||
kwargs['headers'] = headers
|
||||
|
||||
# TODO handle OSError and 500/502, like dpy does
|
||||
async with self.http.request(method, self.BASE_URL + path, **kwargs) as resp:
|
||||
if resp.status == HTTPStatus.TOO_MANY_REQUESTS:
|
||||
return await self._handle_rl(resp, method, path, guild_id, **kwargs)
|
||||
# TODO handle OSError and 500/502, like dpy does
|
||||
async with self.http.request(method, self.BASE_URL + path, **kwargs) as resp:
|
||||
if resp.status == HTTPStatus.TOO_MANY_REQUESTS:
|
||||
return await self._handle_rl(resp, method, path, guild_id, **kwargs)
|
||||
|
||||
data = await json_or_text(resp)
|
||||
if resp.status in range(200, 300):
|
||||
return data
|
||||
data = await json_or_text(resp)
|
||||
if resp.status in range(200, 300):
|
||||
return data
|
||||
|
||||
error_cls = self.HTTP_ERROR_CLASSES.get(resp.status, HTTPException)
|
||||
raise error_cls(resp, data)
|
||||
error_cls = self.HTTP_ERROR_CLASSES.get(resp.status, HTTPException)
|
||||
raise error_cls(resp, data)
|
||||
|
||||
# optimization method that lets us check the RL before downloading the user's image.
|
||||
# also lets us preemptively check the RL before doing a request
|
||||
def check_rl(self, guild_id):
|
||||
try:
|
||||
retry_at = self.guild_rls[guild_id]
|
||||
except KeyError:
|
||||
return
|
||||
# optimization method that lets us check the RL before downloading the user's image.
|
||||
# also lets us preemptively check the RL before doing a request
|
||||
def check_rl(self, guild_id):
|
||||
try:
|
||||
retry_at = self.guild_rls[guild_id]
|
||||
except KeyError:
|
||||
return
|
||||
|
||||
now = datetime.datetime.now(tz=datetime.timezone.utc).timestamp()
|
||||
if retry_at < now:
|
||||
del self.guild_rls[guild_id]
|
||||
return
|
||||
now = datetime.datetime.now(tz=datetime.timezone.utc).timestamp()
|
||||
if retry_at < now:
|
||||
del self.guild_rls[guild_id]
|
||||
return
|
||||
|
||||
raise RateLimitedError(retry_at)
|
||||
raise RateLimitedError(retry_at)
|
||||
|
||||
async def _handle_rl(self, resp, method, path, guild_id, **kwargs):
|
||||
retry_after = (await resp.json())['retry_after'] / 1000.0
|
||||
retry_at = datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(seconds=retry_after)
|
||||
async def _handle_rl(self, resp, method, path, guild_id, **kwargs):
|
||||
retry_after = (await resp.json())['retry_after'] / 1000.0
|
||||
retry_at = datetime.datetime.now(
|
||||
tz=datetime.timezone.utc) + datetime.timedelta(seconds=retry_after)
|
||||
|
||||
# cache unconditionally in case request() is called again while we're sleeping
|
||||
self.guild_rls[guild_id] = retry_at.timestamp()
|
||||
# cache unconditionally in case request() is called again while we're sleeping
|
||||
self.guild_rls[guild_id] = retry_at.timestamp()
|
||||
|
||||
if retry_after < 10.0:
|
||||
await asyncio.sleep(retry_after)
|
||||
# woo mutual recursion
|
||||
return await self.request(method, path, guild_id, **kwargs)
|
||||
if retry_after < 10.0:
|
||||
await asyncio.sleep(retry_after)
|
||||
# woo mutual recursion
|
||||
return await self.request(method, path, guild_id, **kwargs)
|
||||
|
||||
# we've been hit with one of those crazy high rate limits, which only occur for specific methods
|
||||
raise RateLimitedError(retry_at)
|
||||
# we've been hit with one of those crazy high rate limits, which only occur for specific methods
|
||||
raise RateLimitedError(retry_at)
|
||||
|
||||
async def create(self, *, guild, name, image: bytes, role_ids=(), reason=None):
|
||||
data = await self.request(
|
||||
'POST', f'/guilds/{guild.id}/emojis',
|
||||
guild.id,
|
||||
json=dict(name=name, image=image_utils.image_to_base64_url(image), roles=role_ids),
|
||||
reason=reason,
|
||||
)
|
||||
return PartialEmoji(animated=data.get('animated', False), name=data.get('name'), id=data.get('id'))
|
||||
async def create(self, *, guild, name, image: bytes, role_ids=(), reason=None):
|
||||
data = await self.request(
|
||||
'POST', f'/guilds/{guild.id}/emojis',
|
||||
guild.id,
|
||||
json=dict(name=name, image=image_utils.image_to_base64_url(
|
||||
image), roles=role_ids),
|
||||
reason=reason,
|
||||
)
|
||||
return PartialEmoji(animated=data.get('animated', False), name=data.get('name'), id=data.get('id'))
|
||||
|
||||
async def __aenter__(self):
|
||||
self.http = await self.http.__aenter__()
|
||||
return self
|
||||
async def __aenter__(self):
|
||||
self.http = await self.http.__aenter__()
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *excinfo):
|
||||
return await self.http.__aexit__(*excinfo)
|
||||
async def __aexit__(self, *excinfo):
|
||||
return await self.http.__aexit__(*excinfo)
|
||||
|
||||
async def close(self):
|
||||
return await self.http.close()
|
||||
async def close(self):
|
||||
return await self.http.close()
|
||||
|
|
|
|||
111
utils/errors.py
111
utils/errors.py
|
|
@ -5,73 +5,98 @@ import utils
|
|||
import asyncio
|
||||
import humanize
|
||||
import datetime
|
||||
from discord.ext import commands
|
||||
from nextcord.ext import commands
|
||||
|
||||
|
||||
class MissingManageEmojisPermission(commands.MissingPermissions):
|
||||
"""The invoker or the bot doesn't have permissions to manage server emojis."""
|
||||
"""The invoker or the bot doesn't have permissions to manage server emojis."""
|
||||
|
||||
def __init__(self):
|
||||
super(Exception, self).__init__(
|
||||
f'{utils.SUCCESS_EMOJIS[False]} '
|
||||
"Sorry, you don't have enough permissions to run this command. "
|
||||
'You and I both need the Manage Emojis permission.')
|
||||
|
||||
def __init__(self):
|
||||
super(Exception, self).__init__(
|
||||
f'{utils.SUCCESS_EMOJIS[False]} '
|
||||
"Sorry, you don't have enough permissions to run this command. "
|
||||
'You and I both need the Manage Emojis permission.')
|
||||
|
||||
class EmoteManagerError(commands.CommandError):
|
||||
"""Generic error with the bot. This can be used to catch all bot errors."""
|
||||
pass
|
||||
"""Generic error with the bot. This can be used to catch all bot errors."""
|
||||
pass
|
||||
|
||||
|
||||
class ImageProcessingTimeoutError(EmoteManagerError, asyncio.TimeoutError):
|
||||
pass
|
||||
pass
|
||||
|
||||
|
||||
class ImageResizeTimeoutError(ImageProcessingTimeoutError):
|
||||
"""Resizing the image took too long."""
|
||||
def __init__(self):
|
||||
super().__init__('Error: resizing the image took too long.')
|
||||
"""Resizing the image took too long."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__('Error: resizing the image took too long.')
|
||||
|
||||
|
||||
class ImageConversionTimeoutError(ImageProcessingTimeoutError):
|
||||
def __init__(self):
|
||||
super().__init__('Error: converting the image to a GIF took too long.')
|
||||
def __init__(self):
|
||||
super().__init__('Error: converting the image to a GIF took too long.')
|
||||
|
||||
|
||||
class HTTPException(EmoteManagerError):
|
||||
"""The server did not respond with an OK status code. This is only for non-Discord HTTP requests."""
|
||||
def __init__(self, status):
|
||||
super().__init__(f'URL error: server returned error code {status}')
|
||||
"""The server did not respond with an OK status code. This is only for non-Discord HTTP requests."""
|
||||
|
||||
def __init__(self, status):
|
||||
super().__init__(f'URL error: server returned error code {status}')
|
||||
|
||||
|
||||
class RateLimitedError(EmoteManagerError):
|
||||
def __init__(self, retry_at):
|
||||
if isinstance(retry_at, float):
|
||||
# it took me about an HOUR to realize i had to pass tz because utcfromtimestamp returns a NAÏVE time obj!
|
||||
retry_at = datetime.datetime.fromtimestamp(retry_at, tz=datetime.timezone.utc)
|
||||
# humanize.naturaltime is annoying to work with due to timezones so we use this
|
||||
delta = humanize.naturaldelta(retry_at, when=datetime.datetime.now(tz=datetime.timezone.utc))
|
||||
super().__init__(f'Error: Discord told me to slow down! Please retry this command in {delta}.')
|
||||
def __init__(self, retry_at):
|
||||
if isinstance(retry_at, float):
|
||||
# it took me about an HOUR to realize i had to pass tz because utcfromtimestamp returns a NAÏVE time obj!
|
||||
retry_at = datetime.datetime.fromtimestamp(
|
||||
retry_at, tz=datetime.timezone.utc)
|
||||
# humanize.naturaltime is annoying to work with due to timezones so we use this
|
||||
delta = humanize.naturaldelta(
|
||||
retry_at, when=datetime.datetime.now(tz=datetime.timezone.utc))
|
||||
super().__init__(
|
||||
f'Error: Discord told me to slow down! Please retry this command in {delta}.')
|
||||
|
||||
|
||||
class EmoteNotFoundError(EmoteManagerError):
|
||||
"""An emote with that name was not found"""
|
||||
def __init__(self, name):
|
||||
super().__init__(f'An emote called `{name}` does not exist in this server.')
|
||||
"""An emote with that name was not found"""
|
||||
|
||||
def __init__(self, name):
|
||||
super().__init__(
|
||||
f'An emote called `{name}` does not exist in this server.')
|
||||
|
||||
|
||||
class FileTooBigError(EmoteManagerError):
|
||||
def __init__(self, size, limit):
|
||||
self.size = size
|
||||
self.limit = limit
|
||||
def __init__(self, size, limit):
|
||||
self.size = size
|
||||
self.limit = limit
|
||||
|
||||
|
||||
class InvalidFileError(EmoteManagerError):
|
||||
"""The file is not a zip, tar, GIF, PNG, JPG, or WEBP file."""
|
||||
def __init__(self):
|
||||
super().__init__('Invalid file given.')
|
||||
"""The file is not a zip, tar, GIF, PNG, JPG, or WEBP file."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__('Invalid file given.')
|
||||
|
||||
|
||||
class InvalidImageError(InvalidFileError):
|
||||
"""The image is not a GIF, PNG, or JPG"""
|
||||
def __init__(self):
|
||||
super(Exception, self).__init__('The image supplied was not a GIF, PNG, JPG, or WEBP file.')
|
||||
"""The image is not a GIF, PNG, or JPG"""
|
||||
|
||||
def __init__(self):
|
||||
super(Exception, self).__init__(
|
||||
'The image supplied was not a GIF, PNG, JPG, or WEBP file.')
|
||||
|
||||
|
||||
class PermissionDeniedError(EmoteManagerError):
|
||||
"""Raised when a user tries to modify an emote without the Manage Emojis permission"""
|
||||
def __init__(self, name):
|
||||
super().__init__(f"You're not authorized to modify `{name}`.")
|
||||
"""Raised when a user tries to modify an emote without the Manage Emojis permission"""
|
||||
|
||||
def __init__(self, name):
|
||||
super().__init__(f"You're not authorized to modify `{name}`.")
|
||||
|
||||
|
||||
class DiscordError(Exception):
|
||||
"""Usually raised when the client cache is being baka"""
|
||||
def __init__(self):
|
||||
super().__init__('Discord seems to be having issues right now, please try again later.')
|
||||
"""Usually raised when the client cache is being baka"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__('Discord seems to be having issues right now, please try again later.')
|
||||
|
|
|
|||
219
utils/image.py
219
utils/image.py
|
|
@ -1,6 +1,7 @@
|
|||
# © lambda#0987 <lambda@lambda.dance>
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
|
||||
from utils import errors
|
||||
import asyncio
|
||||
import base64
|
||||
import contextlib
|
||||
|
|
@ -14,143 +15,155 @@ import typing
|
|||
logger = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
import wand.image
|
||||
import wand.image
|
||||
except (ImportError, OSError):
|
||||
logger.warn('Failed to import wand.image. Image manipulation functions will be unavailable.')
|
||||
logger.warn(
|
||||
'Failed to import wand.image. Image manipulation functions will be unavailable.')
|
||||
else:
|
||||
import wand.exceptions
|
||||
import wand.exceptions
|
||||
|
||||
from utils import errors
|
||||
|
||||
def resize_until_small(image_data: io.BytesIO) -> None:
|
||||
"""If the image_data is bigger than 256KB, resize it until it's not."""
|
||||
# It's important that we only attempt to resize the image when we have to,
|
||||
# ie when it exceeds the Discord limit of 256KiB.
|
||||
# Apparently some <256KiB images become larger when we attempt to resize them,
|
||||
# so resizing sometimes does more harm than good.
|
||||
max_resolution = 128 # pixels
|
||||
image_size = size(image_data)
|
||||
if image_size <= 256 * 2**10:
|
||||
return
|
||||
"""If the image_data is bigger than 256KB, resize it until it's not."""
|
||||
# It's important that we only attempt to resize the image when we have to,
|
||||
# ie when it exceeds the Discord limit of 256KiB.
|
||||
# Apparently some <256KiB images become larger when we attempt to resize them,
|
||||
# so resizing sometimes does more harm than good.
|
||||
max_resolution = 128 # pixels
|
||||
image_size = size(image_data)
|
||||
if image_size <= 256 * 2**10:
|
||||
return
|
||||
|
||||
try:
|
||||
with wand.image.Image(blob=image_data) as original_image:
|
||||
while True:
|
||||
logger.debug('image size too big (%s bytes)', image_size)
|
||||
logger.debug('attempting resize to at most%s*%s pixels', max_resolution, max_resolution)
|
||||
try:
|
||||
with wand.image.Image(blob=image_data) as original_image:
|
||||
while True:
|
||||
logger.debug('image size too big (%s bytes)', image_size)
|
||||
logger.debug('attempting resize to at most%s*%s pixels',
|
||||
max_resolution, max_resolution)
|
||||
|
||||
with original_image.clone() as resized:
|
||||
resized.transform(resize=f'{max_resolution}x{max_resolution}')
|
||||
image_size = len(resized.make_blob())
|
||||
if image_size <= 256 * 2**10 or max_resolution < 32: # don't resize past 256KiB or 32×32
|
||||
image_data.truncate(0)
|
||||
image_data.seek(0)
|
||||
resized.save(file=image_data)
|
||||
image_data.seek(0)
|
||||
break
|
||||
with original_image.clone() as resized:
|
||||
resized.transform(
|
||||
resize=f'{max_resolution}x{max_resolution}')
|
||||
image_size = len(resized.make_blob())
|
||||
if image_size <= 256 * 2**10 or max_resolution < 32: # don't resize past 256KiB or 32×32
|
||||
image_data.truncate(0)
|
||||
image_data.seek(0)
|
||||
resized.save(file=image_data)
|
||||
image_data.seek(0)
|
||||
break
|
||||
|
||||
max_resolution //= 2
|
||||
except wand.exceptions.CoderError:
|
||||
raise errors.InvalidImageError
|
||||
|
||||
max_resolution //= 2
|
||||
except wand.exceptions.CoderError:
|
||||
raise errors.InvalidImageError
|
||||
|
||||
def convert_to_gif(image_data: io.BytesIO) -> None:
|
||||
try:
|
||||
with wand.image.Image(blob=image_data) as orig, orig.convert('gif') as converted:
|
||||
# discord tries to stop us from abusing animated gif slots by detecting single frame gifs
|
||||
# so make it two frames
|
||||
converted.sequence[0].delay = 0 # show the first frame forever
|
||||
converted.sequence.append(wand.image.Image(width=1, height=1))
|
||||
try:
|
||||
with wand.image.Image(blob=image_data) as orig, orig.convert('gif') as converted:
|
||||
# discord tries to stop us from abusing animated gif slots by detecting single frame gifs
|
||||
# so make it two frames
|
||||
converted.sequence[0].delay = 0 # show the first frame forever
|
||||
converted.sequence.append(wand.image.Image(width=1, height=1))
|
||||
|
||||
image_data.truncate(0)
|
||||
image_data.seek(0)
|
||||
converted.save(file=image_data)
|
||||
image_data.seek(0)
|
||||
except wand.exceptions.CoderError:
|
||||
raise errors.InvalidImageError
|
||||
|
||||
image_data.truncate(0)
|
||||
image_data.seek(0)
|
||||
converted.save(file=image_data)
|
||||
image_data.seek(0)
|
||||
except wand.exceptions.CoderError:
|
||||
raise errors.InvalidImageError
|
||||
|
||||
def mime_type_for_image(data):
|
||||
if data.startswith(b'\x89PNG\r\n\x1a\n'):
|
||||
return 'image/png'
|
||||
if data.startswith(b'\xFF\xD8') and data.rstrip(b'\0').endswith(b'\xFF\xD9'):
|
||||
return 'image/jpeg'
|
||||
if data.startswith((b'GIF87a', b'GIF89a')):
|
||||
return 'image/gif'
|
||||
if data.startswith(b'RIFF') and data[8:12] == b'WEBP':
|
||||
return 'image/webp'
|
||||
raise errors.InvalidImageError
|
||||
if data.startswith(b'\x89PNG\r\n\x1a\n'):
|
||||
return 'image/png'
|
||||
if data.startswith(b'\xFF\xD8') and data.rstrip(b'\0').endswith(b'\xFF\xD9'):
|
||||
return 'image/jpeg'
|
||||
if data.startswith((b'GIF87a', b'GIF89a')):
|
||||
return 'image/gif'
|
||||
if data.startswith(b'RIFF') and data[8:12] == b'WEBP':
|
||||
return 'image/webp'
|
||||
raise errors.InvalidImageError
|
||||
|
||||
|
||||
def image_to_base64_url(data):
|
||||
fmt = 'data:{mime};base64,{data}'
|
||||
mime = mime_type_for_image(data)
|
||||
b64 = base64.b64encode(data).decode('ascii')
|
||||
return fmt.format(mime=mime, data=b64)
|
||||
fmt = 'data:{mime};base64,{data}'
|
||||
mime = mime_type_for_image(data)
|
||||
b64 = base64.b64encode(data).decode('ascii')
|
||||
return fmt.format(mime=mime, data=b64)
|
||||
|
||||
|
||||
def main() -> typing.NoReturn:
|
||||
"""resize or convert an image from stdin and write the resized or converted version to stdout."""
|
||||
import sys
|
||||
"""resize or convert an image from stdin and write the resized or converted version to stdout."""
|
||||
import sys
|
||||
|
||||
if sys.argv[1] == 'resize':
|
||||
f = resize_until_small
|
||||
elif sys.argv[1] == 'convert':
|
||||
f = convert_to_gif
|
||||
else:
|
||||
sys.exit(1)
|
||||
if sys.argv[1] == 'resize':
|
||||
f = resize_until_small
|
||||
elif sys.argv[1] == 'convert':
|
||||
f = convert_to_gif
|
||||
else:
|
||||
sys.exit(1)
|
||||
|
||||
data = io.BytesIO(sys.stdin.buffer.read())
|
||||
try:
|
||||
f(data)
|
||||
except errors.InvalidImageError:
|
||||
# 2 is used because 1 is already used by python's default error handler
|
||||
sys.exit(2)
|
||||
data = io.BytesIO(sys.stdin.buffer.read())
|
||||
try:
|
||||
f(data)
|
||||
except errors.InvalidImageError:
|
||||
# 2 is used because 1 is already used by python's default error handler
|
||||
sys.exit(2)
|
||||
|
||||
stdout_write = sys.stdout.buffer.write # getattr optimization
|
||||
stdout_write = sys.stdout.buffer.write # getattr optimization
|
||||
|
||||
while True:
|
||||
buf = data.read(16 * 1024)
|
||||
if not buf:
|
||||
break
|
||||
while True:
|
||||
buf = data.read(16 * 1024)
|
||||
if not buf:
|
||||
break
|
||||
|
||||
stdout_write(buf)
|
||||
stdout_write(buf)
|
||||
|
||||
sys.exit(0)
|
||||
|
||||
sys.exit(0)
|
||||
|
||||
async def process_image_in_subprocess(command_name, image_data: bytes):
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
sys.executable, '-m', __name__, command_name,
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
sys.executable, '-m', __name__, command_name,
|
||||
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE)
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE)
|
||||
|
||||
try:
|
||||
image_data, err = await asyncio.wait_for(proc.communicate(image_data), timeout=float('inf'))
|
||||
except asyncio.TimeoutError:
|
||||
proc.send_signal(signal.SIGINT)
|
||||
raise errors.ImageResizeTimeoutError if command_name == 'resize' else errors.ImageConversionTimeoutError
|
||||
else:
|
||||
if proc.returncode == 2:
|
||||
raise errors.InvalidImageError
|
||||
if proc.returncode != 0:
|
||||
raise RuntimeError(err.decode('utf-8') + f'Return code: {proc.returncode}')
|
||||
try:
|
||||
image_data, err = await asyncio.wait_for(proc.communicate(image_data), timeout=float('inf'))
|
||||
except asyncio.TimeoutError:
|
||||
proc.send_signal(signal.SIGINT)
|
||||
raise errors.ImageResizeTimeoutError if command_name == 'resize' else errors.ImageConversionTimeoutError
|
||||
else:
|
||||
if proc.returncode == 2:
|
||||
raise errors.InvalidImageError
|
||||
if proc.returncode != 0:
|
||||
raise RuntimeError(err.decode('utf-8') +
|
||||
f'Return code: {proc.returncode}')
|
||||
|
||||
return image_data
|
||||
return image_data
|
||||
|
||||
resize_in_subprocess = functools.partial(process_image_in_subprocess, 'resize')
|
||||
convert_to_gif_in_subprocess = functools.partial(process_image_in_subprocess, 'convert')
|
||||
convert_to_gif_in_subprocess = functools.partial(
|
||||
process_image_in_subprocess, 'convert')
|
||||
|
||||
|
||||
def size(fp):
|
||||
"""return the size, in bytes, of the data a file-like object represents"""
|
||||
with preserve_position(fp):
|
||||
fp.seek(0, io.SEEK_END)
|
||||
return fp.tell()
|
||||
"""return the size, in bytes, of the data a file-like object represents"""
|
||||
with preserve_position(fp):
|
||||
fp.seek(0, io.SEEK_END)
|
||||
return fp.tell()
|
||||
|
||||
|
||||
class preserve_position(contextlib.AbstractContextManager):
|
||||
def __init__(self, fp):
|
||||
self.fp = fp
|
||||
self.old_pos = fp.tell()
|
||||
def __init__(self, fp):
|
||||
self.fp = fp
|
||||
self.old_pos = fp.tell()
|
||||
|
||||
def __exit__(self, *excinfo):
|
||||
self.fp.seek(self.old_pos)
|
||||
|
||||
def __exit__(self, *excinfo):
|
||||
self.fp.seek(self.old_pos)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
main()
|
||||
|
|
|
|||
|
|
@ -5,47 +5,51 @@
|
|||
|
||||
import asyncio
|
||||
|
||||
import discord
|
||||
import nextcord
|
||||
|
||||
|
||||
def format_user(user, *, mention=False):
|
||||
"""Format a user object for audit log purposes."""
|
||||
# not mention: @null byte#8191 (140516693242937345)
|
||||
# mention: <@140516693242937345> (null byte#8191)
|
||||
# this allows people to still see the username and discrim
|
||||
# if they don't share a server with that user
|
||||
if mention:
|
||||
return f'{user.mention} (@{user})'
|
||||
else:
|
||||
return f'@{user} ({user.id})'
|
||||
"""Format a user object for audit log purposes."""
|
||||
# not mention: @null byte#8191 (140516693242937345)
|
||||
# mention: <@140516693242937345> (null byte#8191)
|
||||
# this allows people to still see the username and discrim
|
||||
# if they don't share a server with that user
|
||||
if mention:
|
||||
return f'{user.mention} (@{user})'
|
||||
else:
|
||||
return f'@{user} ({user.id})'
|
||||
|
||||
def format_http_exception(exception: discord.HTTPException):
|
||||
"""Formats a discord.HTTPException for relaying to the user.
|
||||
Sample return value:
|
||||
|
||||
BAD REQUEST (status code: 400):
|
||||
Invalid Form Body
|
||||
In image: File cannot be larger than 256 kb.
|
||||
"""
|
||||
return (
|
||||
f'{exception.response.reason} (status code: {exception.response.status}):'
|
||||
f'\n{exception.text}')
|
||||
def format_http_exception(exception: nextcord.HTTPException):
|
||||
"""Formats a nextcord.HTTPException for relaying to the user.
|
||||
Sample return value:
|
||||
|
||||
BAD REQUEST (status code: 400):
|
||||
Invalid Form Body
|
||||
In image: File cannot be larger than 256 kb.
|
||||
"""
|
||||
return (
|
||||
f'{exception.response.reason} (status code: {exception.response.status}):'
|
||||
f'\n{exception.text}')
|
||||
|
||||
|
||||
def strip_angle_brackets(string):
|
||||
"""Strip leading < and trailing > from a string.
|
||||
Useful if a user sends you a url like <this> to avoid embeds, or to convert emotes to reactions."""
|
||||
if string.startswith('<') and string.endswith('>'):
|
||||
return string[1:-1]
|
||||
return string
|
||||
"""Strip leading < and trailing > from a string.
|
||||
Useful if a user sends you a url like <this> to avoid embeds, or to convert emotes to reactions."""
|
||||
if string.startswith('<') and string.endswith('>'):
|
||||
return string[1:-1]
|
||||
return string
|
||||
|
||||
|
||||
async def gather_or_cancel(*awaitables, loop=None):
|
||||
"""run the awaitables in the sequence concurrently. If any of them raise an exception,
|
||||
propagate the first exception raised and cancel all other awaitables.
|
||||
"""
|
||||
gather_task = asyncio.gather(*awaitables, loop=loop)
|
||||
try:
|
||||
return await gather_task
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except:
|
||||
gather_task.cancel()
|
||||
raise
|
||||
"""run the awaitables in the sequence concurrently. If any of them raise an exception,
|
||||
propagate the first exception raised and cancel all other awaitables.
|
||||
"""
|
||||
gather_task = asyncio.gather(*awaitables, loop=loop)
|
||||
try:
|
||||
return await gather_task
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except:
|
||||
gather_task.cancel()
|
||||
raise
|
||||
|
|
|
|||
|
|
@ -5,145 +5,148 @@ import asyncio
|
|||
import contextlib
|
||||
import typing
|
||||
|
||||
import discord
|
||||
from discord.ext.commands import Context
|
||||
import nextcord
|
||||
from nextcord.ext.commands import Context
|
||||
|
||||
# Copyright © 2016-2017 Pandentia and contributors
|
||||
# https://github.com/Thessia/Liara/blob/75fa11948b8b2ea27842d8815a32e51ef280a999/cogs/utils/paginator.py
|
||||
|
||||
|
||||
class Paginator:
|
||||
def __init__(self, ctx: Context, pages: typing.Iterable, *, timeout=300, delete_message=False,
|
||||
delete_message_on_timeout=False, text_message=None):
|
||||
def __init__(self, ctx: Context, pages: typing.Iterable, *, timeout=300, delete_message=False,
|
||||
delete_message_on_timeout=False, text_message=None):
|
||||
|
||||
self.pages = list(pages)
|
||||
self.timeout = timeout
|
||||
self.author = ctx.author
|
||||
self.target = ctx.channel
|
||||
self.delete_msg = delete_message
|
||||
self.delete_msg_timeout = delete_message_on_timeout
|
||||
self.text_message = text_message
|
||||
self.pages = list(pages)
|
||||
self.timeout = timeout
|
||||
self.author = ctx.author
|
||||
self.target = ctx.channel
|
||||
self.delete_msg = delete_message
|
||||
self.delete_msg_timeout = delete_message_on_timeout
|
||||
self.text_message = text_message
|
||||
|
||||
self._stopped = None # we use this later
|
||||
self._embed = None
|
||||
self._message = None
|
||||
self._client = ctx.bot
|
||||
self._stopped = None # we use this later
|
||||
self._embed = None
|
||||
self._message = None
|
||||
self._client = ctx.bot
|
||||
|
||||
self.footer = 'Page {} of {}'
|
||||
self.navigation = {
|
||||
'\N{BLACK LEFT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}': self.first_page,
|
||||
'\N{BLACK LEFT-POINTING TRIANGLE}': self.previous_page,
|
||||
'\N{BLACK RIGHT-POINTING TRIANGLE}': self.next_page,
|
||||
'\N{BLACK RIGHT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}': self.last_page,
|
||||
'\N{BLACK SQUARE FOR STOP}': self.stop
|
||||
}
|
||||
self.footer = 'Page {} of {}'
|
||||
self.navigation = {
|
||||
'\N{BLACK LEFT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}': self.first_page,
|
||||
'\N{BLACK LEFT-POINTING TRIANGLE}': self.previous_page,
|
||||
'\N{BLACK RIGHT-POINTING TRIANGLE}': self.next_page,
|
||||
'\N{BLACK RIGHT-POINTING DOUBLE TRIANGLE WITH VERTICAL BAR}': self.last_page,
|
||||
'\N{BLACK SQUARE FOR STOP}': self.stop
|
||||
}
|
||||
|
||||
self._page = None
|
||||
self._page = None
|
||||
|
||||
def react_check(self, reaction: discord.RawReactionActionEvent):
|
||||
if reaction.user_id != self.author.id:
|
||||
return False
|
||||
def react_check(self, reaction: nextcord.RawReactionActionEvent):
|
||||
if reaction.user_id != self.author.id:
|
||||
return False
|
||||
|
||||
if reaction.message_id != self._message.id:
|
||||
return False
|
||||
if reaction.message_id != self._message.id:
|
||||
return False
|
||||
|
||||
target_emoji = str(reaction.emoji)
|
||||
return bool(discord.utils.find(lambda emoji: target_emoji == emoji, self.navigation))
|
||||
target_emoji = str(reaction.emoji)
|
||||
return bool(nextcord.utils.find(lambda emoji: target_emoji == emoji, self.navigation))
|
||||
|
||||
async def begin(self):
|
||||
"""Starts pagination"""
|
||||
self._stopped = False
|
||||
self._embed = discord.Embed()
|
||||
await self.first_page()
|
||||
for button in self.navigation:
|
||||
await self._message.add_reaction(button)
|
||||
while not self._stopped:
|
||||
try:
|
||||
reaction: RawReactionActionEvent = await self._client.wait_for(
|
||||
'raw_reaction_add',
|
||||
check=self.react_check,
|
||||
timeout=self.timeout)
|
||||
except asyncio.TimeoutError:
|
||||
await self.stop(delete=self.delete_msg_timeout)
|
||||
continue
|
||||
async def begin(self):
|
||||
"""Starts pagination"""
|
||||
self._stopped = False
|
||||
self._embed = nextcord.Embed()
|
||||
await self.first_page()
|
||||
for button in self.navigation:
|
||||
await self._message.add_reaction(button)
|
||||
while not self._stopped:
|
||||
try:
|
||||
reaction: RawReactionActionEvent = await self._client.wait_for(
|
||||
'raw_reaction_add',
|
||||
check=self.react_check,
|
||||
timeout=self.timeout)
|
||||
except asyncio.TimeoutError:
|
||||
await self.stop(delete=self.delete_msg_timeout)
|
||||
continue
|
||||
|
||||
await self.navigation[str(reaction.emoji)]()
|
||||
await self.navigation[str(reaction.emoji)]()
|
||||
|
||||
await asyncio.sleep(0.2)
|
||||
with contextlib.suppress(discord.HTTPException):
|
||||
await self._message.remove_reaction(reaction.emoji, discord.Object(reaction.user_id))
|
||||
await asyncio.sleep(0.2)
|
||||
with contextlib.suppress(nextcord.HTTPException):
|
||||
await self._message.remove_reaction(reaction.emoji, nextcord.Object(reaction.user_id))
|
||||
|
||||
async def stop(self, *, delete=None):
|
||||
"""Aborts pagination."""
|
||||
if delete is None:
|
||||
delete = self.delete_msg
|
||||
async def stop(self, *, delete=None):
|
||||
"""Aborts pagination."""
|
||||
if delete is None:
|
||||
delete = self.delete_msg
|
||||
|
||||
if delete:
|
||||
with contextlib.suppress(discord.HTTPException):
|
||||
await self._message.delete()
|
||||
else:
|
||||
await self._clear_reactions()
|
||||
self._stopped = True
|
||||
if delete:
|
||||
with contextlib.suppress(nextcord.HTTPException):
|
||||
await self._message.delete()
|
||||
else:
|
||||
await self._clear_reactions()
|
||||
self._stopped = True
|
||||
|
||||
async def _clear_reactions(self):
|
||||
try:
|
||||
await self._message.clear_reactions()
|
||||
except discord.Forbidden:
|
||||
for button in self.navigation:
|
||||
with contextlib.suppress(discord.HTTPException):
|
||||
await self._message.remove_reaction(button, self._message.author)
|
||||
except discord.HTTPException:
|
||||
pass
|
||||
async def _clear_reactions(self):
|
||||
try:
|
||||
await self._message.clear_reactions()
|
||||
except nextcord.Forbidden:
|
||||
for button in self.navigation:
|
||||
with contextlib.suppress(nextcord.HTTPException):
|
||||
await self._message.remove_reaction(button, self._message.author)
|
||||
except nextcord.HTTPException:
|
||||
pass
|
||||
|
||||
async def format_page(self):
|
||||
self._embed.description = self.pages[self._page]
|
||||
self._embed.set_footer(text=self.footer.format(self._page + 1, len(self.pages)))
|
||||
async def format_page(self):
|
||||
self._embed.description = self.pages[self._page]
|
||||
self._embed.set_footer(text=self.footer.format(
|
||||
self._page + 1, len(self.pages)))
|
||||
|
||||
kwargs = {'embed': self._embed}
|
||||
if self.text_message:
|
||||
kwargs['content'] = self.text_message
|
||||
kwargs = {'embed': self._embed}
|
||||
if self.text_message:
|
||||
kwargs['content'] = self.text_message
|
||||
|
||||
if self._message:
|
||||
await self._message.edit(**kwargs)
|
||||
else:
|
||||
self._message = await self.target.send(**kwargs)
|
||||
if self._message:
|
||||
await self._message.edit(**kwargs)
|
||||
else:
|
||||
self._message = await self.target.send(**kwargs)
|
||||
|
||||
async def first_page(self):
|
||||
self._page = 0
|
||||
await self.format_page()
|
||||
async def first_page(self):
|
||||
self._page = 0
|
||||
await self.format_page()
|
||||
|
||||
async def next_page(self):
|
||||
self._page += 1
|
||||
if self._page == len(self.pages): # avoid the inevitable IndexError
|
||||
self._page = 0
|
||||
await self.format_page()
|
||||
async def next_page(self):
|
||||
self._page += 1
|
||||
if self._page == len(self.pages): # avoid the inevitable IndexError
|
||||
self._page = 0
|
||||
await self.format_page()
|
||||
|
||||
async def previous_page(self):
|
||||
self._page -= 1
|
||||
if self._page < 0: # ditto
|
||||
self._page = len(self.pages) - 1
|
||||
await self.format_page()
|
||||
async def previous_page(self):
|
||||
self._page -= 1
|
||||
if self._page < 0: # ditto
|
||||
self._page = len(self.pages) - 1
|
||||
await self.format_page()
|
||||
|
||||
async def last_page(self):
|
||||
self._page = len(self.pages) - 1
|
||||
await self.format_page()
|
||||
|
||||
async def last_page(self):
|
||||
self._page = len(self.pages) - 1
|
||||
await self.format_page()
|
||||
|
||||
class ListPaginator(Paginator):
|
||||
def __init__(self, ctx, _list: list, per_page=10, **kwargs):
|
||||
pages = []
|
||||
page = ''
|
||||
c = 0
|
||||
l = len(_list)
|
||||
for i in _list:
|
||||
if c > l:
|
||||
break
|
||||
if c % per_page == 0 and page:
|
||||
pages.append(page.strip())
|
||||
page = ''
|
||||
page += '{}. {}\n'.format(c+1, i)
|
||||
def __init__(self, ctx, _list: list, per_page=10, **kwargs):
|
||||
pages = []
|
||||
page = ''
|
||||
c = 0
|
||||
l = len(_list)
|
||||
for i in _list:
|
||||
if c > l:
|
||||
break
|
||||
if c % per_page == 0 and page:
|
||||
pages.append(page.strip())
|
||||
page = ''
|
||||
page += '{}. {}\n'.format(c+1, i)
|
||||
|
||||
c += 1
|
||||
pages.append(page.strip())
|
||||
# shut up, IDEA
|
||||
# noinspection PyArgumentList
|
||||
super().__init__(ctx, pages, **kwargs)
|
||||
self.footer += ' ({} entries)'.format(l)
|
||||
c += 1
|
||||
pages.append(page.strip())
|
||||
# shut up, IDEA
|
||||
# noinspection PyArgumentList
|
||||
super().__init__(ctx, pages, **kwargs)
|
||||
self.footer += ' ({} entries)'.format(l)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue