Browse Source

update(core): change to >=3.8

master
Romain J 4 months ago
parent
commit
ba53228d44
  1. 3
      .gitignore
  2. 3
      .idea/tuxbot_bot.iml
  3. 2
      Makefile
  4. 4
      README.rst
  5. 2
      setup.cfg
  6. 2
      setup.py
  7. 3
      tuxbot/cogs/Custom/custom.py
  8. 2
      tuxbot/cogs/Logs/functions/utils.py
  9. 8
      tuxbot/cogs/Logs/logs.py
  10. 11
      tuxbot/cogs/Mod/functions/converters.py
  11. 4
      tuxbot/cogs/Mod/functions/exceptions.py
  12. 24
      tuxbot/cogs/Mod/functions/utils.py
  13. 132
      tuxbot/cogs/Mod/mod.py
  14. 1
      tuxbot/cogs/Mod/models/__init__.py
  15. 46
      tuxbot/cogs/Mod/models/mutes.py
  16. 22
      tuxbot/cogs/Network/functions/utils.py
  17. 2
      tuxbot/cogs/Network/images/load_fail.py
  18. 4
      tuxbot/cogs/Network/network.py
  19. 8
      tuxbot/cogs/Polls/polls.py
  20. 6
      tuxbot/cogs/Tags/functions/utils.py
  21. 8
      tuxbot/core/bot.py
  22. 8
      tuxbot/core/i18n.py
  23. 4
      tuxbot/core/utils/functions/utils.py
  24. 18
      tuxbot/setup.py

3
.gitignore

@ -33,6 +33,9 @@ __pycache__/
__pypackages__/
venv
venv3.8
venv3.9
venv3.11
dist
build
*.egg

3
.idea/tuxbot_bot.iml

@ -7,6 +7,9 @@
<excludeFolder url="file://$MODULE_DIR$/venv" />
<excludeFolder url="file://$MODULE_DIR$/data" />
<excludeFolder url="file://$MODULE_DIR$/.mypy_cache" />
<excludeFolder url="file://$MODULE_DIR$/venv3.8" />
<excludeFolder url="file://$MODULE_DIR$/venv3.9" />
<excludeFolder url="file://$MODULE_DIR$/venv3.11" />
</content>
<orderEntry type="jdk" jdkName="Python 3.10 (tuxbot_bot)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />

2
Makefile

@ -35,7 +35,7 @@ update-all:
.PHONY: dev
dev: style update
tuxbot
$(VIRTUAL_ENV)/bin/tuxbot
# Docker
.PHONY: docker

4
README.rst

@ -14,7 +14,7 @@ Installing the pre-requirements
- The pre-requirements are:
- Python 3.10 or greater
- Python 3.8 or greater
- Pip
- Git
@ -134,7 +134,7 @@ To update the whole bot after a :bash:`git pull`, just execute
$ make update
.. |image0| image:: https://img.shields.io/badge/python-3.10-%23007ec6
.. |image0| image:: https://img.shields.io/badge/python-3.8%20%7C%203.9%20%7C%203.10-%23007ec6
.. |image1| image:: https://img.shields.io/github/issues/Rom1-J/tuxbot-bot
.. |image2| image:: https://img.shields.io/badge/code%20style-black-000000.svg
.. |image3| image:: https://wakatime.com/badge/github/Rom1-J/tuxbot-bot.svg

2
setup.cfg

@ -13,7 +13,7 @@ platforms = linux
[options]
packages = find_namespace:
python_requires = >=3.10
python_requires = >=3.9
install_requires =
aiocache>=0.11.1
asyncpg>=0.21.0

2
setup.py

@ -1,5 +1,5 @@
from setuptools import setup
setup(
python_requires=">=3.10",
python_requires=">=3.8",
)

3
tuxbot/cogs/Custom/custom.py

@ -1,4 +1,5 @@
import logging
from typing import List
import discord
from discord.ext import commands
@ -76,7 +77,7 @@ class Custom(commands.Cog):
@_custom.command(name="alias", aliases=["aliases"])
async def _custom_alias(self, ctx: ContextPlus, *, alias: AliasConvertor):
args: list[str] = str(alias).split(" | ")
args: List[str] = str(alias).split(" | ")
command = args[0]
custom = args[1]

