update(requirements): clean up non useful requirement precision

This commit is contained in:
Romain J 2020-08-29 01:01:34 +02:00
parent 331599eb38
commit d68d54be44
18 changed files with 37 additions and 865 deletions

View file

@ -1,15 +1,18 @@
<component name="ProjectDictionaryState">
<dictionary name="romain">
<words>
<w>appdirs</w>
<w>asctime</w>
<w>commandstats</w>
<w>francais</w>
<w>gnous</w>
<w>ipinfo</w>
<w>iplocalise</w>
<w>jishaku</w>
<w>levelname</w>
<w>localiseip</w>
<w>postgresql</w>
<w>pred</w>
<w>releaselevel</w>
<w>socketstats</w>
<w>splt</w>

View file

@ -9,49 +9,16 @@ url = https://git.gnous.eu/gnouseu/tuxbot-bot/
[options]
packages = find_namespace:
python_requires = >=3.7
;todo: remove flatten_dict (core/config.py)
install_requires =
aiohttp==3.6.2
aiosqlite==0.13.0
appdirs==1.4.4
astunparse==1.6.3
async-timeout==3.0.1
asyncpg==0.20.1
attrs==19.3.0
babel==2.8.0
black==19.10b0
braceexpand==0.1.5
cachetools==4.1.0
certifi==2020.4.5.1
chardet==3.0.4
ciso8601==2.1.3
colorama==0.4.3
discord-flags==2.1.1
discord.py==1.3.4
dnspython==1.16.0
flatten-dict==0.3.0
humanize==2.4.0
idna==2.9
import-expression==1.1.3
ipinfo==3.0.0
ipwhois==1.1.0
iso8601==0.1.12
jishaku==1.18.2.188
multidict==4.7.6
pathspec<1,>=0.6
psutil==5.7.0
PyPika==0.37.7
pytz==2020.1
regex==2020.6.7
requests==2.23.0
Babel==2.8.0
discord.py==1.4.1
discord_flags==2.1.1
flatten_dict==0.3.0
jishaku==1.19.1.200
PyYAML==5.3.1
rich==6.0.0
six==1.15.0
toml>=0.9.4
tortoise-orm==0.16.13
typed-ast>=1.4.0
typing-extensions==3.7.4.2
urllib3==1.25.9
wheel==0.34.2
yarl==1.4.2
[options.entry_points]
console_scripts =

View file

@ -290,7 +290,7 @@ def main() -> NoReturn:
" [red]Please use <prefix>quit instead of Ctrl+C to Shutdown!"
)
log.warning("Please use <prefix>quit instead of Ctrl+C to Shutdown!")
log.error("Received KeyboardInterrupt")
log.info("Received KeyboardInterrupt")
console.print("[i]Trying to shutdown...")
if tux is not None:
loop.run_until_complete(shutdown_handler(tux, signal.SIGINT))
@ -299,8 +299,8 @@ def main() -> NoReturn:
if tux is not None:
loop.run_until_complete(shutdown_handler(tux, None, exc.code))
except Exception as exc:
log.error("Unexpected exception (%s): ", type(exc))
console.print_exception()
log.exception("Unexpected exception (%s): ", type(exc), exc_info=exc)
if tux is not None:
loop.run_until_complete(shutdown_handler(tux, None, 1))
finally:

View file

@ -1,18 +0,0 @@
from collections import namedtuple
from .anti_raid import AntiRaid
from ...core.bot import Tux
VersionInfo = namedtuple("VersionInfo", "major minor micro releaselevel")
version_info = VersionInfo(major=1, minor=0, micro=0, releaselevel="alpha")
__version__ = "v{}.{}.{}-{}".format(
version_info.major,
version_info.minor,
version_info.micro,
version_info.releaselevel,
).replace("\n", "")
def setup(bot: Tux):
bot.add_cog(AntiRaid(bot))

View file

