fix(workflow): nvm

.forgejo/workflows/config/.pylintrc

@@ -18,4 +18,5 @@
     import-self,
     relative-beyond-top-level,
     too-many-instance-attributes,
-    duplicate-code
+    duplicate-code,
+    too-many-nested-blocks

.forgejo/workflows/workflow.yaml

@@ -58,7 +58,7 @@ jobs:
           npx -p "@getmeli/cli" meli upload ./site \
             --url "https://pages.coastalcommits.com" \
             --site "${{ vars.MELI_SITE_ID }}" \
-            --token "${{ secrets.MELI_SITE_SECRET }}" \
+            --token "${{ secrets.MELI_SECRET }}" \
             --release "$CI_ACTION_REF_NAME_SLUG/${{ env.GITHUB_SHA }}" \
             --branch "$CI_ACTION_REF_NAME_SLUG"
 

backup/backup.py

@@ -100,7 +100,7 @@ class Backup(commands.Cog):
         except (json.JSONDecodeError, IndexError):
             try:
                 export = json.loads(await ctx.message.reference.resolved.attachments[0].read())
-            except (json.JSONDecodeError, IndexError):
+            except (json.JSONDecodeError, IndexError, AttributeError):
                 await ctx.send(error("Please provide a valid JSON export file."))
                 return
 

emojiinfo/model.py

@@ -81,6 +81,7 @@ class PartialEmoji(discord.PartialEmoji):
         with open(path, "r", encoding="UTF-8") as file:
             emojis: dict = json.load(file)
         emoji_aliases = []
+        emoji_group = None
         for dict_name, group in emojis.items():
             for k, v in group.items():
                 if v == value:

poetry.lock

@@ -228,6 +228,27 @@ files = [
 [package.extras]
 dev = ["freezegun (>=1.0,<2.0)", "pytest (>=6.0)", "pytest-cov"]
 
+[[package]]
+name = "beautifulsoup4"
+version = "4.12.3"
+description = "Screen-scraping library"
+optional = false
+python-versions = ">=3.6.0"
+files = [
+    {file = "beautifulsoup4-4.12.3-py3-none-any.whl", hash = "sha256:b80878c9f40111313e55da8ba20bdba06d8fa3969fc68304167741bbf9e082ed"},
+    {file = "beautifulsoup4-4.12.3.tar.gz", hash = "sha256:74e3d1928edc070d21748185c46e3fb33490f22f52a3addee9aee0f4f7781051"},
+]
+
+[package.dependencies]
+soupsieve = ">1.2"
+
+[package.extras]
+cchardet = ["cchardet"]
+chardet = ["chardet"]
+charset-normalizer = ["charset-normalizer"]
+html5lib = ["html5lib"]
+lxml = ["lxml"]
+
 [[package]]
 name = "brotli"
 version = "1.1.0"
@@ -890,6 +911,21 @@ profiling = ["gprof2dot"]
 rtd = ["jupyter_sphinx", "mdit-py-plugins", "myst-parser", "pyyaml", "sphinx", "sphinx-copybutton", "sphinx-design", "sphinx_book_theme"]
 testing = ["coverage", "pytest", "pytest-cov", "pytest-regressions"]
 
+[[package]]
+name = "markdownify"
+version = "0.12.1"
+description = "Convert HTML to markdown."
+optional = false
+python-versions = "*"
+files = [
+    {file = "markdownify-0.12.1-py3-none-any.whl", hash = "sha256:a3805abd8166dbb7b27783c5599d91f54f10d79894b2621404d85b333c7ce561"},
+    {file = "markdownify-0.12.1.tar.gz", hash = "sha256:1fb08c618b30e0ee7a31a39b998f44a18fb28ab254f55f4af06b6d35a2179e27"},
+]
+
+[package.dependencies]
+beautifulsoup4 = ">=4.9,<5"
+six = ">=1.15,<2"
+
 [[package]]
 name = "markupsafe"
 version = "2.1.5"
@@ -2111,6 +2147,17 @@ files = [
     {file = "smmap-5.0.1.tar.gz", hash = "sha256:dceeb6c0028fdb6734471eb07c0cd2aae706ccaecab45965ee83f11c8d3b1f62"},
 ]
 
+[[package]]
+name = "soupsieve"
+version = "2.5"
+description = "A modern CSS selector implementation for Beautiful Soup."
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "soupsieve-2.5-py3-none-any.whl", hash = "sha256:eaa337ff55a1579b6549dc679565eac1e3d000563bcb1c8ab0d0fefbc0c2cdc7"},
+    {file = "soupsieve-2.5.tar.gz", hash = "sha256:5663d5a7b3bfaeee0bc4372e7fc48f9cff4940b3eec54a6451cc5299f1097690"},
+]
+
 [[package]]
 name = "tinycss2"
 version = "1.2.1"