2
tuxbot/cogs/Logs/functions/utils.py

@ -2,7 +2,7 @@ from collections import Counter
from typing import Dict
def sort_by(_events: Counter) -> dict[str, dict]:
def sort_by(_events: Counter) -> Dict[str, dict]:
majors = (
"guild",
"channel",

8
tuxbot/cogs/Logs/logs.py

@ -6,7 +6,7 @@ import textwrap
import traceback
from collections import defaultdict
from logging import LogRecord
from typing import Any, Dict
from typing import Any, Dict, List, DefaultDict
import discord
import humanize
@ -53,7 +53,7 @@ class Logs(commands.Cog):
self.bot = bot
self.process = psutil.Process()
self._batch_lock = asyncio.Lock()
self._data_batch: list[Dict[str, Any]] = []
self._data_batch: List[Dict[str, Any]] = []
self._gateway_queue: asyncio.Queue = asyncio.Queue()
self.gateway_worker.start() # pylint: disable=no-member
@ -62,8 +62,8 @@ class Logs(commands.Cog):
LogsConfig,
).config
self._resumes: list[datetime.datetime] = []
self._identifies: defaultdict[Any, list] = defaultdict(list)
self._resumes: List[datetime.datetime] = []
self._identifies: DefaultDict[Any, list] = defaultdict(list)
self.old_on_error = bot.on_error
bot.on_error = self.on_error

11
tuxbot/cogs/Mod/functions/converters.py

@ -6,6 +6,7 @@ from tuxbot.cogs.Mod.functions.exceptions import (
UnknownRuleException,
NonMessageException,
NonBotMessageException,
ReasonTooLongException,
)
from tuxbot.cogs.Mod.models import Rule
@ -52,3 +53,13 @@ class BotMessageConverter(commands.Converter):
raise NonMessageException(
_("Please provide a message in this guild")
)
class ReasonConverter(commands.Converter):
async def convert(self, ctx: Context, argument: str): # skipcq: PYL-W0613
if len(argument) > 300:
raise ReasonTooLongException(
_("Reason length must be 300 characters or lower.")
)
return argument

4
tuxbot/cogs/Mod/functions/exceptions.py

@ -19,3 +19,7 @@ class NonMessageException(ModException):
class NonBotMessageException(ModException):
pass
class ReasonTooLongException(ModException):
pass

24
tuxbot/cogs/Mod/functions/utils.py

@ -1,3 +1,6 @@
from typing import Optional, List
from tuxbot.cogs.Mod.models import MuteRole
from tuxbot.cogs.Mod.models.rules import Rule
from tuxbot.core.config import set_for_key
from tuxbot.core.config import Config
@ -10,15 +13,15 @@ async def save_lang(bot: Tux, ctx: ContextPlus, lang: str) -> None:
set_for_key(bot.config.Servers, ctx.guild.id, Config.Server, locale=lang)
async def get_server_rules(guild_id: int) -> list[Rule]:
async def get_server_rules(guild_id: int) -> List[Rule]:
return await Rule.filter(server_id=guild_id).all().order_by("rule_id")
def get_most_recent_server_rules(rules: list[Rule]) -> Rule:
def get_most_recent_server_rules(rules: List[Rule]) -> Rule:
return sorted(rules, key=lambda r: r.updated_at, reverse=True)[0]
def paginate_server_rules(rules: list[Rule]) -> list[str]:
def paginate_server_rules(rules: List[Rule]) -> List[str]:
body = [""]
for rule in rules:
@ -32,3 +35,18 @@ def paginate_server_rules(rules: list[Rule]) -> list[str]:
def format_rule(rule: Rule) -> str:
return f"**{rule.rule_id}** - {rule.content}"
async def get_mute_role(guild_id: int) -> Optional[MuteRole]:
return await MuteRole.get_or_none(server_id=guild_id)
async def create_mute_role(guild_id: int, role_id: int) -> MuteRole:
role_row = await MuteRole()
role_row.server_id = guild_id # type: ignore
role_row.role_id = role_id # type: ignore
await role_row.save()
return role_row