@ -1,24 +0,0 @@
import logging
from discord.ext import commands
from tuxbot.core import checks
from tuxbot.core.bot import Tux
from tuxbot.core.i18n import Translator
log = logging.getLogger("tuxbot.cogs.anti_raid")
T_ = Translator("AntiRaid", __file__)
class AntiRaid(commands.Cog, name="AntiRaid"):
def __init__(self, bot: Tux):
self.bot = bot
@commands.group(
name="anti_raid",
alias=["anti-raid", "raid_protect", "raid-protect", "no_raid", "no-raid"],
)
@commands.guild_only()
@checks.is_admin()
async def _anti_raid(self, ctx: commands.Context):
pass

View file

@ -1,18 +0,0 @@
from collections import namedtuple
from .images import Images
from ...core.bot import Tux
VersionInfo = namedtuple("VersionInfo", "major minor micro releaselevel")
version_info = VersionInfo(major=1, minor=0, micro=0, releaselevel="alpha")
__version__ = "v{}.{}.{}-{}".format(
version_info.major,
version_info.minor,
version_info.micro,
version_info.releaselevel,
).replace("\n", "")
def setup(bot: Tux):
bot.add_cog(Images(bot))

View file

@ -1,188 +0,0 @@
import logging
from io import BytesIO
import discord
from discord.ext import commands, flags
from app import TuxBot
from utils.functions.extra import ContextPlus, command_extra
log = logging.getLogger(__name__)
class Images(commands.Cog, name="Images"):
def __init__(self, bot):
self.bot = bot
self.image_api = "http://0.0.0.0:8080"
async def _send_meme(self, ctx: ContextPlus, endpoint: str, **passed_flags):
async with ctx.typing():
url = f"{self.image_api}/{endpoint}?"
for key, val in passed_flags.items():
if val:
url += f"{key}={val}&"
async with self.bot.session.get(url) as r:
if r.status != 200:
return await ctx.send("Failed...")
data = BytesIO(await r.read())
await ctx.send(file=discord.File(data, "output.png"))
@command_extra(name="phcomment")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _phcomment(
self,
ctx: ContextPlus,
user: discord.User = None,
*,
message: commands.clean_content(
fix_channel_mentions=True, escape_markdown=True
),
):
async with ctx.typing():
message = message.replace("&", "%26")
if user is None:
avatar = ctx.author.avatar_url_as(format="png")
username = ctx.author.name
else:
avatar = user.avatar_url_as(format="png")
username = user.name
url = (
f"{self.image_api}/ph/comment"
f"?image={avatar}"
f"&username={username}"
f"&message={message}"
)
async with self.bot.session.get(url) as r:
if r.status != 200:
return await ctx.send("Failed...")
data = BytesIO(await r.read())
await ctx.send(file=discord.File(data, "output.png"))
@command_extra(name="phvideo")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _phvideo(
self,
ctx: ContextPlus,
image: str,
author: discord.User,
*,
title: commands.clean_content(fix_channel_mentions=True, escape_markdown=True),
):
async with ctx.typing():
url = (
f"{self.image_api}/ph/video"
f"?image={image}"
f"&username={author.name}"
f"&title={title}"
)
async with self.bot.session.get(url) as r:
if r.status != 200:
return await ctx.send("Failed...")
data = BytesIO(await r.read())
await ctx.send(file=discord.File(data, "output.png"))
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str)
@flags.add_flag("--text3", type=str)
@command_extra(name="balloon")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _balloon(self, ctx: ContextPlus, **passed_flags):
passed_flags["text3"] = passed_flags.get("text3")
passed_flags["text4"] = passed_flags.get("text1")
passed_flags["text5"] = passed_flags.get("text2")
await self._send_meme(ctx, "balloon", **passed_flags)
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str)
@flags.add_flag("--text3", type=str)
@command_extra(name="butterfly")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _butterfly(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "butterfly", **passed_flags)
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str)
@command_extra(name="buttons")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _buttons(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "buttons", **passed_flags)
@flags.add_flag("--text1", type=str)
@command_extra(name="cmm")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _cmm(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "change_my_mind", **passed_flags)
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str)
@command_extra(name="drake")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _drake(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "drake", **passed_flags)
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str, default=False)
@command_extra(name="fry")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _fry(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "fry", **passed_flags)
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str, default=False)
@command_extra(name="imagination")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _imagination(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "imagination", **passed_flags)
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str, default=False)
@command_extra(name="everywhere")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _everywhere(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "everywhere", **passed_flags)
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str)
@flags.add_flag("--text3", type=str)
@command_extra(name="choice")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _choice(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "choice", **passed_flags)
@flags.add_flag("--text1", type=str)
@command_extra(name="pika")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _pika(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "pika", **passed_flags)
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str)
@flags.add_flag("--text3", type=str)
@command_extra(name="pkp")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _pkp(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "pkp", **passed_flags)
@flags.add_flag("--text1", type=str)
@flags.add_flag("--text2", type=str)
@command_extra(name="puppet")
@commands.cooldown(1, 5, commands.BucketType.user)
async def _puppet(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "puppet", **passed_flags)
@flags.add_flag("--text1", type=str)
@command_extra(name="scroll_of_truth", alias=["sot"])
@commands.cooldown(1, 5, commands.BucketType.user)
async def _sot(self, ctx: ContextPlus, **passed_flags):
await self._send_meme(ctx, "scroll_of_truth", **passed_flags)