@@ -2451,4 +2498,4 @@ multidict = ">=4.0"
 [metadata]
 lock-version = "2.0"
 python-versions = ">=3.11,<3.12"
-content-hash = "0ac382e0399d9c23c5f89a0ffeb3aae056dc8b28e864b22f815c0e3eb34175bd"
+content-hash = "229d7fd39618cf708f3cd5409dde2e6e25b822e4f936e14b3ade9800bf00daab"

pyproject.toml

@@ -15,6 +15,8 @@ websockets = "^12.0"
 pillow = "^10.3.0"
 numpy = "^1.26.4"
 colorthief = "^0.2.1"
+beautifulsoup4 = "^4.12.3"
+markdownify = "^0.12.1"
 
 [tool.poetry.group.dev]
 optional = true

seautils/info.json

@@ -8,5 +8,6 @@
   "hidden": true,
   "disabled": false,
   "min_bot_version": "3.5.0",
-  "min_python_version": [3, 8, 0]
+  "min_python_version": [3, 8, 0],
+  "requirements": ["beautifulsoup4", "markdownify"]
 }

seautils/seautils.py

@@ -5,13 +5,20 @@
 #  ____) |  __/ (_| \__ \\ V  V /| | | | | | | | | | | |  __/ |
 # |_____/ \___|\__,_|___/ \_/\_/ |_|_| |_| |_|_| |_| |_|\___|_|
 
+import asyncio
 import inspect
 import operator
+import re
+from asyncio.subprocess import Process
 from functools import partial, partialmethod
 from typing import Any
 
-from discord import Embed, app_commands
+import aiohttp
+import yaml
+from bs4 import BeautifulSoup
+from discord import Color, Embed, app_commands
 from discord.utils import CachedSlotProperty, cached_property
+from markdownify import MarkdownConverter
 from redbot.core import commands
 from redbot.core.bot import Red
 from redbot.core.dev_commands import cleanup_code
@@ -19,29 +26,38 @@ from redbot.core.utils import chat_formatting as cf
 from redbot.core.utils.views import SimpleMenu
 
 
+def md(soup: BeautifulSoup, **options) -> Any | str:
+    return MarkdownConverter(**options).convert_soup(soup=soup)
+
+def format_rfc_text(text: str, number: int) -> str:
+    one: str = re.sub(r"\(\.\/rfc(\d+)", r"(https://www.rfc-editor.org/rfc/rfc\1.html", text)
+    two: str = re.sub(r"\((#(?:section|page)-\d+(?:.\d+)?)\)", f"(https://www.rfc-editor.org/rfc/rfc{number}.html\1)", one)
+    three: str = re.sub(r"\n{3,}", "\n\n", two)
+    return three
+
 class SeaUtils(commands.Cog):
     """A collection of random utilities."""
 
     __author__ = ["SeaswimmerTheFsh"]
     __version__ = "1.0.0"
 
-    def __init__(self, bot: Red):
+    def __init__(self, bot: Red) -> None:
         self.bot = bot
 
     def format_help_for_context(self, ctx: commands.Context) -> str:
-        pre_processed = super().format_help_for_context(ctx) or ""
+        pre_processed = super().format_help_for_context(ctx=ctx) or ""
         n = "\n" if "\n\n" not in pre_processed else ""
         text = [
             f"{pre_processed}{n}",
             f"Cog Version: **{self.__version__}**",
-            f"Author: {cf.humanize_list(self.__author__)}"
+            f"Author: {cf.humanize_list(items=self.__author__)}"
         ]
         return "\n".join(text)
 
     def format_src(self, obj: Any) -> str:
         """A large portion of this code is repurposed from Zephyrkul's RTFS cog.
         https://github.com/Zephyrkul/FluffyCogs/blob/master/rtfs/rtfs.py"""
-        obj = inspect.unwrap(obj)
+        obj = inspect.unwrap(func=obj)
         src: Any = getattr(obj, "__func__", obj)
         if isinstance(obj, (commands.Command, app_commands.Command)):
             src = obj.callback
@@ -51,11 +67,11 @@ class SeaUtils(commands.Cog):
             src = obj.fget
         elif isinstance(obj, (cached_property, CachedSlotProperty)):
             src = obj.function
-        return inspect.getsource(src)
+        return inspect.getsource(object=src)
 
     @commands.command(aliases=["source", "src", "code", "showsource"])
     @commands.is_owner()
-    async def showcode(self, ctx: commands.Context, *, object: str): # pylint: disable=redefined-builtin
+    async def showcode(self, ctx: commands.Context, *, object: str) -> None: # pylint: disable=redefined-builtin
         """Show the code for a particular object."""
         try:
             if object.startswith("/") and (obj := ctx.bot.tree.get_command(object[1:])):
@@ -64,6 +80,8 @@ class SeaUtils(commands.Cog):
                 text = self.format_src(type(obj))
             elif obj := ctx.bot.get_command(object):
                 text = self.format_src(obj)