132
tuxbot/cogs/Mod/mod.py

@ -8,12 +8,14 @@ from tuxbot.cogs.Mod.functions.converters import (
RuleConverter,
RuleIDConverter,
BotMessageConverter,
ReasonConverter,
)
from tuxbot.cogs.Mod.functions.exceptions import (
RuleTooLongException,
UnknownRuleException,
NonMessageException,
NonBotMessageException,
ReasonTooLongException,
)
from tuxbot.cogs.Mod.functions.utils import (
save_lang,
@ -21,6 +23,8 @@ from tuxbot.cogs.Mod.functions.utils import (
format_rule,
get_most_recent_server_rules,
paginate_server_rules,
get_mute_role,
create_mute_role,
)
from tuxbot.cogs.Mod.models.rules import Rule
from tuxbot.core.utils import checks
@ -35,6 +39,7 @@ from tuxbot.core.i18n import (
from tuxbot.core.utils.functions.extra import (
group_extra,
ContextPlus,
command_extra,
)
log = logging.getLogger("tuxbot.cogs.Mod")
@ -53,9 +58,10 @@ class Mod(commands.Cog):
UnknownRuleException,
NonMessageException,
NonBotMessageException,
ReasonTooLongException,
),
):
await ctx.send(_(str(error), ctx, self.bot.config))
return await ctx.send(_(str(error), ctx, self.bot.config))
# =========================================================================
# =========================================================================
@ -163,7 +169,9 @@ class Mod(commands.Cog):
rule_row.server_id = ctx.guild.id
rule_row.author_id = ctx.message.author.id
rule_row.rule_id = len(await get_server_rules(ctx.guild.id)) + 1 # type: ignore
rule_row.rule_id = (
len(await get_server_rules(ctx.guild.id)) + 1 # type: ignore
)
rule_row.content = str(rule) # type: ignore
await rule_row.save()
@ -243,10 +251,13 @@ class Mod(commands.Cog):
pages = paginate_server_rules(rules)
# noinspection PyTypeChecker
to_edit: discord.Message = message
if len(pages) == 1:
embed.description = pages[0]
await message.edit(content="", embed=embed)
await to_edit.edit(content="", embed=embed)
else:
for i, page in enumerate(pages):
embed.title = _(
@ -254,4 +265,117 @@ class Mod(commands.Cog):
).format(ctx.guild.name, str(i + 1), str(len(pages)))
embed.description = page
await message.edit(content="", embed=embed)
await to_edit.edit(content="", embed=embed)
# =========================================================================
@group_extra(
name="mute",
deletable=True,
invoke_without_command=True,
)
@commands.guild_only()
@checks.is_admin()
async def _mute(
self,
ctx: ContextPlus,
members: commands.Greedy[discord.Member],
*,
reason: ReasonConverter,
):
if not members:
return await ctx.send(_("Missing members", ctx, self.bot.config))
role_row = await get_mute_role(ctx.guild.id)
if role_row is None:
return await ctx.send(
_(
"No mute role has been specified for this guild",
ctx,
self.bot.config,
)
)
for member in members:
await member.add_roles(
discord.Object(id=int(role_row.role_id)), reason=reason
)
await ctx.send("\N{THUMBS UP SIGN}")
@_mute.command(name="show", aliases=["role"])
async def _mute_show(
self,
ctx: ContextPlus,
):
role_row = await get_mute_role(ctx.guild.id)
if (
role_row is None
or (role := ctx.guild.get_role(int(role_row.role_id))) is None
):
return await ctx.send(
_(
"No mute role has been specified for this guild",
ctx,
self.bot.config,
)
)
muted_members = [m for m in ctx.guild.members if role in m.roles]
e = discord.Embed(
title=f"Role: {role.name} (ID: {role.id})", color=role.color
)
e.add_field(name="Total mute:", value=len(muted_members))
await ctx.send(embed=e)
@_mute.command(name="set", aliases=["define"])
async def _mute_set(self, ctx: ContextPlus, role: discord.Role):
role_row = await get_mute_role(ctx.guild.id)
if role_row is None:
await create_mute_role(ctx.guild.id, role.id)
else:
role_row.role_id = role.id # type: ignore
await role_row.save()
await ctx.send(
_("Mute role successfully defined", ctx, self.bot.config)
)
@command_extra(
name="unmute",
deletable=True,
)
@commands.guild_only()
@checks.is_admin()
async def _unmute(
self,
ctx: ContextPlus,
members: commands.Greedy[discord.Member],
*,
reason: ReasonConverter,
):
if not members:
return await ctx.send(_("Missing members", ctx, self.bot.config))
role_row = await get_mute_role(ctx.guild.id)
if role_row is None:
return await ctx.send(
_(
"No mute role has been specified for this guild",
ctx,
self.bot.config,
)
)
for member in members:
await member.remove_roles(
discord.Object(id=int(role_row.role_id)), reason=reason
)
await ctx.send("\N{THUMBS UP SIGN}")