View file

@ -1,26 +0,0 @@
import logging
from collections import namedtuple
from discord.ext import commands
from .logs import Logs, GatewayHandler, on_error
from ...core.bot import Tux
VersionInfo = namedtuple("VersionInfo", "major minor micro releaselevel")
version_info = VersionInfo(major=2, minor=0, micro=0, releaselevel="alpha")
__version__ = "v{}.{}.{}-{}".format(
version_info.major,
version_info.minor,
version_info.micro,
version_info.releaselevel,
).replace("\n", "")
def setup(bot: Tux):
cog = Logs(bot)
bot.add_cog(cog)
handler = GatewayHandler(cog)
logging.getLogger().addHandler(handler)
commands.AutoShardedBot.on_error = on_error

View file

@ -1,22 +0,0 @@
{
"dm": {
"description": "Webhook url for DMs events",
"value": "str"
},
"mentions": {
"description": "Webhook url for mentions events",
"value": "str"
},
"guilds": {
"description": "Webhook url for guilds events",
"value": "str"
},
"errors": {
"description": "Webhook url for errors events",
"value": "str"
},
"gateway": {
"description": "Webhook url for gateway events",
"value": "str"
}
}

View file

@ -1,319 +0,0 @@
"""
Based on https://github.com/Rapptz/RoboDanny/blob/3d94e89ef27f702a5f57f432a9131bdfb60bb3ec/cogs/stats.py
Adapted by Romain J.
"""
import asyncio
import datetime
import json
import logging
import textwrap
import traceback
from collections import defaultdict
import discord
import humanize
import psutil
from discord.ext import commands, tasks
from app import TuxBot
from utils.functions.extra import command_extra
log = logging.getLogger(__name__)
class GatewayHandler(logging.Handler):
def __init__(self, cog):
self.cog = cog
super().__init__(logging.INFO)
def filter(self, record):
return (
record.name == "discord.gateway"
or "Shard ID" in record.msg
or "Websocket closed " in record.msg
)
def emit(self, record):
self.cog.add_record(record)
class Logs(commands.Cog):
def __init__(self, bot: TuxBot):
self.bot = bot
self.process = psutil.Process()
self._batch_lock = asyncio.Lock(loop=bot.loop)
self._data_batch = []
self._gateway_queue = asyncio.Queue(loop=bot.loop)
self.gateway_worker.start()
self._resumes = []
self._identifies = defaultdict(list)
def _clear_gateway_data(self):
one_week_ago = datetime.datetime.utcnow() - datetime.timedelta(days=7)
to_remove = [
index for index, dt in enumerate(self._resumes) if dt < one_week_ago
]
for index in reversed(to_remove):
del self._resumes[index]
for shard_id, dates in self._identifies.items():
to_remove = [index for index, dt in enumerate(dates) if dt < one_week_ago]
for index in reversed(to_remove):
del dates[index]
@tasks.loop(seconds=0.0)
async def gateway_worker(self):
record = await self._gateway_queue.get()
await self.notify_gateway_status(record)
async def register_command(self, ctx):
if ctx.command is None:
return
command = ctx.command.qualified_name
self.bot.command_stats[command] += 1
message = ctx.message
if ctx.guild is None:
destination = "Private Message"
guild_id = None
else:
destination = f"#{message.channel} ({message.guild})"
guild_id = ctx.guild.id
log.info(
f"{message.created_at}: {message.author} "
f"in {destination}: {message.content}"
)
async with self._batch_lock:
self._data_batch.append(
{
"guild": guild_id,
"channel": ctx.channel.id,
"author": ctx.author.id,
"used": message.created_at.isoformat(),
"prefix": ctx.prefix,
"command": command,
"failed": ctx.command_failed,
}
)
@commands.Cog.listener()
async def on_command_completion(self, ctx):
await self.register_command(ctx)
@commands.Cog.listener()
async def on_socket_response(self, msg):
self.bot.socket_stats[msg.get("t")] += 1
@property
def logs(self):
webhooks = {}
for key, value in self.bot.logs_channels.items():
webhooks[key] = discord.Webhook.partial(
id=value.get("webhook")["id"],
token=value.get("webhook")["token"],
adapter=discord.AsyncWebhookAdapter(self.bot.session),
)
return webhooks
async def log_error(self, *, ctx=None, extra=None):
e = discord.Embed(title="Error", colour=0xDD5F53)
e.description = f"```py\n{traceback.format_exc()}\n```"
e.add_field(name="Extra", value=extra, inline=False)
e.timestamp = datetime.datetime.utcnow()
if ctx is not None:
fmt = "{0} (ID: {0.id})"
author = fmt.format(ctx.author)
channel = fmt.format(ctx.channel)
guild = "None" if ctx.guild is None else fmt.format(ctx.guild)
e.add_field(name="Author", value=author)
e.add_field(name="Channel", value=channel)
e.add_field(name="Guild", value=guild)
await self.logs.get("errors").send(embed=e)
async def send_guild_stats(self, e, guild):
e.add_field(name="Name", value=guild.name)
e.add_field(name="ID", value=guild.id)
e.add_field(name="Shard ID", value=guild.shard_id or "N/A")
e.add_field(name="Owner", value=f"{guild.owner} (ID: {guild.owner.id})")
bots = sum(member.bot for member in guild.members)
total = guild.member_count
online = sum(member.status is discord.Status.online for member in guild.members)
e.add_field(name="Members", value=str(total))
e.add_field(name="Bots", value=f"{bots} ({bots / total:.2%})")
e.add_field(name="Online", value=f"{online} ({online / total:.2%})")
if guild.icon:
e.set_thumbnail(url=guild.icon_url)
if guild.me:
e.timestamp = guild.me.joined_at
await self.logs.get("guilds").send(embed=e)
@commands.Cog.listener()
async def on_guild_join(self, guild: discord.guild):
e = discord.Embed(colour=0x53DDA4, title="New Guild") # green colour
await self.send_guild_stats(e, guild)
@commands.Cog.listener()
async def on_guild_remove(self, guild: discord.guild):
e = discord.Embed(colour=0xDD5F53, title="Left Guild") # red colour
await self.send_guild_stats(e, guild)
@commands.Cog.listener()
async def on_message(self, message: discord.message):
ctx = await self.bot.get_context(message)
if ctx.valid:
return
if isinstance(message.channel, discord.DMChannel):
if message.author is self.bot.user:
e = discord.Embed(
title=f"DM to: {message.channel.recipient}",
description=message.content,
color=0x39E326,
)
else:
e = discord.Embed(
title="New DM:", description=message.content, color=0x0A97F5
)
e.set_author(
name=message.channel.recipient,
icon_url=message.channel.recipient.avatar_url_as(format="png"),
)
if message.attachments:
attachment_url = message.attachments[0].url
e.set_image(url=attachment_url)
e.set_footer(text=f"User ID: {message.channel.recipient.id}")
await self.logs["dm"].send(embed=e)
@commands.Cog.listener()
async def on_command_error(self, ctx, error):
await self.register_command(ctx)
if not isinstance(
error, (commands.CommandInvokeError, commands.ConversionError)
):
return
error = error.original
if isinstance(error, (discord.Forbidden, discord.NotFound)):
return
e = discord.Embed(title="Command Error", colour=0xCC3366)
e.add_field(name="Name", value=ctx.command.qualified_name)
e.add_field(name="Author", value=f"{ctx.author} (ID: {ctx.author.id})")
fmt = f"Channel: {ctx.channel} (ID: {ctx.channel.id})"
if ctx.guild:
fmt = f"{fmt}\nGuild: {ctx.guild} (ID: {ctx.guild.id})"
e.add_field(name="Location", value=fmt, inline=False)
e.add_field(
name="Content", value=textwrap.shorten(ctx.message.content, width=512)
)
exc = "".join(
traceback.format_exception(
type(error), error, error.__traceback__, chain=False
)
)
e.description = f"```py\n{exc}\n```"
e.timestamp = datetime.datetime.utcnow()
await self.logs.get("errors").send(embed=e)
@commands.Cog.listener()
async def on_socket_raw_send(self, data):
if '"op":2' not in data and '"op":6' not in data:
return
back_to_json = json.loads(data)
if back_to_json["op"] == 2:
payload = back_to_json["d"]
inner_shard = payload.get("shard", [0])
self._identifies[inner_shard[0]].append(datetime.datetime.utcnow())
else:
self._resumes.append(datetime.datetime.utcnow())
self._clear_gateway_data()
def add_record(self, record):
self._gateway_queue.put_nowait(record)
async def notify_gateway_status(self, record):
types = {"INFO": ":information_source:", "WARNING": ":warning:"}
emoji = types.get(record.levelname, ":heavy_multiplication_x:")
dt = datetime.datetime.utcfromtimestamp(record.created)
msg = f"{emoji} `[{dt:%Y-%m-%d %H:%M:%S}] {record.message}`"
await self.logs.get("gateway").send(msg)
@command_extra(name="commandstats")
@commands.is_owner()
async def _commandstats(self, ctx, limit=20):
counter = self.bot.command_stats
width = len(max(counter, key=len))
if limit > 0:
common = counter.most_common(limit)
else:
common = counter.most_common()[limit:]
output = "\n".join(f"{k:<{width}}: {c}" for k, c in common)
await ctx.send(f"```\n{output}\n```")
@commands.command("socketstats")
@commands.is_owner()
async def _socketstats(self, ctx):
delta = datetime.datetime.utcnow() - self.bot.uptime
minutes = delta.total_seconds() / 60
total = sum(self.bot.socket_stats.values())
cpm = total / minutes
await ctx.send(
f"{total} socket events observed ({cpm:.2f}/minute):\n"
f"{self.bot.socket_stats}"
)
@commands.command("uptime")
async def _uptime(self, ctx):
uptime = humanize.naturaltime(datetime.datetime.utcnow() - self.bot.uptime)
await ctx.send(f"Uptime: **{uptime}**")
async def on_error(self, event, *args):
e = discord.Embed(title="Event Error", colour=0xA32952)
e.add_field(name="Event", value=event)
e.description = f"```py\n{traceback.format_exc()}\n```"
e.timestamp = datetime.datetime.utcnow()
args_str = ["```py"]
for index, arg in enumerate(args):
args_str.append(f"[{index}]: {arg!r}")
args_str.append("```")
e.add_field(name="Args", value="\n".join(args_str), inline=False)
hook = self.get_cog("Logs").logs.get("errors")
try:
await hook.send(embed=e)
except (
discord.HTTPException,
discord.NotFound,
discord.Forbidden,
discord.InvalidArgument,
):
pass