+            else:
+                raise AttributeError
             temp_content = cf.pagify(
                 text=cleanup_code(text),
                 escape_mass_mentions=True,
@@ -82,3 +100,151 @@ class SeaUtils(commands.Cog):
                 await ctx.send(embed=embed, reference=ctx.message.to_reference(fail_if_not_exists=False))
             else:
                 await ctx.send(content="Object not found!", reference=ctx.message.to_reference(fail_if_not_exists=False))
+
+    @commands.command(name='dig', aliases=['dnslookup', 'nslookup'])
+    @commands.is_owner()
+    async def dig(self, ctx: commands.Context, name: str, record_type: str | None = None, server: str | None = None, port: int = 53) -> None:
+        """Retrieve DNS information for a domain.
+
+        Uses `dig` to perform a DNS query. Will fall back to `nslookup` if `dig` is not installed on the system.
+        `nslookup` does not provide as much information as `dig`, so only the `name` parameter will be used if `nslookup` is used.
+        Will return the A, AAAA, and CNAME records for a domain by default. You can specify a different record type with the `type` parameter."""
+        command_opts: list[str | int] = ['dig']
+        query_types: list[str] = [record_type] if record_type else ['A', 'AAAA', 'CNAME']
+        if server:
+            command_opts.extend(['@', server])
+        for query_type in query_types:
+            command_opts.extend([name, query_type])
+        command_opts.extend(['-p', str(port), '+yaml'])
+
+        try:
+            process: Process = await asyncio.create_subprocess_exec(*command_opts, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+            stdout, stderr = await process.communicate()
+            if stderr:
+                await ctx.maybe_send_embed(message="An error was encountered!\n" + cf.box(text=stderr.decode()))
+            else:
+                data = yaml.safe_load(stdout.decode())
+                message_data: dict = data[0]['message']
+                response_data: dict = message_data['response_message_data']
+                if ctx.embed_requested():
+                    embed = Embed(
+                        title="DNS Query Result",
+                        color=await ctx.embed_color(),
+                        timestamp=message_data['response_time']
+                    )
+                    embed.add_field(name="Response Address", value=message_data['response_address'], inline=True)
+                    embed.add_field(name="Response Port", value=message_data['response_port'], inline=True)
+                    embed.add_field(name="Query Address", value=message_data['query_address'], inline=True)
+                    embed.add_field(name="Query Port", value=message_data['query_port'], inline=True)
+                    embed.add_field(name="Status", value=response_data['status'], inline=True)
+                    embed.add_field(name="Flags", value=response_data['flags'], inline=True)
+
+                    if response_data.get('status') != 'NOERROR':
+                        embed.colour = Color.red()
+                        embed.description = cf.error("Dig query did not return `NOERROR` status.")
+
+                    questions = []
+                    answers = []
+                    authorities = []
+                    for m in data:
+                        response = m['message']['response_message_data']
+                        if 'QUESTION_SECTION' in response:
+                            for question in response['QUESTION_SECTION']:
+                                if question not in questions:
+                                    questions.append(question)
+
+                        if 'ANSWER_SECTION' in response:
+                            for answer in response['ANSWER_SECTION']:
+                                if answer not in answers:
+                                    answers.append(answer)
+
+                        if 'AUTHORITY_SECTION' in response:
+                            for authority in response['AUTHORITY_SECTION']:
+                                if authority not in authorities:
+                                    authorities.append(authority)
+
+                    if questions:
+                        question_section = "\n".join(questions)
+                        embed.add_field(name="Question Section", value=f"{cf.box(text=question_section, lang='prolog')}", inline=False)
+
+                    if answers:
+                        answer_section = "\n".join(answers)
+                        if len(answer_section) > 1024:
+                            embed.description = cf.warning("Answer section is too long to fit within embed field, falling back to description.") + cf.box(answer_section)
+                        else:
+                            embed.add_field(name="Answer Section", value=f"{cf.box(text=answer_section, lang='prolog')}", inline=False)
+
+                    if authorities:
+                        authority_section = "\n".join(authorities)
+                        embed.add_field(name="Authority Section", value=f"{cf.box(text=authority_section, lang='prolog')}", inline=False)
+                    await ctx.send(embed=embed)
+                else:
+                    await ctx.send(content=cf.box(text=stdout, lang='yaml'))
+        except (FileNotFoundError):
+            try:
+                ns_process = await asyncio.create_subprocess_exec('nslookup', name, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+                ns_stdout, ns_stderr = await ns_process.communicate()
+                if ns_stderr:
+                    await ctx.maybe_send_embed(message="An error was encountered!\n" + cf.box(text=ns_stderr.decode()))
+                else:
+                    warning = cf.warning("`dig` is not installed! Defaulting to `nslookup`.\nThis command provides more information when `dig` is installed on the system.\n")
+                    if await ctx.embed_requested():
+                        embed = Embed(
+                            title="DNS Query Result",
+                            color=await ctx.embed_color(),
+                            timestamp=ctx.message.created_at
+                        )
+                        embed.description = warning + cf.box(text=ns_stdout.decode())
+                        await ctx.send(embed=embed)
+                    else:
+                        await ctx.send(content = warning + cf.box(text=ns_stdout.decode()))
+            except (FileNotFoundError):
+                await ctx.maybe_send_embed(message=cf.error("Neither `dig` nor `nslookup` are installed on the system. Unable to resolve DNS query."))
+
+    @commands.command()
+    async def rfc(self, ctx: commands.Context, number: int) -> None:
+        """Retrieve the text of an RFC document.
+
+        This command uses the [RFC Editor website](https://www.rfc-editor.org/) to fetch the text of an RFC document.
+        A [Request for Comments (RFC)](https://en.wikipedia.org/wiki/Request_for_Comments) is a publication in a series from the principal technical development and standards-setting bodies for the [Internet](https://en.wikipedia.org/wiki/Internet), most prominently the [Internet Engineering Task Force](https://en.wikipedia.org/wiki/Internet_Engineering_Task_Force). An RFC is authored by individuals or groups of engineers and [computer scientists](https://en.wikipedia.org/wiki/Computer_scientist) in the form of a [memorandum](https://en.wikipedia.org/wiki/Memorandum) describing methods, behaviors, research, or innovations applicable to the working of the Internet and Internet-connected systems. It is submitted either for [peer review](https://en.wikipedia.org/wiki/Peer_review) or to convey new concepts, information, or, occasionally, engineering humor.""" # noqa: E501
+        url = f"https://www.rfc-editor.org/rfc/rfc{number}.html"
+        datatracker_url = f"https://datatracker.ietf.org/doc/rfc{number}"
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url=url) as response:
+                if response.status == 200:
+                    html = await response.text()
+                    soup = BeautifulSoup(html, 'html.parser')
+                    pre_tags = soup.find_all('pre')
+                    content: list[Embed | str] = []
+                    for pre_tag in pre_tags:
+                        text = format_rfc_text(md(pre_tag), number)
+                        if len(text) > 4096:
+                            pagified_text = cf.pagify(text, delims=["\n\n"], page_length=4096)
+                            for page in pagified_text:
+                                if await ctx.embed_requested():
+                                    embed = Embed(
+                                        title=f"RFC Document {number}",
+                                        url=datatracker_url,
+                                        description=page,
+                                        color=await ctx.embed_color()
+                                    )
+                                    content.append(embed)
+                                else:
+                                    content.append(page)
+                        else:
+                            if await ctx.embed_requested():
+                                embed = Embed(
+                                    title=f"RFC Document {number}",
+                                    url=datatracker_url,
+                                    description=text,
+                                    color=await ctx.embed_color()
+                                )
+                                content.append(embed)
+                            else:
+                                content.append(text)
+                    if await ctx.embed_requested():
+                        for embed in content:
+                            embed.set_footer(text=f"Page {content.index(embed) + 1}/{len(content)}")
+                    await SimpleMenu(pages=content, disable_after_timeout=True, timeout=300).start(ctx)
+                else:
+                    await ctx.maybe_send_embed(content=cf.error(f"An error occurred while fetching RFC {number}. Status code: {response.status}."))

speedtest/__init__.py

@@ -1,5 +0,0 @@
-from .speedtest import Speedtest
-
-
-async def setup(bot):
-    await bot.add_cog(Speedtest(bot))

speedtest/info.json

@@ -1,14 +0,0 @@
-{
-  "author" : ["SeaswimmerTheFsh (seasw.)"],
-  "install_msg" : "Thank you for installing Speedtest!\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs).",
-  "name" : "Speedtest",
-  "short" : "A collection of useful utilities.",
-  "description" : "A collection of useful utilities.",
-  "end_user_data_statement" : "This cog does not store end user data.",
-  "hidden": true,
-  "disabled": false,
-  "min_bot_version": "3.5.0",
-  "min_python_version": [3, 10, 0],
-  "tags" : ["utility", "information"],
-  "requirements": ["pydantic"]
-}

speedtest/models.py

@@ -1,76 +0,0 @@
-from datetime import datetime
-
-from pydantic import BaseModel
-
-
-class Speedtest(BaseModel):
-    type: str
-    timestamp: datetime
-    ping: "Ping"
-    download: "Bandwidth"
-    upload: "Bandwidth"
-    isp: str
-    interface: "Interface"
-    server: "Server"
-    result: "Result"
-
-    @classmethod
-    def from_json(cls, data: dict) -> "Speedtest":
-        return cls(
-            type=data["type"],
-            timestamp=datetime.fromisoformat(data["timestamp"]),
-            ping=Ping(**data["ping"]),
-            download=Bandwidth(**data["download"]),
-            upload=Bandwidth(**data["upload"]),
-            isp=data["isp"],
-            interface=Interface(**data["interface"]),
-            server=Server(**data["server"]),
-            result=Result(**data["result"])
-        )
-
-class Bandwidth(BaseModel):
-    bandwidth: float
-    bytes: int
-    elapsed: int
-    latency: "Latency"
-
-    @property
-    def mbps(self) -> float:
-        return self.bandwidth / 1_000_000
-
-class Latency(BaseModel):
-    iqm: float
-    low: float
-    high: float
-    jitter: float
-
-class Interface(BaseModel):
-    internalIp: str
-    name: str
-    macAddr: str
-    isVpn: bool
-    externalIp: str
-
-class Ping(BaseModel):
-    jitter: float
-    latency: float
-    low: float
-    high: float
-
-class Server(BaseModel):
-    id: int
-    name: str
-    location: str
-    country: str
-    host: str
-    port: int
-    ip: str
-
-class Result(BaseModel):
-    id: str
-    url: str
-    persisted: bool
-
-    @property
-    def image(self) -> str:
-        return self.url + ".png"

speedtest/speedtest.py

@@ -1,71 +0,0 @@
-#   _____                         _
-#  / ____|                       (_)
-# | (___   ___  __ _ _____      ___ _ __ ___  _ __ ___   ___ _ __
-#  \___ \ / _ \/ _` / __\ \ /\ / / | '_ ` _ \| '_ ` _ \ / _ \ '__|
-#  ____) |  __/ (_| \__ \\ V  V /| | | | | | | | | | | |  __/ |
-# |_____/ \___|\__,_|___/ \_/\_/ |_|_| |_| |_|_| |_| |_|\___|_|
-
-import asyncio
-import json
-import subprocess
-
-import discord
-from redbot.core import commands
-from redbot.core.bot import Red
-from redbot.core.utils import chat_formatting as cf
-
-from .models import Speedtest as sp
-
-
-class Speedtest(commands.Cog):
-    """A collection of random utilities."""
-
-    __author__ = ["SeaswimmerTheFsh"]
-    __version__ = "1.0.0"
-
-    def __init__(self, bot: Red):
-        self.bot = bot
-
-    def format_help_for_context(self, ctx: commands.Context) -> str:
-        pre_processed = super().format_help_for_context(ctx) or ""
-        n = "\n" if "\n\n" not in pre_processed else ""
-        text = [
-            f"{pre_processed}{n}",
-            f"Cog Version: **{self.__version__}**",
-            f"Author: {cf.humanize_list(self.__author__)}"
-        ]
-        return "\n".join(text)
-
-    async def run_speedtest(self) -> str | sp:
-        try:
-            process = await asyncio.create_subprocess_exec(
-                "speedtest", "-f", "json", "--accept-license",
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE
-            )
-        except FileNotFoundError:
-            return "Speedtest CLI is not installed."
-        stdout, stderr = await process.communicate()
-        if process.returncode != 0:
-            return stderr.decode("utf-8")
-        return sp.from_json(json.loads(stdout.decode("utf-8")))
-
-    @commands.command()
-    @commands.is_owner()
-    async def speedtest(self, ctx: commands.Context) -> None:
-        """Run a speedtest."""
-        msg = await ctx.maybe_send_embed("Running speedtest...")
-        async with ctx.typing():
-            speedtest = await self.run_speedtest()
-            if await ctx.embed_requested():
-                if not isinstance(speedtest, sp):
-                    await msg.edit(embed=discord.Embed(description=f"An error occurred! {speedtest}", color=discord.Colour.red()))
-                    return
-                embed = discord.Embed(title="Speedtest Results", url=speedtest.result.url, color=await ctx.embed_color())
-                embed.set_image(url=speedtest.result.image)
-                await msg.edit(embed=embed)
-            else:
-                if not isinstance(speedtest, sp):
-                    await msg.edit(content=f"An error occurred! \n`{speedtest}`")
-                    return
-                await msg.edit(content=f"**[Result]({speedtest.result.url})**")