1
tuxbot/cogs/Mod/models/__init__.py

@ -1,2 +1,3 @@
from .rules import *
from .warns import *
from .mutes import *

46
tuxbot/cogs/Mod/models/mutes.py

@ -0,0 +1,46 @@
import tortoise
from tortoise import fields
class MuteRole(tortoise.Model):
id = fields.BigIntField(pk=True)
server_id = fields.BigIntField()
role_id = fields.BigIntField()
class Meta:
table = "mute_role"
def __str__(self):
return (
f"<MuteRole id={self.id} "
f"server_id={self.server_id} "
f"role_id={self.role_id}>"
)
__repr__ = __str__
class Mute(tortoise.Model):
id = fields.BigIntField(pk=True)
server_id = fields.BigIntField()
author_id = fields.BigIntField()
reason = fields.TextField(max_length=300)
member_id = fields.BigIntField()
created_at = fields.DatetimeField(auto_now_add=True)
expire_at = fields.DatetimeField(null=True)
class Meta:
table = "mutes"
def __str__(self):
return (
f"<Mute id={self.id} "
f"server_id={self.server_id} "
f"author_id={self.author_id} "
f"reason='{self.reason}' "
f"member_id={self.member_id} "
f"created_at={self.created_at} "
f"expire_at={self.expire_at}>"
)
__repr__ = __str__

22
tuxbot/cogs/Network/functions/utils.py