View file

@ -1,18 +0,0 @@
from collections import namedtuple
from .network import Network
from ...core.bot import Tux
VersionInfo = namedtuple("VersionInfo", "major minor micro releaselevel")
version_info = VersionInfo(major=2, minor=0, micro=0, releaselevel="alpha")
__version__ = "v{}.{}.{}-{}".format(
version_info.major,
version_info.minor,
version_info.micro,
version_info.releaselevel,
).replace("\n", "")
def setup(bot: Tux):
bot.add_cog(Network(bot))

View file

@ -1,6 +0,0 @@
{
"ipinfo": {
"description": "API token for ipinfo.io",
"value": "str"
}
}

View file

@ -1,112 +0,0 @@
import logging
import socket
import ipinfo
import discord
from discord.ext import commands, flags
from ipwhois import Net
from ipwhois.asn import IPASN
from ipinfo.exceptions import RequestQuotaExceededError
from requests.exceptions import HTTPError
from app import TuxBot
from utils.functions.extra import ContextPlus, command_extra
log = logging.getLogger(__name__)
class Network(commands.Cog, name="Useless"):
def __init__(self, bot: TuxBot):
self.bot = bot
@flags.add_flag(
"-i", "--ip", type=str, default="v4", choices=["v4", "4", "v6", "6"]
)
@command_extra(name="iplocalise", aliases=["localiseip"])
@commands.cooldown(1, 5, commands.BucketType.user)
async def _iplocalise(self, ctx: ContextPlus, target: str, **passed_flags):
loading = await ctx.send("_Récupération des informations..._", deletable=False)
def get_hostname(dtl, tgt):
try:
return dtl.hostname
except AttributeError:
try:
return socket.gethostbyaddr(tgt)[0]
except (ValueError, socket.herror):
return "N/A"
ip_type = passed_flags.get("ip")
target_copy = target
# clean https://, last /, ...
spltTgt = target.split("://")
target = (
spltTgt[(0, 1)[len(spltTgt) > 1]]
.split("?")[0]
.split("/")[0]
.split(":")[0]
.lower()
)
try:
target = socket.getaddrinfo(
target,
None,
socket.AF_INET if ip_type in ["v4", "4"] else socket.AF_INET6,
)[1][4][0]
except socket.gaierror:
return await ctx.send("Erreur, cette adresse n'est pas disponible.")
net = Net(target)
obj = IPASN(net)
ip_info = obj.lookup()
try:
handler = ipinfo.getHandler(self.bot.config.ipinfo)
details = handler.getDetails(target)
api_result = True
except (RequestQuotaExceededError, HTTPError):
details = None
api_result = False
if api_result:
belongs = f"{details.org}"
osm = (
f"https://www.openstreetmap.org/"
f"?mlat={details.latitude}"
f"&mlon={details.longitude}"
f"#map=5/{details.latitude}/{details.longitude}"
f"&layers=H"
)
region = (
f"[{details.city} - {details.region} " f"({details.country})]({osm})"
)
flag = f"https://www.countryflags.io/" f"{details.country}/shiny/64.png"
else:
belongs = f"{ip_info['asn_description']} (AS{ip_info['asn']})"
region = f"{ip_info['asn_country_code']}"
flag = (
f"https://www.countryflags.io/"
f"{ip_info['asn_country_code']}/shiny/64.png"
)
e = discord.Embed(
title=f"**Information sur __{target_copy}__ :**" f" `{target}`",
color=0x5858D7,
)
e.add_field(name="Appartient à :", value=belongs)
e.add_field(name="RIR :", value=f"{ip_info['asn_registry']}")
e.add_field(name="Region :", value=region)
e.add_field(
name="Nom de l'hôte :", value=get_hostname(details, target), inline=False
)
e.set_thumbnail(url=flag)
await loading.delete()
await ctx.send(embed=e)