@ -1,6 +1,6 @@
import io
import socket
from typing import NoReturn, Optional
from typing import NoReturn, Optional, Union
import asyncio
import re
@ -37,7 +37,7 @@ def _(x):
namespace="network",
)
async def get_ip(loop, ip: str, inet: Optional[dict]) -> str:
_inet: socket.AddressFamily | int = 0 # pylint: disable=no-member
_inet: Union[socket.AddressFamily, int] = 0 # pylint: disable=no-member
if inet:
if inet["inet"] == "6":
@ -89,8 +89,8 @@ async def get_hostname(loop, ip: str) -> str:
cache=Cache.MEMORY,
namespace="network",
)
async def get_ipwhois_result(loop, ip: str) -> NoReturn | dict:
def _get_ipwhois_result(_ip: str) -> NoReturn | dict:
async def get_ipwhois_result(loop, ip: str) -> Union[NoReturn, dict]:
def _get_ipwhois_result(_ip: str) -> Union[NoReturn, dict]:
try:
net = Net(ip)
obj = IPASN(net)
@ -121,7 +121,7 @@ async def get_ipwhois_result(loop, ip: str) -> NoReturn | dict:
namespace="network",
)
async def get_ipinfo_result(loop, apikey: str, ip: str) -> dict:
def _get_ipinfo_result(_ip: str) -> NoReturn | dict:
def _get_ipinfo_result(_ip: str) -> Union[NoReturn, dict]:
"""
Q. Why no getHandlerAsync ?
A. Use of this return "Unclosed client session" and "Unclosed connector"
@ -251,11 +251,11 @@ async def get_map_bytes(apikey: str, latlon: str) -> Optional[io.BytesIO]:
namespace="network",
)
async def get_pydig_result(
loop, domain: str, query_type: str, dnssec: str | bool
loop, domain: str, query_type: str, dnssec: Union[str, bool]
) -> list:
additional_args = [] if dnssec is False else ["+dnssec"]
def _get_pydig_result(_domain: str) -> NoReturn | dict:
def _get_pydig_result(_domain: str) -> Union[NoReturn, dict]:
resolver = pydig.Resolver(
nameservers=[
"80.67.169.40",
@ -275,14 +275,16 @@ async def get_pydig_result(
return []
def check_ip_version_or_raise(version: Optional[dict]) -> bool | NoReturn:
def check_ip_version_or_raise(
version: Optional[dict],
) -> Union[bool, NoReturn]:
if version is None or version["inet"] in ("4", "6", ""):
return True
raise InvalidIp(_("Invalid ip version"))
def check_query_type_or_raise(query_type: str) -> bool | NoReturn:
def check_query_type_or_raise(query_type: str) -> Union[bool, NoReturn]:
query_types = (
"a",
"aaaa",
@ -306,7 +308,7 @@ def check_query_type_or_raise(query_type: str) -> bool | NoReturn:
)
def check_asn_or_raise(asn: str) -> bool | NoReturn:
def check_asn_or_raise(asn: str) -> Union[bool, NoReturn]:
if asn.isdigit() and int(asn) < 4_294_967_295:
return True

2
tuxbot/cogs/Network/images/load_fail.py

File diff suppressed because one or more lines are too long

4
tuxbot/cogs/Network/network.py

@ -2,7 +2,7 @@ import asyncio
import logging
import time
from datetime import datetime
from typing import Optional
from typing import Optional, Union
import aiohttp
import discord
@ -272,7 +272,7 @@ class Network(commands.Cog):
ctx: ContextPlus,
domain: IPConverter,
query_type: QueryTypeConverter,
dnssec: str | bool = False,
dnssec: Union[str, bool] = False,
):
check_query_type_or_raise(str(query_type))

8
tuxbot/cogs/Polls/polls.py

@ -1,6 +1,6 @@
import json
import logging
from typing import Dict
from typing import Dict, Union, List
import discord
from discord.ext import commands
@ -45,7 +45,7 @@ class Polls(commands.Cog):
self,
ctx: ContextPlus,
question: str,
answers: list[str],
answers: List[str],
anonymous=False,
):
emotes = utils_emotes.get(len(answers))
@ -90,7 +90,7 @@ class Polls(commands.Cog):
async def get_poll(
self, pld: discord.RawReactionActionEvent
) -> bool | Poll:
) -> Union[bool, Poll]:
if pld.user_id != self.bot.user.id:
poll = await Poll.get_or_none(message_id=pld.message_id)
@ -225,7 +225,7 @@ class Polls(commands.Cog):
async def get_suggest(
self, pld: discord.RawReactionActionEvent
) -> bool | Suggest:
) -> Union[bool, Suggest]:
if pld.user_id != self.bot.user.id:
suggest = await Suggest.get_or_none(message_id=pld.message_id)

6
tuxbot/cogs/Tags/functions/utils.py

@ -1,4 +1,4 @@
from typing import Optional
from typing import Optional, List
import discord
@ -12,7 +12,7 @@ async def get_tag(guild_id: int, name: str) -> Tag:
async def get_all_tags(
guild_id: int, author: Optional[discord.Member] = None
) -> list[Tag]:
) -> List[Tag]:
if author is not None:
return (
await Tag.filter(server_id=guild_id, author_id=author.id)
@ -23,7 +23,7 @@ async def get_all_tags(
return await Tag.filter(server_id=guild_id).all().order_by("-uses")
async def search_tags(guild_id: int, q: str) -> list[Tag]:
async def search_tags(guild_id: int, q: str) -> List[Tag]:
return (
await Tag.filter(server_id=guild_id, name__icontains=q)
.all()

8
tuxbot/core/bot.py

@ -2,8 +2,9 @@ import asyncio
import datetime
import importlib
import logging
import sys
from collections import Counter
from typing import List, Tuple
from typing import List, Tuple, Union
import aiohttp
import discord
@ -177,6 +178,7 @@ class Tux(commands.AutoShardedBot):
table.add_row(f"Language: {self.config.Core.locale}")
table.add_row(f"Tuxbot Version: {__version__}")
table.add_row(f"Discord.py Version: {discord.__version__}")
table.add_row(f"Python Version: {sys.version.split(' ')[0]}")
table.add_row(f"Instance name: {self.instance_name}")
table.add_row(f"Shards: {self.shard_count}")
table.add_row(f"Servers: {len(self.guilds)}")
@ -200,13 +202,13 @@ class Tux(commands.AutoShardedBot):
self.console.print()
async def is_owner(
self, user: discord.User | discord.Member | discord.Object
self, user: Union[discord.User, discord.Member, discord.Object]
) -> bool:
"""Determines if the user is a bot owner.
Parameters
----------
user: discord.User | discord.Member
user: Union[discord.User, discord.Member]
Returns
-------