View file

@ -1,18 +0,0 @@
from collections import namedtuple
from .warnings import Warnings
from ...core.bot import Tux
VersionInfo = namedtuple("VersionInfo", "major minor micro releaselevel")
version_info = VersionInfo(major=1, minor=0, micro=0, releaselevel="alpha")
__version__ = "v{}.{}.{}-{}".format(
version_info.major,
version_info.minor,
version_info.micro,
version_info.releaselevel,
).replace("\n", "")
def setup(bot: Tux):
bot.add_cog(Warnings(bot))

View file

@ -1,40 +0,0 @@
from typing import Union
import discord
from discord.ext import commands
from tuxbot.core import checks
from tuxbot.core.bot import Tux
class Warnings(commands.Cog, name="Warnings"):
def __init__(self, bot: Tux):
self.bot = bot
@commands.group(name="warn", alias=["warning"])
@commands.guild_only()
@checks.is_mod()
async def _warn(self, ctx: commands.Context):
division_by_zero = 1 / 0
@_warn.command(name="add")
@commands.guild_only()
async def _warn_add(
self,
ctx: commands.Context,
member: Union[discord.User, discord.Member],
reason: str,
):
pass
@_warn.command(name="delete", aliases=["del", "remove"])
@commands.guild_only()
async def action_del(self, ctx: commands.Context, warn_id: int, reason: str = ""):
pass
@_warn.command(name="list", aliases=["all"])
@commands.guild_only()
async def action_del(
self, ctx: commands.Context, member: Union[discord.User, discord.Member] = None
):
pass