8
tuxbot/core/i18n.py

@ -1,7 +1,7 @@
import logging
import os
from pathlib import Path
from typing import Dict, NoReturn, Any, Tuple
from typing import Dict, NoReturn, Any, Tuple, Union
from babel.messages.pofile import read_po
@ -19,7 +19,7 @@ available_locales: Dict[str, Tuple] = {
}
def find_locale(locale: str) -> str | NoReturn:
def find_locale(locale: str) -> Union[str, NoReturn]:
"""We suppose `locale` is in `_available_locales.values()`"""
for key, val in available_locales.items():
@ -46,7 +46,9 @@ def get_locale_name(locale: str) -> str:
class Translator:
"""Class to load texts at init."""
def __init__(self, name: str, file_location: Path | os.PathLike | str):
def __init__(
self, name: str, file_location: Union[Path, os.PathLike, str]
):
"""Initializes the Translator object.
Parameters

4
tuxbot/core/utils/functions/utils.py

@ -1,6 +1,6 @@
import asyncio
import functools
from typing import Dict, Optional
from typing import Dict, Optional, Tuple
import aiohttp
from discord.ext import commands
@ -29,7 +29,7 @@ def typing(func):
async def shorten(
text: str, length: int, fail: bool = False
) -> tuple[bool, dict]:
) -> Tuple[bool, dict]:
output: Dict[str, str] = {"text": text[:length], "link": ""}
if len(text) > length:

18
tuxbot/setup.py

@ -7,7 +7,7 @@ import sys
import json
from argparse import Namespace
from pathlib import Path
from typing import List
from typing import List, Union
from urllib import request
from rich.prompt import Prompt, IntPrompt
@ -121,7 +121,7 @@ def get_ip() -> str:
def get_multiple(
question: str, confirmation: str, value_type: type
) -> List[str | int]:
) -> List[Union[str, int]]:
"""Give possibility to user to fill multiple value.
Parameters
@ -135,10 +135,12 @@ def get_multiple(
Returns
-------
List[str | int]
List[Union[str, int]]
List containing user filled values.
"""
prompt: IntPrompt | Prompt = IntPrompt() if value_type is int else Prompt()
prompt: Union[IntPrompt, Prompt] = (
IntPrompt() if value_type is int else Prompt()
)
user_input = prompt.ask(question, console=console)
@ -166,12 +168,14 @@ def get_multiple(
return values
def get_extra(question: str, value_type: type) -> str | int:
prompt: IntPrompt | Prompt = IntPrompt() if value_type is int else Prompt()
def get_extra(question: str, value_type: type) -> Union[str, int]:
prompt: Union[IntPrompt, Prompt] = (
IntPrompt() if value_type is int else Prompt()
)
return prompt.ask(question, console=console)
def additional_config(cogs: str | list = "**"):
def additional_config(cogs: Union[str, list] = "**"):
"""Asking for additional configs in cogs.
Returns

Loading…
Cancel
Save