View file

@ -25,14 +25,6 @@ log = logging.getLogger("tuxbot")
console = Console()
install(console=console)
NAME = r"""
_____ _ _ _ _
|_ _| ___ _| |__ ___ | |_ | |__ ___ | |_
| || | | \ \/ / '_ \ / _ \| __|____| '_ \ / _ \| __|
| || |_| |> <| |_) | (_) | ||_____| |_) | (_) | |_
|_| \__,_/_/\_\_.__/ \___/ \__| |_.__/ \___/ \__|
"""
packages: List[str] = ["jishaku", "tuxbot.cogs.warnings", "tuxbot.cogs.admin"]
@ -155,7 +147,7 @@ class Tux(commands.AutoShardedBot):
if extension in self.extensions:
status = f"[green]:heavy_check_mark: {extension}"
else:
status = f"[red]:cross_mark: {extension} "
status = f"[red]:heavy_multiplication_x: {extension}"
table.add_row(status)
columns.add_renderable(table)

View file

@ -1,4 +1,4 @@
from typing import Awaitable, Dict
from typing import Dict
import discord
from discord.ext import commands
@ -29,7 +29,9 @@ def is_mod():
async def pred(ctx):
if await ctx.bot.is_owner(ctx.author):
return True
permissions: discord.Permissions = ctx.channel.permissions_for(ctx.author)
permissions: discord.Permissions = ctx.channel.permissions_for(
ctx.author
)
return permissions.manage_messages
return commands.check(pred)
@ -43,7 +45,9 @@ def is_admin():
async def pred(ctx):
if await ctx.bot.is_owner(ctx.author):
return True
permissions: discord.Permissions = ctx.channel.permissions_for(ctx.author)
permissions: discord.Permissions = ctx.channel.permissions_for(
ctx.author
)
return permissions.administrator
return commands.check(pred)
@ -66,7 +70,9 @@ async def check_permissions(ctx: "ContextPlus", **perms: Dict[str, bool]):
return False
resolved = ctx.channel.permissions_for(ctx.author)
return all(getattr(resolved, name, None) == value for name, value in perms.items())
return all(
getattr(resolved, name, None) == value for name, value in perms.items()
)
def guild_owner_or_permissions(**perms: Dict[str, bool]):

View file

@ -1,11 +1,24 @@
import asyncio
import yaml
import discord
from discord import Embed
from discord.ext import commands, flags
class ContextPlus(commands.Context):
async def send(self, content=None, *args, **kwargs):
if content is not None:
content = content.replace(
self.bot.config('core').get('token'), '<token>'
)
if kwargs.get('embed'):
e = str(kwargs.get('embed').to_dict())
e = e.replace(self.bot.config('core').get('token'), '<token>')
e = yaml.load(e, Loader=yaml.FullLoader)
kwargs['embed'] = Embed.from_dict(e)
if (
hasattr(self.command, "deletable") and self.command.deletable
) and kwargs.pop("deletable", True):