nex/college-bot-v1
1
0
Fork 0
mirror of https://github.com/nexy7574/LCC-bot.git synced 2024-09-16 16:43:40 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Make embeds pretty & reformat

Browse source
This commit is contained in:
Nexus 2023-11-04 17:18:39 +00:00
parent 4ca6b7f1d9
commit e26a91d437
Signed by: nex
GPG key ID: 0FA334385D0B689F
22 changed files with 432 additions and 639 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
cogs/assignments.py
View file

@ -3,18 +3,19 @@ import sqlite3
import textwrap
from typing import Optional
import config
import discord
from discord.ext import commands, tasks
import config
from utils import (
Assignments,
Tutors,
simple_embed_paginator,
get_or_none,
Student,
hyperlink,
console,
SelectAssigneesView,
Student,
Tutors,
console,
get_or_none,
hyperlink,
simple_embed_paginator,
)
BOOL_EMOJI = {True: "\N{white heavy check mark}", False: "\N{cross mark}"}

175
cogs/events.py
View file

@ -1,3 +1,5 @@
import asyncio
import glob
import hashlib
import inspect
import io
@ -5,21 +7,22 @@ import json
import os
import random
import re
import asyncio
import textwrap
import subprocess
import textwrap
import traceback
import glob
import warnings
from datetime import datetime, timezone, timedelta
from bs4 import BeautifulSoup
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional, Tuple, Dict, Any
from typing import Any, Dict, Optional, Tuple
import discord
import httpx
from discord.ext import commands, pages, tasks
from utils import Student, get_or_none, console
from bs4 import BeautifulSoup
from config import guilds
from discord.ext import commands, pages, tasks
from utils import Student, console, get_or_none
try:
from config import dev
except ImportError:
@ -29,8 +32,7 @@ try:
except ImportError:
OAUTH_REDIRECT_URI = None
try:
from config import GITHUB_USERNAME
from config import GITHUB_PASSWORD
from config import GITHUB_PASSWORD, GITHUB_USERNAME
except ImportError:
GITHUB_USERNAME = None
GITHUB_PASSWORD = None
@ -143,16 +145,12 @@ class Events(commands.Cog):
_re = re.match(
r"https://github\.com/(?P<repo>[a-zA-Z0-9-]+/[\w.-]+)/blob/(?P<path>[^#>]+)(\?[^#>]+)?"
r"(#L(?P<start_line>\d+)(([-~:]|(\.\.))L(?P<end_line>\d+))?)",
message.content
message.content,
)
if _re:
branch, path = _re.group("path").split("/", 1)
_p = Path(path).suffix
url = RAW_URL.format(
repo=_re.group("repo"),
branch=branch,
path=path
)
url = RAW_URL.format(repo=_re.group("repo"), branch=branch, path=path)
if all((GITHUB_PASSWORD, GITHUB_USERNAME)):
auth = (GITHUB_USERNAME, GITHUB_PASSWORD)
else:
@ -180,7 +178,7 @@ class Events(commands.Cog):
RAW_URL = "https://github.com/{repo}/archive/refs/heads/{branch}.zip"
_full_re = re.finditer(
r"https://github\.com/(?P<repo>[a-zA-Z0-9-]+/[\w.-]+)(/tree/(?P<branch>[^#>]+))?\.(git|zip)",
message.content
message.content,
)
for _match in _full_re:
repo = _match.group("repo")
@ -206,11 +204,7 @@ class Events(commands.Cog):
await message.edit(suppress=True)
@commands.Cog.listener()
async def on_voice_state_update(
self,
member: discord.Member,
*_
):
async def on_voice_state_update(self, member: discord.Member, *_):
me_voice = member.guild.me.voice
if me_voice is None or me_voice.channel is None or member.guild.voice_client is None:
return
@ -240,57 +234,49 @@ class Events(commands.Cog):
except asyncio.TimeoutError:
await message.channel.trigger_typing()
await message.reply(
"I'd play the song but discord's voice servers are shit.",
file=discord.File(_file)
"I'd play the song but discord's voice servers are shit.", file=discord.File(_file)
)
region = message.author.voice.channel.rtc_region
# noinspection PyUnresolvedReferences
console.log(
"Timed out connecting to voice channel: {0.name} in {0.guild.name} "
"(region {1})".format(
message.author.voice.channel,
region.name if region else "auto (unknown)"
message.author.voice.channel, region.name if region else "auto (unknown)"
)
)
return
if voice.channel != message.author.voice.channel:
await voice.move_to(message.author.voice.channel)
if message.guild.me.voice.self_mute or message.guild.me.voice.mute:
await message.channel.trigger_typing()
await message.reply("Unmute me >:(", file=discord.File(_file))
else:
def after(err):
self.bot.loop.create_task(
_dc(voice),
)
if err is not None:
console.log(f"Error playing audio: {err}")
self.bot.loop.create_task(
message.add_reaction("\N{speaker with cancellation stroke}")
)
self.bot.loop.create_task(message.add_reaction("\N{speaker with cancellation stroke}"))
else:
self.bot.loop.create_task(
message.remove_reaction("\N{speaker with three sound waves}", self.bot.user)
)
self.bot.loop.create_task(
message.add_reaction("\N{speaker}")
)
self.bot.loop.create_task(message.add_reaction("\N{speaker}"))
# noinspection PyTypeChecker
src = discord.FFmpegPCMAudio(str(_file.resolve()), stderr=subprocess.DEVNULL)
src = discord.PCMVolumeTransformer(src, volume=0.5)
voice.play(
src,
after=after
)
voice.play(src, after=after)
if message.channel.permissions_for(message.guild.me).add_reactions:
await message.add_reaction("\N{speaker with three sound waves}")
else:
await message.channel.trigger_typing()
await message.reply(file=discord.File(_file))
return internal
async def send_what():
@ -305,29 +291,18 @@ class Events(commands.Cog):
await message.reply("You really are deaf, aren't you.")
elif not msg.content:
await message.reply(
"Maybe *I* need to get my hearing checked, I have no idea what {} said.".format(
msg.author.mention
)
"Maybe *I* need to get my hearing checked, I have no idea what {} said.".format(msg.author.mention)
)
else:
text = "{0.author.mention} said '{0.content}', you deaf sod.".format(
msg
)
_content = textwrap.shorten(
text, width=2000, placeholder="[...]"
)
text = "{0.author.mention} said '{0.content}', you deaf sod.".format(msg)
_content = textwrap.shorten(text, width=2000, placeholder="[...]")
await message.reply(_content, allowed_mentions=discord.AllowedMentions.none())
def get_sloc_count():
root = Path.cwd()
root_files = list(root.glob("**/*.py"))
root_files.append(root / "main.py")
root_files = list(
filter(
lambda f: "venv" not in f.parents and "venv" not in f.parts,
root_files
)
)
root_files = list(filter(lambda f: "venv" not in f.parents and "venv" not in f.parts, root_files))
lines = 0
for file in root_files:
@ -361,10 +336,10 @@ class Events(commands.Cog):
"content_type": a.content_type,
}
for a in message.attachments
]
],
}
if message.author.discriminator != "0":
payload["author"] += '#%s' % message.author.discriminator
payload["author"] += "#%s" % message.author.discriminator
if message.author != self.bot.user and (payload["content"] or payload["attachments"]):
await self.bot.bridge_queue.put(payload)
@ -388,78 +363,48 @@ class Events(commands.Cog):
},
r"it just works": {
"func": play_voice(assets / "it-just-works.ogg"),
"meta": {
"check": (assets / "it-just-works.ogg").exists
}
"meta": {"check": (assets / "it-just-works.ogg").exists},
},
"count to (3|three)": {
"func": play_voice(assets / "count-to-three.ogg"),
"meta": {
"check": (assets / "count-to-three.ogg").exists
}
"meta": {"check": (assets / "count-to-three.ogg").exists},
},
r"^linux$": {
"content": lambda: (assets / "copypasta.txt").read_text(),
"meta": {
"needs_mention": True,
"check": (assets / "copypasta.txt").exists
}
"meta": {"needs_mention": True, "check": (assets / "copypasta.txt").exists},
},
r"carat": {
"file": discord.File(assets / "carat.jpg"),
"delete_after": None,
"meta": {
"check": (assets / "carat.jpg").exists
}
"meta": {"check": (assets / "carat.jpg").exists},
},
r"(lupupa|fuck(ed)? the hell out\W*)": {
"file": discord.File(assets / "lupupa.jpg"),
"meta": {
"check": (assets / "lupupa.jpg").exists
}
"meta": {"check": (assets / "lupupa.jpg").exists},
},
r"[s5]+(m)+[e3]+[g9]+": {
# "func": send_smeg,
"file": lambda: discord.File(random.choice(list((assets / "smeg").iterdir()))),
"delete_after": 30,
"meta": {
"sub": {
r"pattern": r"([-_.\s\u200b])+",
r"with": ''
},
"check": (assets / "smeg").exists
}
},
r"(what|huh)(\?|!)*$": {
"func": send_what,
"meta": {
"check": lambda: message.reference is not None
}
"meta": {"sub": {r"pattern": r"([-_.\s\u200b])+", r"with": ""}, "check": (assets / "smeg").exists},
},
r"(what|huh)(\?|!)*$": {"func": send_what, "meta": {"check": lambda: message.reference is not None}},
("year", "linux", "desktop"): {
"content": lambda: "%s will be the year of the GNU+Linux desktop." % datetime.now().year,
"delete_after": None
"delete_after": None,
},
r"mine(ing|d)? (diamonds|away)": {
"func": play_voice(assets / "mine-diamonds.ogg"),
"meta": {
"check": (assets / "mine-diamonds.ogg").exists
}
"meta": {"check": (assets / "mine-diamonds.ogg").exists},
},
r"v[ei]r[mg]in(\sme(d|m[a]?)ia\W*)?(\W\w*\W*)?$": {
"content": "Get virgin'd",
"file": lambda: discord.File(
random.choice(list(Path(assets / 'virgin').iterdir()))
),
"meta": {
"check": (assets / 'virgin').exists
}
"file": lambda: discord.File(random.choice(list(Path(assets / "virgin").iterdir()))),
"meta": {"check": (assets / "virgin").exists},
},
r"richard|(dick\W*$)": {
"file": discord.File(assets / "visio.png"),
"meta": {
"check": (assets / "visio.png").exists
}
"meta": {"check": (assets / "visio.png").exists},
},
r"thank(\syou|s)(,)? jimmy": {
"content": "You're welcome, %s!" % message.author.mention,
@ -467,18 +412,12 @@ class Events(commands.Cog):
r"(ok )?jimmy (we|i) g[eo]t it": {
"content": "No need to be so rude! Cunt.",
},
r"c(mon|ome on) jimmy": {
"content": "IM TRYING"
},
r"(bor(r)?is|johnson)": {
"file": discord.File(assets / "boris.jpeg")
},
r"c(mon|ome on) jimmy": {"content": "IM TRYING"},
r"(bor(r)?is|johnson)": {"file": discord.File(assets / "boris.jpeg")},
r"\W?(s(ource\w)?)?l(ines\s)?o(f\s)?c(ode)?(\W)?$": {
"content": lambda: "I have {:,} lines of source code across {:,} files!".format(*get_sloc_count())
},
r"t(ry\s)?i(t\s)?a(nd\s)?see.*": {
"content": "https://tryitands.ee"
}
r"t(ry\s)?i(t\s)?a(nd\s)?see.*": {"content": "https://tryitands.ee"},
}
# Stop responding to any bots
if message.author.bot is True:
@ -515,11 +454,7 @@ class Events(commands.Cog):
continue
if meta.get("sub") is not None and isinstance(meta["sub"], dict):
content = re.sub(
meta["sub"]["pattern"],
meta["sub"]["with"],
message.content
)
content = re.sub(meta["sub"]["pattern"], meta["sub"]["with"], message.content)
else:
content = message.content
@ -566,7 +501,7 @@ class Events(commands.Cog):
r"mpreg|lupupa|\U0001fac3": "\U0001fac3", # mpreg
r"(trans(gender)?($|\W+)|%s)" % T_EMOJI: T_EMOJI, # trans
r"gay|%s" % G_EMOJI: G_EMOJI,
r"(femboy|trans(gender)?($|\W+))": C_EMOJI
r"(femboy|trans(gender)?($|\W+))": C_EMOJI,
}
if message.channel.permissions_for(message.guild.me).add_reactions:
is_naus = random.randint(1, 100) == 32
@ -588,9 +523,7 @@ class Events(commands.Cog):
if channel is None or not channel.can_send(discord.Embed()):
warnings.warn("Cannot send to spam channel, disabling feed fetcher")
return
headers = {
"User-Agent": f"python-httpx/{httpx.__version__} (Like Akregator/5.22.3); syndication"
}
headers = {"User-Agent": f"python-httpx/{httpx.__version__} (Like Akregator/5.22.3); syndication"}
file = Path.home() / ".cache" / "lcc-bot" / "discord.atom"
if not file.exists():
@ -675,22 +608,18 @@ class Events(commands.Cog):
content = f"[open on discordstatus.com (too large to display)]({entry.link['href']})"
embed = discord.Embed(
title=title,
description=content,
color=colour,
url=entry.link["href"],
timestamp=updated
title=title, description=content, color=colour, url=entry.link["href"], timestamp=updated
)
embed.set_author(
name="Discord Status",
url="https://discordstatus.com/",
icon_url="https://raw.githubusercontent.com/EEKIM10/LCC-bot/"
"fe0cb6dd932f9fc2cb0a26433aff8e4cce19279a/assets/discord.png"
"fe0cb6dd932f9fc2cb0a26433aff8e4cce19279a/assets/discord.png",
)
embed.set_footer(
text="Published: {} | Updated: {}".format(
datetime.fromisoformat(entry.find("published").text).strftime("%Y-%m-%d %H:%M:%S"),
updated.strftime("%Y-%m-%d %H:%M:%S")
updated.strftime("%Y-%m-%d %H:%M:%S"),
)
)

16
cogs/extremism.py
View file

@ -1,20 +1,18 @@
# You have been warned - this file is very EXTREME!
import discord
import asyncio
import io
import numpy
import blend_modes
from functools import partial
from discord.ext import commands
import blend_modes
import discord
import numpy
import PIL.Image
from discord.ext import commands
from PIL import Image
def _overlay_images(
background: PIL.Image.Image,
foreground: PIL.Image.Image,
mode=blend_modes.overlay,
opacity: float = 1.0
background: PIL.Image.Image, foreground: PIL.Image.Image, mode=blend_modes.overlay, opacity: float = 1.0
) -> PIL.Image.Image:
background = background.convert("RGBA")
foreground = foreground.convert("RGBA")
@ -70,7 +68,7 @@ def extremify(img: PIL.Image.Image) -> PIL.Image.Image:
class Extremism(commands.Cog):
def __init__(self, bot):
self.bot = bot
@commands.slash_command(name="radicalise")
async def radicalise(self, ctx, image: discord.Attachment = None, user: discord.User = None):
"""Makes an image extremely radical."""

32
cogs/info.py
View file

@ -1,8 +1,10 @@
import discord
import httpx
from discord.ext import commands
from utils import get_or_none
from utils.db import AccessTokens
try:
from config import OAUTH_ID, OAUTH_REDIRECT_URI
except ImportError:
@ -13,7 +15,7 @@ class InfoCog(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.client = httpx.AsyncClient(base_url="https://discord.com/api")
async def get_user_info(self, token: str):
try:
response = await self.client.get("/users/@me", headers={"Authorization": f"Bearer {token}"})
@ -21,7 +23,7 @@ class InfoCog(commands.Cog):
except (httpx.HTTPError, httpx.RequestError, ConnectionError):
return
return response.json()
async def get_user_guilds(self, token: str):
try:
response = await self.client.get("/users/@me/guilds", headers={"Authorization": f"Bearer {token}"})
@ -37,13 +39,13 @@ class InfoCog(commands.Cog):
except (httpx.HTTPError, httpx.RequestError, ConnectionError):
return
return response.json()
@commands.slash_command()
@commands.cooldown(1, 10, commands.BucketType.user)
async def me(self, ctx: discord.ApplicationCommand):
"""Displays oauth info about you"""
await ctx.defer()
user = await get_or_none(AccessTokens, user_id=ctx.author.id)
if not user:
url = OAUTH_REDIRECT_URI
@ -51,7 +53,7 @@ class InfoCog(commands.Cog):
embed=discord.Embed(
title="You must link your account first!",
description="Don't worry, [I only store your IP information. And the access token.](%s)" % url,
url=url
url=url,
)
)
user_data = await self.get_user_info(user.access_token)
@ -61,8 +63,20 @@ class InfoCog(commands.Cog):
title="Your info",
)
if user_data:
for field in ("bot", "system", "mfa_enabled", "banner", "accent_color", "mfa_enabled", "locale",
"verified", "email", "flags", "premium_type", "public_flags"):
for field in (
"bot",
"system",
"mfa_enabled",
"banner",
"accent_color",
"mfa_enabled",
"locale",
"verified",
"email",
"flags",
"premium_type",
"public_flags",
):
user_data.setdefault(field, None)
lines = [
"ID: {0[id]}",
@ -88,14 +102,14 @@ class InfoCog(commands.Cog):
name="User Info",
value="\n".join(lines).format(user_data, email),
)
if guilds:
guilds = sorted(guilds, key=lambda x: x["name"])
embed.add_field(
name="Guilds (%d):" % len(guilds),
value="\n".join(f"{guild['name']} ({guild['id']})" for guild in guilds),
)
if connections:
embed.add_field(
name="Connections (%d):" % len(connections),

27
cogs/mod.py
View file

@ -1,13 +1,15 @@
from datetime import datetime
from typing import Sized
import discord
from discord.ext import commands
from utils import Student, get_or_none, BannedStudentID, owner_or_admin, JimmyBans
from typing import Sized
from utils import BannedStudentID, JimmyBans, Student, get_or_none, owner_or_admin
class LimitedList(list):
"""FIFO Limited list"""
def __init__(self, iterable: Sized = None, size: int = 5000):
if iterable:
assert len(iterable) <= size, "Initial iterable too big."
@ -131,7 +133,7 @@ class Mod(commands.Cog):
message: discord.Message
if message.author == user:
query_ = query.lower() if query else None
content_ = str(message.clean_content or '').lower()
content_ = str(message.clean_content or "").lower()
if query_ is not None and (query_ in content_ or content_ in query_):
break
else:
@ -144,22 +146,15 @@ class Mod(commands.Cog):
colour=message.author.colour,
timestamp=message.created_at,
fields=[
discord.EmbedField(
"Attachment count",
str(len(message.attachments)),
False
),
discord.EmbedField(
"Location",
str(message.channel.mention)
),
discord.EmbedField("Attachment count", str(len(message.attachments)), False),
discord.EmbedField("Location", str(message.channel.mention)),
discord.EmbedField(
"Times",
f"Created: {discord.utils.format_dt(message.created_at, 'R')} | Edited: "
f"{'None' if message.edited_at is None else discord.utils.format_dt(message.edited_at, 'R')}"
)
]
).set_author(name=message.author.display_name, icon_url=message.author.display_avatar.url)
f"{'None' if message.edited_at is None else discord.utils.format_dt(message.edited_at, 'R')}",
),
],
).set_author(name=message.author.display_name, icon_url=message.author.display_avatar.url),
]
await ctx.reply(embeds=embeds)

359
cogs/other.py
View file

@ -1,14 +1,13 @@
import asyncio
import functools
import glob
import hashlib
import io
import json
import math
import os
import shutil
import random
import re
import shutil
import subprocess
import sys
import tempfile
@ -16,24 +15,22 @@ import textwrap
import traceback
from functools import partial
from io import BytesIO
import dns.resolver
import httpx
from dns import asyncresolver
import aiofiles
import pyttsx3
from time import time, time_ns, sleep
from typing import Literal
from typing import Tuple, Optional, Dict
from pathlib import Path
from time import sleep, time, time_ns
from typing import Dict, Literal, Optional, Tuple
from urllib.parse import urlparse
from PIL import Image
import pytesseract
import aiofiles
import aiohttp
import discord
import dns.resolver
import httpx
import psutil
import pytesseract
import pyttsx3
from discord.ext import commands, pages
from dns import asyncresolver
from PIL import Image
from rich.tree import Tree
from selenium import webdriver
from selenium.common.exceptions import WebDriverException
@ -41,7 +38,8 @@ from selenium.webdriver.chrome.options import Options as ChromeOptions
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.firefox.service import Service as FirefoxService
from utils import console, Timer
from utils import Timer, console
try:
from config import proxy
@ -129,22 +127,21 @@ class OtherCog(commands.Cog):
with tempfile.TemporaryDirectory(prefix="jimmy-ytdl", suffix="-info") as tempdir:
with yt_dlp.YoutubeDL(
{
"windowsfilenames": True,
"restrictfilenames": True,
"noplaylist": True,
"nocheckcertificate": True,
"no_color": True,
"noprogress": True,
"logger": NullLogger(),
"paths": {"home": tempdir, "temp": tempdir},
"cookiefile": Path(__file__).parent.parent / "jimmy-cookies.txt"
}
{
"windowsfilenames": True,
"restrictfilenames": True,
"noplaylist": True,
"nocheckcertificate": True,
"no_color": True,
"noprogress": True,
"logger": NullLogger(),
"paths": {"home": tempdir, "temp": tempdir},
"cookiefile": Path(__file__).parent.parent / "jimmy-cookies.txt",
}
) as downloader:
try:
info = await self.bot.loop.run_in_executor(
None,
partial(downloader.extract_info, url, download=False)
None, partial(downloader.extract_info, url, download=False)
)
except yt_dlp.utils.DownloadError:
return {}
@ -157,8 +154,8 @@ class OtherCog(commands.Cog):
"acodec": fmt.get("acodec", "?"),
"vcodec": fmt.get("vcodec", "?"),
"resolution": fmt.get("resolution", "?x?"),
"filesize": fmt.get("filesize", float('inf')),
"format": fmt.get("format", '?'),
"filesize": fmt.get("filesize", float("inf")),
"format": fmt.get("format", "?"),
}
for fmt in info["formats"]
}
@ -820,10 +817,11 @@ class OtherCog(commands.Cog):
f"* URL: <{friendly_url}>\n"
f"* Load time: {fetch_time:.2f}ms\n"
f"* Screenshot render time: {screenshot_time:.2f}ms\n"
f"* Total time: {(fetch_time + screenshot_time):.2f}ms\n" +
(
'* Probability of being scat or something else horrifying: 100%'
if ctx.user.id == 1019233057519177778 else ''
f"* Total time: {(fetch_time + screenshot_time):.2f}ms\n"
+ (
"* Probability of being scat or something else horrifying: 100%"
if ctx.user.id == 1019233057519177778
else ""
),
file=screenshot,
)
@ -880,22 +878,15 @@ class OtherCog(commands.Cog):
@commands.slash_command(name="yt-dl")
@commands.max_concurrency(1, commands.BucketType.user)
async def yt_dl_2(
self,
ctx: discord.ApplicationContext,
url: discord.Option(
description="The URL to download.",
type=str
),
_format: discord.Option(
name="format",
description="The format to download.",
type=str,
autocomplete=format_autocomplete,
default=""
) = "",
extract_audio: bool = False,
cookies_txt: discord.Attachment = None,
disable_filesize_buffer: bool = False
self,
ctx: discord.ApplicationContext,
url: discord.Option(description="The URL to download.", type=str),
_format: discord.Option(
name="format", description="The format to download.", type=str, autocomplete=format_autocomplete, default=""
) = "",
extract_audio: bool = False,
cookies_txt: discord.Attachment = None,
disable_filesize_buffer: bool = False,
):
"""Downloads a video using youtube-dl"""
cookies = io.StringIO()
@ -903,6 +894,7 @@ class OtherCog(commands.Cog):
await ctx.defer()
from urllib.parse import parse_qs
formats = await self.list_formats(url)
if _format:
_fmt = _format
@ -925,7 +917,7 @@ class OtherCog(commands.Cog):
stdout = tempdir / "stdout.txt"
stderr = tempdir / "stderr.txt"
default_cookies_txt = (Path.cwd() / "jimmy-cookies.txt")
default_cookies_txt = Path.cwd() / "jimmy-cookies.txt"
real_cookies_txt = tempdir / "cookies.txt"
if cookies_txt is not None:
await cookies_txt.save(fp=real_cookies_txt)
@ -982,17 +974,17 @@ class OtherCog(commands.Cog):
"trim_file_name": 128,
"extract_audio": extract_audio,
"format_sort": [
"vcodec:h264",
"acodec:aac",
"vcodec:vp9",
"acodec:opus",
"vcodec:h264",
"acodec:aac",
"vcodec:vp9",
"acodec:opus",
"acodec:vorbis",
"vcodec:vp8",
"ext"
"vcodec:vp8",
"ext",
],
"merge_output_format": "webm/mp4/mov/flv/avi/ogg/m4a/wav/mp3/opus/mka/mkv",
"source_address": "0.0.0.0",
"cookiefile": str(real_cookies_txt.resolve().absolute())
"cookiefile": str(real_cookies_txt.resolve().absolute()),
}
description = ""
proxy_url = "socks5://localhost:1090"
@ -1000,23 +992,26 @@ class OtherCog(commands.Cog):
proxy_down = await self.check_proxy("socks5://localhost:1090")
if proxy_down > 0:
if proxy_down == 1:
description += ":warning: (SHRoNK) Proxy check leaked IP - trying backup proxy\n"
description += ":warning: (SHRoNK) Proxy check leaked IP - trying backup proxy.\n"
elif proxy_down == 2:
description += ":warning: (SHRoNK) Proxy connection failed - trying backup proxy\n"
description += ":warning: (SHRoNK) Proxy connection failed - trying backup proxy.\n"
else:
description += ":warning: (SHRoNK) Unknown proxy error - trying backup proxy\n"
description += ":warning: (SHRoNK) Unknown proxy error - trying backup proxy.\n"
proxy_down = await self.check_proxy("socks5://localhost:1080")
if proxy_down > 0:
if proxy_down == 1:
description += ":warning: (NexBox) Proxy check leaked IP.\n"
description += ":warning: (NexBox) Proxy check leaked IP..\n"
elif proxy_down == 2:
description += ":warning: (NexBox) Proxy connection failed\n"
description += ":warning: (NexBox) Proxy connection failed.\n"
else:
description += ":warning: (NexBox) Unknown proxy error\n"
description += ":warning: (NexBox) Unknown proxy error.\n"
proxy_url = None
else:
proxy_url = "socks5://localhost:1080"
description += "\N{white heavy check mark} Using fallback NexBox proxy."
else:
description += "\N{white heavy check mark} Using the SHRoNK proxy."
except Exception as e:
traceback.print_exc()
description += f":warning: Failed to check proxy (`{e}`). Going unproxied."
@ -1024,11 +1019,7 @@ class OtherCog(commands.Cog):
args["proxy"] = proxy_url
if extract_audio:
args["postprocessors"] = [
{
"key": "FFmpegExtractAudio",
"preferredquality": "48",
"preferredcodec": "opus"
}
{"key": "FFmpegExtractAudio", "preferredquality": "24", "preferredcodec": "opus"}
]
args["format"] = args["format"] or f"(ba/b)[filesize<={MAX_SIZE_MB}M]/ba/b"
@ -1037,11 +1028,22 @@ class OtherCog(commands.Cog):
with yt_dlp.YoutubeDL(args) as downloader:
try:
await ctx.respond(
embed=discord.Embed(
title="Downloading...", description=description, colour=discord.Colour.blurple()
)
extracted_info = downloader.extract_info(url, download=False)
except yt_dlp.utils.DownloadError:
pass
else:
title = extracted_info.get("title", url)
thumbnail_url = extracted_info.get("thumbnail") or discord.Embed.Empty
webpage_url = extracted_info.get("webpage_url")
try:
embed = discord.Embed(
title="Downloading %r..." % title,
description=description,
colour=discord.Colour.blurple(),
url=webpage_url,
)
embed.set_thumbnail(url=thumbnail_url)
await ctx.respond(embed=embed)
await self.bot.loop.run_in_executor(None, partial(downloader.download, [url]))
except yt_dlp.utils.DownloadError as e:
traceback.print_exc()
@ -1049,27 +1051,26 @@ class OtherCog(commands.Cog):
embed=discord.Embed(
title="Error",
description=f"Download failed:\n```\n{e}\n```",
colour=discord.Colour.red()
colour=discord.Colour.red(),
url=webpage_url,
),
delete_after=60
delete_after=60,
)
else:
parsed_qs = parse_qs(url)
if 't' in parsed_qs and parsed_qs['t'] and parsed_qs['t'][0].isdigit():
if "t" in parsed_qs and parsed_qs["t"] and parsed_qs["t"][0].isdigit():
# Assume is timestamp
timestamp = round(float(parsed_qs['t'][0]))
timestamp = round(float(parsed_qs["t"][0]))
end_timestamp = None
if len(parsed_qs["t"]) >= 2:
end_timestamp = round(float(parsed_qs['t'][1]))
end_timestamp = round(float(parsed_qs["t"][1]))
if end_timestamp < timestamp:
end_timestamp, timestamp = reversed(
(end_timestamp, timestamp)
)
end_timestamp, timestamp = reversed((end_timestamp, timestamp))
_end = "to %s" % end_timestamp if len(parsed_qs["t"]) == 2 else "onward"
embed = discord.Embed(
title="Trimming...",
description=f"Trimming from {timestamp} seconds {_end}.\nThis may take a while.",
colour=discord.Colour.blurple()
colour=discord.Colour.blurple(),
)
await ctx.edit(embed=embed)
for file in tempdir.glob("%s-*" % ctx.user.id):
@ -1087,7 +1088,7 @@ class OtherCog(commands.Cog):
"-y",
"-c",
"copy",
str(file)
str(file),
]
if end_timestamp is not None:
minutes, seconds = divmod(end_timestamp, 60)
@ -1096,13 +1097,7 @@ class OtherCog(commands.Cog):
_args.insert(6, "{!s}:{!s}:{!s}".format(*map(round, (hours, minutes, seconds))))
await self.bot.loop.run_in_executor(
None,
partial(
subprocess.run,
_args,
check=True,
capture_output=True
)
None, partial(subprocess.run, _args, check=True, capture_output=True)
)
bak.unlink(True)
except subprocess.CalledProcessError as e:
@ -1111,16 +1106,15 @@ class OtherCog(commands.Cog):
embed=discord.Embed(
title="Error",
description=f"Trimming failed:\n```\n{e}\n```",
colour=discord.Colour.red()
colour=discord.Colour.red(),
),
delete_after=30
delete_after=30,
)
embed = discord.Embed(
title="Downloaded!",
description="",
colour=discord.Colour.green()
title="Downloaded %r!" % title, description="", colour=discord.Colour.green(), url=webpage_url
)
embed.set_thumbnail(url=thumbnail_url)
del logger
files = []
@ -1135,13 +1129,15 @@ class OtherCog(commands.Cog):
while st_r > 1024:
st_r /= 1024
units.pop(0)
embed.description += "\N{warning sign}\ufe0f {} is too large to upload ({!s}{}" \
", max is {}MB).\n".format(
file.name,
round(st_r, 2),
units[0],
REAL_MAX_SIZE_MB,
)
embed.description += (
"\N{warning sign}\ufe0f {} is too large to upload ({!s}{}"
", max is {}MB).\n".format(
file.name,
round(st_r, 2),
units[0],
REAL_MAX_SIZE_MB,
)
)
continue
else:
files.append(discord.File(file, file.name))
@ -1149,13 +1145,13 @@ class OtherCog(commands.Cog):
if not files:
embed.description += "No files to upload. Directory list:\n%s" % (
"\n".join(r'\* ' + f.name for f in tempdir.iterdir())
"\n".join(r"\* " + f.name for f in tempdir.iterdir())
)
return await ctx.edit(embed=embed)
else:
_desc = embed.description
embed.description += f"Uploading {len(files)} file(s):\n%s" % (
"\n".join('* `%s`' % f.filename for f in files)
"\n".join("* `%s`" % f.filename for f in files)
)
await ctx.edit(embed=embed)
await ctx.channel.trigger_typing()
@ -1172,25 +1168,21 @@ class OtherCog(commands.Cog):
await ctx.edit(embed=None)
except discord.NotFound:
pass
self.bot.loop.create_task(bgtask())
@commands.slash_command(name="text-to-mp3")
@commands.cooldown(5, 600, commands.BucketType.user)
async def text_to_mp3(
self,
self,
ctx: discord.ApplicationContext,
speed: discord.Option(
int,
"The speed of the voice. Default is 150.",
required=False,
default=150
),
speed: discord.Option(int, "The speed of the voice. Default is 150.", required=False, default=150),
voice: discord.Option(
str,
"The voice to use. Some may cause timeout.",
autocomplete=discord.utils.basic_autocomplete(VOICES),
default="default"
)
default="default",
),
):
"""Converts text to MP3. 5 uses per 10 minutes."""
if voice not in VOICES:
@ -1207,9 +1199,9 @@ class OtherCog(commands.Cog):
placeholder="Enter text to read",
min_length=1,
max_length=4000,
style=discord.InputTextStyle.long
style=discord.InputTextStyle.long,
),
title="Convert text to an MP3"
title="Convert text to an MP3",
)
async def callback(self, interaction: discord.Interaction):
@ -1233,12 +1225,12 @@ class OtherCog(commands.Cog):
assert no_exists < 300, "File does not exist for 5 minutes."
no_exists += 1
return True
stat = os.stat(target_fn)
for _result in last_3_sizes:
if stat.st_size != _result:
return True
return False
while should_loop():
@ -1261,9 +1253,7 @@ class OtherCog(commands.Cog):
_msg = await interaction.followup.send("Downloading text...")
try:
response = await _self.http.get(
_url,
headers={"User-Agent": "Mozilla/5.0"},
follow_redirects=True
_url, headers={"User-Agent": "Mozilla/5.0"}, follow_redirects=True
)
if response.status_code != 200:
await _msg.edit(content=f"Failed to download text. Status code: {response.status_code}")
@ -1291,10 +1281,7 @@ class OtherCog(commands.Cog):
start_time = time()
task = _bot.loop.create_task(assurance_task())
try:
mp3, size = await asyncio.wait_for(
_bot.loop.run_in_executor(None, _convert, text_pre),
timeout=600
)
mp3, size = await asyncio.wait_for(_bot.loop.run_in_executor(None, _convert, text_pre), timeout=600)
except asyncio.TimeoutError:
task.cancel()
await _msg.edit(content="Failed to convert text to MP3 - Timeout. Try shorter/less complex text.")
@ -1308,7 +1295,7 @@ class OtherCog(commands.Cog):
if size >= ctx.guild.filesize_limit - 1500:
await _msg.edit(
content=f"MP3 is too large ({size / 1024 / 1024}Mb vs "
f"{ctx.guild.filesize_limit / 1024 / 1024}Mb)"
f"{ctx.guild.filesize_limit / 1024 / 1024}Mb)"
)
return
fn = ""
@ -1323,19 +1310,16 @@ class OtherCog(commands.Cog):
fn += word + "-"
fn = fn[:-1]
fn = fn[:28]
await _msg.edit(
content="Here's your MP3!",
file=discord.File(mp3, filename=fn + ".mp3")
)
await _msg.edit(content="Here's your MP3!", file=discord.File(mp3, filename=fn + ".mp3"))
await ctx.send_modal(TextModal())
@commands.slash_command()
@commands.cooldown(5, 10, commands.BucketType.user)
@commands.max_concurrency(1, commands.BucketType.user)
async def quote(self, ctx: discord.ApplicationContext):
"""Generates a random quote"""
emoji = discord.PartialEmoji(name='loading', animated=True, id=1101463077586735174)
emoji = discord.PartialEmoji(name="loading", animated=True, id=1101463077586735174)
async def get_quote() -> str | discord.File:
try:
@ -1358,10 +1342,7 @@ class OtherCog(commands.Cog):
class GenerateNewView(discord.ui.View):
def __init__(self):
super().__init__(
timeout=300,
disable_on_timeout=True
)
super().__init__(timeout=300, disable_on_timeout=True)
async def __aenter__(self):
self.disable_all_items()
@ -1381,7 +1362,7 @@ class OtherCog(commands.Cog):
@discord.ui.button(
label="New Quote",
style=discord.ButtonStyle.green,
emoji=discord.PartialEmoji.from_str("\U000023ed\U0000fe0f")
emoji=discord.PartialEmoji.from_str("\U000023ed\U0000fe0f"),
)
async def new_quote(self, _, interaction: discord.Interaction):
await interaction.response.defer(invisible=True)
@ -1394,9 +1375,7 @@ class OtherCog(commands.Cog):
return await followup.edit(content=new_result, view=GenerateNewView())
@discord.ui.button(
label="Regenerate",
style=discord.ButtonStyle.blurple,
emoji=discord.PartialEmoji.from_str("\U0001f504")
label="Regenerate", style=discord.ButtonStyle.blurple, emoji=discord.PartialEmoji.from_str("\U0001f504")
)
async def regenerate(self, _, interaction: discord.Interaction):
await interaction.response.defer(invisible=True)
@ -1406,7 +1385,7 @@ class OtherCog(commands.Cog):
return await interaction.followup.send(
"\N{cross mark} Message is starred and cannot be regenerated. You can press "
"'New Quote' to generate a new quote instead.",
ephemeral=True
ephemeral=True,
)
new_result = await get_quote()
if isinstance(new_result, discord.File):
@ -1414,11 +1393,7 @@ class OtherCog(commands.Cog):
else:
return await interaction.edit_original_response(content=new_result)
@discord.ui.button(
label="Delete",
style=discord.ButtonStyle.red,
emoji="\N{wastebasket}\U0000fe0f"
)
@discord.ui.button(label="Delete", style=discord.ButtonStyle.red, emoji="\N{wastebasket}\U0000fe0f")
async def delete(self, _, interaction: discord.Interaction):
await interaction.response.defer(invisible=True)
await interaction.delete_original_response()
@ -1435,13 +1410,12 @@ class OtherCog(commands.Cog):
@commands.cooldown(1, 30, commands.BucketType.user)
@commands.max_concurrency(1, commands.BucketType.user)
async def ocr(
self,
ctx: discord.ApplicationContext,
attachment: discord.Option(
discord.SlashCommandOptionType.attachment,
description="Image to perform OCR on",
)
self,
ctx: discord.ApplicationContext,
attachment: discord.Option(
discord.SlashCommandOptionType.attachment,
description="Image to perform OCR on",
),
):
"""OCRs an image"""
await ctx.defer()
@ -1468,13 +1442,8 @@ class OtherCog(commands.Cog):
response = await self.http.put(
"https://api.mystb.in/paste",
json={
"files": [
{
"filename": "ocr.txt",
"content": text
}
],
}
"files": [{"filename": "ocr.txt", "content": text}],
},
)
response.raise_for_status()
except httpx.HTTPError:
@ -1484,7 +1453,7 @@ class OtherCog(commands.Cog):
with Timer(timings, "Respond (URL)"):
embed = discord.Embed(
description="View on [mystb.in](%s)" % ("https://mystb.in/" + data["id"]),
colour=discord.Colour.dark_theme()
colour=discord.Colour.dark_theme(),
)
await ctx.respond(embed=embed)
timings["Upload text to mystbin"] = _t.total
@ -1541,11 +1510,7 @@ class OtherCog(commands.Cog):
@commands.cooldown(1, 180, commands.BucketType.user)
@commands.max_concurrency(1, commands.BucketType.user)
async def sherlock(
self,
ctx: discord.ApplicationContext,
username: str,
search_nsfw: bool = False,
use_tor: bool = False
self, ctx: discord.ApplicationContext, username: str, search_nsfw: bool = False, use_tor: bool = False
):
"""Sherlocks a username."""
# git clone https://github.com/sherlock-project/sherlock.git && cd sherlock && docker build -t sherlock .
@ -1563,11 +1528,9 @@ class OtherCog(commands.Cog):
embed = discord.Embed(
title="Sherlocking username %s" % chars[n % 4],
description=f"Elapsed: {elapsed:.0f}s",
colour=discord.Colour.dark_theme()
)
await ctx.edit(
embed=embed
colour=discord.Colour.dark_theme(),
)
await ctx.edit(embed=embed)
n += 1
await ctx.defer()
@ -1582,9 +1545,10 @@ class OtherCog(commands.Cog):
"-v",
f"{tempdir}:/opt/sherlock/results",
"sherlock",
"--folderoutput", "/opt/sherlock/results",
"--folderoutput",
"/opt/sherlock/results",
"--print-found",
"--csv"
"--csv",
]
if search_nsfw:
command.append("--nsfw")
@ -1647,6 +1611,7 @@ class OtherCog(commands.Cog):
@discord.guild_only()
async def opusinate(self, ctx: discord.ApplicationContext, file: discord.Attachment, size_mb: float = 8):
"""Converts the given file into opus with the given size."""
def humanise(v: int) -> str:
units = ["B", "KB", "MB", "GB", "TB", "PB", "EB"]
while v > 1024:
@ -1685,7 +1650,7 @@ class OtherCog(commands.Cog):
"a", # select audio-nly
str(location),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
@ -1698,9 +1663,7 @@ class OtherCog(commands.Cog):
try:
stream = metadata["streams"].pop()
except IndexError:
return await ctx.respond(
":x: No audio streams to transcode."
)
return await ctx.respond(":x: No audio streams to transcode.")
duration = float(metadata["format"]["duration"])
bit_rate = math.floor(int(metadata["format"]["bit_rate"]) / 1024)
channels = int(stream["channels"])
@ -1728,29 +1691,27 @@ class OtherCog(commands.Cog):
"-b:a",
"%sK" % end_br,
"-y",
output_file.name
output_file.name,
]
process = await asyncio.create_subprocess_exec(
command[0],
*command[1:],
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
command[0], *command[1:], stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
return await ctx.respond(
":x: There was an error while transcoding:\n```\n%s\n```" % discord.utils.escape_markdown(
stderr.decode()
)
":x: There was an error while transcoding:\n```\n%s\n```"
% discord.utils.escape_markdown(stderr.decode())
)
output_location = Path(output_file.name)
stat = output_location.stat()
content = ("\N{white heavy check mark} Transcoded from %r to opus @ %dkbps.\n\n"
"* Source: %dKbps\n* Target: %dKbps\n* Ceiling: %dKbps\n* Calculated: %dKbps\n"
"* Duration: %.1f seconds\n* Input size: %s\n* Output size: %s\n* Difference: %s"
" (%dKbps)") % (
content = (
"\N{white heavy check mark} Transcoded from %r to opus @ %dkbps.\n\n"
"* Source: %dKbps\n* Target: %dKbps\n* Ceiling: %dKbps\n* Calculated: %dKbps\n"
"* Duration: %.1f seconds\n* Input size: %s\n* Output size: %s\n* Difference: %s"
" (%dKbps)"
) % (
codec,
end_br,
bit_rate,
@ -1761,33 +1722,21 @@ class OtherCog(commands.Cog):
humanise(file.size),
humanise(stat.st_size),
humanise(file.size - stat.st_size),
bit_rate - end_br
bit_rate - end_br,
)
if stat.st_size <= max_size or share is False:
if stat.st_size >= (size_bytes - 100):
return await ctx.respond(
":x: File was too large."
)
return await ctx.respond(
content,
file=discord.File(output_location)
)
return await ctx.respond(":x: File was too large.")
return await ctx.respond(content, file=discord.File(output_location))
else:
share_location = Path("/mnt/vol/share/tmp/") / output_location.name
share_location.touch(0o755)
await self.bot.loop.run_in_executor(
None,
functools.partial(
shutil.copy,
output_location,
share_location
)
None, functools.partial(shutil.copy, output_location, share_location)
)
return await ctx.respond(
"%s\n* [Download](https://droplet.nexy7574.co.uk/share/tmp/%s)" % (
content,
output_location.name
)
"%s\n* [Download](https://droplet.nexy7574.co.uk/share/tmp/%s)"
% (content, output_location.name)
)

15
cogs/starboard.py
View file

@ -1,14 +1,14 @@
import asyncio
import textwrap
import io
import textwrap
from typing import Tuple
from urllib.parse import urlparse
import httpx
from typing import Tuple
import discord
import httpx
import orm
from discord.ext import commands
from utils.db import StarBoardMessage
@ -22,7 +22,7 @@ class StarBoardCog(commands.Cog):
async with httpx.AsyncClient(
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.69; Win64; x64) "
"LCC-Bot-Scraper/0 (https://github.com/EEKIM10/LCC-bot)"
"LCC-Bot-Scraper/0 (https://github.com/EEKIM10/LCC-bot)"
}
) as session:
image = starboard_message.embeds[0].image
@ -46,10 +46,7 @@ class StarBoardCog(commands.Cog):
file.seek(0)
embed = starboard_message.embeds[0].copy()
embed.set_image(url="attachment://" + filename)
embeds = [
embed,
*starboard_message.embeds[1:]
]
embeds = [embed, *starboard_message.embeds[1:]]
await starboard_message.edit(embeds=embeds, file=discord.File(file, filename=filename))
async def generate_starboard_embed(self, message: discord.Message) -> discord.Embed:

32
cogs/timetable.py
View file

@ -1,14 +1,14 @@
import random
from typing import Optional, Union, Dict
import discord
from discord.ext import commands, tasks
import json
import random
from datetime import datetime, time, timedelta, timezone
from pathlib import Path
from typing import Dict, Optional, Union
import config
from utils import console, TimeTableDaySwitcherView
from datetime import time, datetime, timedelta, timezone
import discord
from discord.ext import commands, tasks
from utils import TimeTableDaySwitcherView, console
def schedule_times():
@ -178,8 +178,7 @@ class TimeTableCog(commands.Cog):
next_lesson.setdefault("room", "unknown")
next_lesson.setdefault("start_datetime", discord.utils.utcnow())
text = "[tt] Next Lesson: {0[name]!r} with {0[tutor]} in {0[room]} - Starts {1}".format(
next_lesson,
discord.utils.format_dt(next_lesson['start_datetime'], 'R')
next_lesson, discord.utils.format_dt(next_lesson["start_datetime"], "R")
)
else:
lesson.setdefault("name", "unknown")
@ -188,14 +187,13 @@ class TimeTableCog(commands.Cog):
lesson.setdefault("start_datetime", discord.utils.utcnow())
if lesson["name"].lower() != "lunch":
text = "[tt] Current Lesson: {0[name]!r} with {0[tutor]} in {0[room]} - ends {1}".format(
lesson,
discord.utils.format_dt(lesson['end_datetime'], 'R')
lesson, discord.utils.format_dt(lesson["end_datetime"], "R")
)
else:
text = "[tt] \U0001f37d\U0000fe0f Lunch! {0}-{1}, ends in {2}".format(
discord.utils.format_dt(lesson['start_datetime'], 't'),
discord.utils.format_dt(lesson['end_datetime'], 't'),
discord.utils.format_dt(lesson['end_datetime'], 'R')
discord.utils.format_dt(lesson["start_datetime"], "t"),
discord.utils.format_dt(lesson["end_datetime"], "t"),
discord.utils.format_dt(lesson["end_datetime"], "R"),
)
next_lesson = self.next_lesson(date)
if next_lesson:
@ -209,8 +207,8 @@ class TimeTableCog(commands.Cog):
)
else:
text = "[tt] \U0001f37d\U0000fe0f Lunch! {0}-{1}.".format(
discord.utils.format_dt(lesson['start_datetime'], 't'),
discord.utils.format_dt(lesson['end_datetime'], 't'),
discord.utils.format_dt(lesson["start_datetime"], "t"),
discord.utils.format_dt(lesson["end_datetime"], "t"),
)
if no_prefix:
@ -282,7 +280,7 @@ class TimeTableCog(commands.Cog):
view = TimeTableDaySwitcherView(ctx.user, self, date)
view.update_buttons()
await ctx.respond(text, view=view)
@commands.slash_command(name="exams")
async def _exams(self, ctx: discord.ApplicationContext):
"""Shows when exams are."""

41
cogs/uptime.py
View file

@ -1,18 +1,19 @@
import asyncio
import datetime
import hashlib
import json
import random
import hashlib
import time
from datetime import timedelta
from pathlib import Path
from typing import Dict, Tuple, List, Optional
from urllib.parse import urlparse, parse_qs, ParseResult
from typing import Dict, List, Optional, Tuple
from urllib.parse import ParseResult, parse_qs, urlparse
import discord
import httpx
from discord.ext import commands, pages, tasks
from httpx import AsyncClient, Response
from discord.ext import commands, tasks, pages
from utils import UptimeEntry, console
"""
@ -339,9 +340,7 @@ class UptimeCompetition(commands.Cog):
embeds = entries = []
for _target in targets:
_target = self.get_target(_target, _target)
query = UptimeEntry.objects.filter(timestamp__gte=look_back_timestamp).filter(
target_id=_target["id"]
)
query = UptimeEntry.objects.filter(timestamp__gte=look_back_timestamp).filter(target_id=_target["id"])
query = query.order_by("-timestamp")
start = time.time()
entries = await query.all()
@ -366,12 +365,10 @@ class UptimeCompetition(commands.Cog):
new_embed.set_footer(text=f"Total query time: {minutes:.0f}m {seconds:.2f}s")
else:
new_embed.set_footer(text=f"Total query time: {total_query_time:.2f}s")
new_embed.set_footer(
text=f"{new_embed.footer.text} | {len(entries):,} rows"
)
new_embed.set_footer(text=f"{new_embed.footer.text} | {len(entries):,} rows")
embeds = [new_embed]
await ctx.respond(embeds=embeds)
@uptime.command(name="speedtest")
@commands.is_owner()
async def stats_speedtest(self, ctx: discord.ApplicationContext):
@ -430,22 +427,17 @@ class UptimeCompetition(commands.Cog):
return (end - start) * 1000, 1
case "random":
start = time.time()
e = await UptimeEntry.objects.offset(
random.randint(0, 1000)
).first()
e = await UptimeEntry.objects.offset(random.randint(0, 1000)).first()
end = time.time()
return (end - start) * 1000, 1
case _:
raise ValueError(f"Unknown test name: {name}")
def gen_embed(_copy):
embed = discord.Embed(
title='\N{HOURGLASS} Speedtest Results',
colour=discord.Colour.red()
)
embed = discord.Embed(title="\N{HOURGLASS} Speedtest Results", colour=discord.Colour.red())
for _name in _copy.keys():
if _copy[_name] is None:
embed.add_field(name=_name, value='Waiting...')
embed.add_field(name=_name, value="Waiting...")
else:
_time, _row = _copy[_name]
_time /= 1000
@ -453,14 +445,11 @@ class UptimeCompetition(commands.Cog):
if _time >= 60:
minutes, seconds = divmod(_time, 60)
ts = f'{minutes:.0f}m {seconds:.2f}s'
ts = f"{minutes:.0f}m {seconds:.2f}s"
else:
ts = f'{_time:.2f}s'
ts = f"{_time:.2f}s"
embed.add_field(
name=_name,
value=f'{ts}, {_row:,} rows ({rows_per_second:.2f} rows/s)'
)
embed.add_field(name=_name, value=f"{ts}, {_row:,} rows ({rows_per_second:.2f} rows/s)")
return embed
embed = gen_embed(tests)
@ -470,7 +459,7 @@ class UptimeCompetition(commands.Cog):
tests[test_key] = await run_test(test_key)
embed = gen_embed(tests)
await ctx.edit(embed=embed)
embed.colour = discord.Colour.green()
await ctx.edit(embed=embed)

80
cogs/voice.py
View file

@ -1,15 +1,15 @@
import asyncio
import functools
import io
import json
import shutil
import asyncio
import subprocess
import tempfile
from pathlib import Path
import discord
import httpx
import yt_dlp
import tempfile
from pathlib import Path
from discord.ext import commands
@ -17,7 +17,7 @@ class TransparentQueue(asyncio.Queue):
def __init__(self, maxsize: int = 0) -> None:
super().__init__(maxsize)
self._internal_queue = []
async def put(self, item):
await super().put(item)
self._internal_queue.append(item)
@ -35,7 +35,7 @@ class YTDLSource(discord.PCMVolumeTransformer):
self.title = data.get("title")
self.url = data.get("url")
@property
def duration(self):
return self.data.get("duration")
@ -44,9 +44,7 @@ class YTDLSource(discord.PCMVolumeTransformer):
async def from_url(cls, ytdl: yt_dlp.YoutubeDL, url, *, loop=None, stream=False):
ffmpeg_options = {"options": "-vn -b:a 44.1k"}
loop = loop or asyncio.get_event_loop()
data = await loop.run_in_executor(
None, lambda: ytdl.extract_info(url, download=not stream)
)
data = await loop.run_in_executor(None, lambda: ytdl.extract_info(url, download=not stream))
if "entries" in data:
if not data["entries"]:
@ -79,7 +77,7 @@ class VoiceCog(commands.Cog):
"paths": {
"home": str(self.cache),
"temp": str(self.cache),
}
},
}
self.yt_dl = yt_dlp.YoutubeDL(self.ytdl_options)
self.queue = TransparentQueue(100)
@ -101,7 +99,7 @@ class VoiceCog(commands.Cog):
embed = discord.Embed(
description=f"Now playing: [{player.title}]({player.url}), as requested by {ctx.author.mention}, "
f"{discord.utils.format_dt(inserted_at, 'R')}.",
f"{discord.utils.format_dt(inserted_at, 'R')}.",
color=discord.Color.green(),
)
try:
@ -127,11 +125,8 @@ class VoiceCog(commands.Cog):
def after(e):
self.song_done.set()
if e:
self.bot.loop.create_task(
ctx.respond(
f"An error occurred while playing the audio: {e}"
)
)
self.bot.loop.create_task(ctx.respond(f"An error occurred while playing the audio: {e}"))
return after
async def unblock(self, func, *args, **kwargs):
@ -140,10 +135,10 @@ class VoiceCog(commands.Cog):
@commands.slash_command(name="play")
async def stream(
self,
ctx: discord.ApplicationContext,
url: str,
volume: float = 100,
self,
ctx: discord.ApplicationContext,
url: str,
volume: float = 100,
):
"""Streams a URL using yt-dl"""
if not ctx.user.voice:
@ -232,6 +227,7 @@ class VoiceCog(commands.Cog):
@commands.slash_command(name="skip")
async def skip(self, ctx: discord.ApplicationContext):
"""Skips the current song"""
class VoteSkipDialog(discord.ui.View):
def __init__(self):
super().__init__()
@ -250,19 +246,13 @@ class VoiceCog(commands.Cog):
ctx.voice_client.stop()
self.stop()
self.disable_all_items()
return await interaction.edit_original_response(
view=self,
content="Skipped song."
)
return await interaction.edit_original_response(view=self, content="Skipped song.")
if len(self.voted) >= target:
ctx.voice_client.stop()
self.stop()
self.disable_all_items()
return await interaction.edit_original_response(
view=self,
content="Skipped song."
)
return await interaction.edit_original_response(view=self, content="Skipped song.")
else:
await ctx.respond(
f"Voted to skip. %d/%d" % (len(self.voted), target),
@ -272,8 +262,7 @@ class VoiceCog(commands.Cog):
self.disable_all_items()
await interaction.edit_original_response(
view=self,
content="Vote skip (%d/%d)." % (len(self.voted), target)
view=self, content="Vote skip (%d/%d)." % (len(self.voted), target)
)
if not ctx.guild.voice_client:
@ -314,7 +303,7 @@ class VoiceCog(commands.Cog):
data = data[key]
last_key.append(key)
else:
content = "Key %r not found in metadata (got as far as %r)." % (key, '.'.join(last_key))
content = "Key %r not found in metadata (got as far as %r)." % (key, ".".join(last_key))
break
json.dump(data, file, indent=4)
file.seek(0)
@ -343,15 +332,12 @@ class VoiceCog(commands.Cog):
@commands.slash_command(name="boost-audio")
async def boost_audio(
self,
ctx: discord.ApplicationContext,
file: discord.Attachment,
level: discord.Option(
float,
"A level (in percentage) of volume (e.g. 150 = 150%)",
min_value=0.1,
max_value=999.99
)
self,
ctx: discord.ApplicationContext,
file: discord.Attachment,
level: discord.Option(
float, "A level (in percentage) of volume (e.g. 150 = 150%)", min_value=0.1, max_value=999.99
),
):
"""Boosts an audio file's audio level."""
await ctx.defer()
@ -361,7 +347,7 @@ class VoiceCog(commands.Cog):
with tempfile.TemporaryDirectory("jimmy-audio-boost-") as temp_dir_raw:
temp_dir = Path(temp_dir_raw).resolve()
_input = temp_dir / file.filename
output = _input.with_name(_input.name + "-processed" + '.'.join(_input.suffixes))
output = _input.with_name(_input.name + "-processed" + ".".join(_input.suffixes))
await file.save(_input)
proc: subprocess.CompletedProcess = await self.bot.loop.run_in_executor(
@ -379,8 +365,8 @@ class VoiceCog(commands.Cog):
"volume=%d" % (level / 100),
str(output),
),
capture_output=True
)
capture_output=True,
),
)
if proc.returncode == 0:
if output.stat().st_size >= (25 * 1024 * 1024) + len(output.name):
@ -389,14 +375,8 @@ class VoiceCog(commands.Cog):
else:
data = {
"files": [
{
"content": proc.stderr.decode() or 'empty',
"filename": "stderr.txt"
},
{
"content": proc.stdout.decode() or 'empty',
"filename": "stdout.txt"
}
{"content": proc.stderr.decode() or "empty", "filename": "stderr.txt"},
{"content": proc.stdout.decode() or "empty", "filename": "stdout.txt"},
]
}
response = await httpx.AsyncClient().put("https://api.mystb.in/paste", json=data)

7
config.example.py
View file

@ -10,6 +10,7 @@
import datetime
import os
import discord
# The IDs of guilds the bot should be in; used to determine where to make slash commands
@ -51,11 +52,7 @@ WEB_SERVER = True # change this to False to disable it
# Or change uvicorn settings (see: https://www.uvicorn.org/settings/)
# Note that passing `host` or `port` will raise an error, as those are configured above.
UVICORN_CONFIG = {
"log_level": "error",
"access_log": False,
"lifespan": "off"
}
UVICORN_CONFIG = {"log_level": "error", "access_log": False, "lifespan": "off"}
# Only change this if you want to test changes to the bot without sending too much traffic to discord.
# Connect modes:

2
config_docker.py
View file

@ -2,8 +2,8 @@
# This format is very limited and environment variables were ditched very early on in favor of a config file
# Do feel free to overwrite this file and re-build the docker image - this is effectively a stub.
import os
import datetime
import os
# A few defaults that can't be set in the environment
reminders = {

13
main.py
View file

@ -1,11 +1,12 @@
import asyncio
import sys
from datetime import datetime, timedelta, timezone
import config
import discord
from discord.ext import commands
import config
from datetime import datetime, timezone, timedelta
from utils import console, get_or_none, JimmyBans, JimmyBanException
from utils import JimmyBanException, JimmyBans, console, get_or_none
from utils.client import bot
@ -97,8 +98,10 @@ if __name__ == "__main__":
bot.started_at = discord.utils.utcnow()
if getattr(config, "WEB_SERVER", True):
from web.server import app
import uvicorn
from web.server import app
app.state.bot = bot
http_config = uvicorn.Config(
@ -106,7 +109,7 @@ if __name__ == "__main__":
host=getattr(config, "HTTP_HOST", "127.0.0.1"),
port=getattr(config, "HTTP_PORT", 3762),
loop="asyncio",
**getattr(config, "UVICORN_CONFIG", {})
**getattr(config, "UVICORN_CONFIG", {}),
)
server = uvicorn.Server(http_config)
console.log("Starting web server...")

12
tests/test_timetable.py
View file

@ -1,9 +1,11 @@
import pytest
from pathlib import Path
from utils import Tutors
from typing import Union
import warnings
import json
import warnings
from pathlib import Path
from typing import Union
import pytest
from utils import Tutors
file = (Path(__file__).parent.parent / "utils" / "timetable.json").resolve()

6
utils/__init__.py
View file

@ -1,12 +1,12 @@
from typing import Optional, TYPE_CHECKING
import time
from typing import TYPE_CHECKING, Optional
from urllib.parse import urlparse
import time
from discord.ext import commands
from ._email import *
from .db import *
from .console import *
from .db import *
from .views import *
if TYPE_CHECKING:

9
utils/_email.py
View file

@ -1,11 +1,10 @@
import secrets
import discord
import config
import aiosmtplib as smtp
from email.message import EmailMessage
import aiosmtplib as smtp
import config
import discord
gmail_cfg = {"addr": "smtp.gmail.com", "username": config.email, "password": config.email_password, "port": 465}
TOKEN_LENGTH = 16

24
utils/client.py
View file

@ -1,19 +1,21 @@
import asyncio
import sys
from pathlib import Path
import discord
import config
from asyncio import Lock
from discord.ext import commands
from datetime import datetime, timezone
from typing import Optional, Dict, TYPE_CHECKING, Union
from pathlib import Path
from typing import TYPE_CHECKING, Dict, Optional, Union
import config
import discord
from discord.ext import commands
if TYPE_CHECKING:
from uvicorn import Server, Config
from asyncio import Task
from uvicorn import Config, Server
__all__ = ("Bot", 'bot')
__all__ = ("Bot", "bot")
# noinspection PyAbstractClass
@ -22,8 +24,9 @@ class Bot(commands.Bot):
web: Optional[Dict[str, Union[Server, Config, Task]]]
def __init__(self, intents: discord.Intents, guilds: list[int], extensions: list[str], prefixes: list[str]):
from .db import registry
from .console import console
from .db import registry
super().__init__(
command_prefix=commands.when_mentioned_or(*prefixes),
debug_guilds=guilds,
@ -50,6 +53,7 @@ class Bot(commands.Bot):
console.log(f"Loaded extension [green]{ext}")
if getattr(config, "CONNECT_MODE", None) == 2:
async def connect(self, *, reconnect: bool = True) -> None:
self.console.log("Exit target 2 reached, shutting down (not connecting to discord).")
return
@ -58,7 +62,7 @@ class Bot(commands.Bot):
e_type, e, tb = sys.exc_info()
if isinstance(e, discord.NotFound) and e.code == 10062: # invalid interaction
return
if isinstance(e, discord.CheckFailure) and 'The global check once functions failed.' in str(e):
if isinstance(e, discord.CheckFailure) and "The global check once functions failed." in str(e):
return
await super().on_error(event, *args, **kwargs)

7
utils/console.py
View file

@ -1,4 +1,5 @@
import shutil
from rich.console import Console
_col, _ = shutil.get_terminal_size((80, 20))
@ -7,8 +8,4 @@ if _col == 80:
__all__ = ("console",)
console = Console(
width=_col,
soft_wrap=True,
tab_size=4
)
console = Console(width=_col, soft_wrap=True, tab_size=4)

10
utils/db.py
View file

@ -1,14 +1,14 @@
import datetime
import os
import sys
import discord
import uuid
from typing import TYPE_CHECKING, Optional, TypeVar
from enum import IntEnum, auto
from pathlib import Path
from typing import TYPE_CHECKING, Optional, TypeVar
import discord
import orm
from databases import Database
import os
from pathlib import Path
class Tutors(IntEnum):
@ -87,7 +87,7 @@ class Student(orm.Model):
"name": orm.String(min_length=2, max_length=32),
"access_token": orm.String(min_length=6, max_length=128, default=None, allow_null=True),
"ip_info": orm.JSON(default=None, allow_null=True),
"access_token_hash": orm.String(min_length=128, max_length=128, default=None, allow_null=True),
"access_token_hash": orm.String(min_length=128, max_length=128, default=None, allow_null=True),
}
if TYPE_CHECKING:
entry_id: uuid.UUID

11
utils/sentiment_analysis.py
View file

@ -1,14 +1,15 @@
# I have NO idea how this works
# I copied it from the tutorial
# However it works
import random
import re
import string
import random
from nltk import FreqDist, classify, NaiveBayesClassifier
from nltk.corpus import twitter_samples, stopwords, movie_reviews
from nltk.tag import pos_tag
from nltk.stem.wordnet import WordNetLemmatizer
from nltk import FreqDist, NaiveBayesClassifier, classify
from nltk.corpus import movie_reviews, stopwords, twitter_samples
from nltk.sentiment.vader import SentimentIntensityAnalyzer
from nltk.stem.wordnet import WordNetLemmatizer
from nltk.tag import pos_tag
positive_tweets = twitter_samples.strings("positive_tweets.json")
negative_tweets = twitter_samples.strings("negative_tweets.json")

14
utils/views.py
View file

@ -1,14 +1,22 @@
import random
import re
import secrets
import typing
from datetime import datetime, timedelta
import discord
import typing
import re
import orm
from discord.ui import View
from utils import send_verification_code, get_or_none, Student, VerifyCode, console, TOKEN_LENGTH, BannedStudentID
from utils import (
TOKEN_LENGTH,
BannedStudentID,
Student,
VerifyCode,
console,
get_or_none,
send_verification_code,
)
if typing.TYPE_CHECKING:
from cogs.timetable import TimeTableCog

164
web/server.py
View file

@ -1,24 +1,22 @@
import asyncio
import ipaddress
import os
import sys
import textwrap
import discord
import os
import httpx
from pathlib import Path
from datetime import datetime, timezone
from hashlib import sha512
from fastapi import FastAPI, HTTPException, Request, Header
from fastapi.responses import JSONResponse, RedirectResponse, HTMLResponse
from http import HTTPStatus
from pathlib import Path
import discord
import httpx
from config import guilds
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from starlette.websockets import WebSocket, WebSocketDisconnect
from utils import Student, get_or_none, VerifyCode, console, BannedStudentID
from utils import BannedStudentID, Student, VerifyCode, console, get_or_none
from utils.db import AccessTokens
from config import guilds
SF_ROOT = Path(__file__).parent / "static"
if SF_ROOT.exists() and SF_ROOT.is_dir():
@ -27,7 +25,7 @@ else:
StaticFiles = None
try:
from config import OAUTH_ID, OAUTH_SECRET, OAUTH_REDIRECT_URI
from config import OAUTH_ID, OAUTH_REDIRECT_URI, OAUTH_SECRET
except ImportError:
OAUTH_ID = OAUTH_SECRET = OAUTH_REDIRECT_URI = None
@ -50,6 +48,7 @@ if StaticFiles:
try:
from utils.client import bot
app.state.bot = bot
except ImportError:
bot = None
@ -60,13 +59,7 @@ app.state.last_sender_ts = datetime.utcnow()
@app.middleware("http")
async def check_bot_instanced(request, call_next):
if not request.app.state.bot:
return JSONResponse(
status_code=503,
content={"message": "Not ready."},
headers={
"Retry-After": "10"
}
)
return JSONResponse(status_code=503, content={"message": "Not ready."}, headers={"Retry-After": "10"})
return await call_next(request)
@ -74,10 +67,10 @@ async def check_bot_instanced(request, call_next):
def ping():
bot_started = datetime.now(tz=timezone.utc) - app.state.bot.started_at
return {
"ping": "pong",
"online": app.state.bot.is_ready(),
"ping": "pong",
"online": app.state.bot.is_ready(),
"latency": max(round(app.state.bot.latency, 2), 0.01),
"uptime": max(round(bot_started.total_seconds(), 2), 1)
"uptime": max(round(bot_started.total_seconds(), 2), 1),
}
@ -85,10 +78,7 @@ def ping():
async def authenticate(req: Request, code: str = None, state: str = None):
"""Begins Oauth flow (browser only)"""
if not OAUTH_ENABLED:
raise HTTPException(
501,
"OAuth is not enabled."
)
raise HTTPException(501, "OAuth is not enabled.")
if not (code and state) or state not in app.state.states:
value = os.urandom(4).hex()
@ -111,21 +101,16 @@ async def authenticate(req: Request, code: str = None, state: str = None):
"Please try again later.",
# Saying a suspected DDOS makes sense, there are 4,294,967,296 possible states, the likelyhood of a
# collision is 1 in 4,294,967,296.
headers={
"Retry-After": "300"
}
headers={"Retry-After": "300"},
)
app.state.states[value] = datetime.now()
return RedirectResponse(
discord.utils.oauth_url(
OAUTH_ID,
redirect_uri=OAUTH_REDIRECT_URI,
scopes=('identify', "connections", "guilds", "email")
) + f"&state={value}&prompt=none",
OAUTH_ID, redirect_uri=OAUTH_REDIRECT_URI, scopes=("identify", "connections", "guilds", "email")
)
+ f"&state={value}&prompt=none",
status_code=HTTPStatus.TEMPORARY_REDIRECT,
headers={
"Cache-Control": "no-store, no-cache"
}
headers={"Cache-Control": "no-store, no-cache"},
)
else:
app.state.states.pop(state)
@ -138,42 +123,30 @@ async def authenticate(req: Request, code: str = None, state: str = None):
"grant_type": "authorization_code",
"code": code,
"redirect_uri": OAUTH_REDIRECT_URI,
}
},
)
if response.status_code != 200:
raise HTTPException(
status_code=response.status_code,
detail=response.text
)
raise HTTPException(status_code=response.status_code, detail=response.text)
data = response.json()
access_token = data["access_token"]
# Now we can generate a token
token = sha512(access_token.encode()).hexdigest()
# Now we can get the user's info
response = app.state.http.get(
"https://discord.com/api/users/@me",
headers={
"Authorization": "Bearer " + data["access_token"]
}
"https://discord.com/api/users/@me", headers={"Authorization": "Bearer " + data["access_token"]}
)
if response.status_code != 200:
raise HTTPException(
status_code=response.status_code,
detail=response.text
)
raise HTTPException(status_code=response.status_code, detail=response.text)
user = response.json()
# Now we need to fetch the student from the database
student = await get_or_none(AccessTokens, user_id=user["id"])
if not student:
student = await AccessTokens.objects.create(
user_id=user["id"],
access_token=access_token
)
student = await AccessTokens.objects.create(user_id=user["id"], access_token=access_token)
# Now send a request to https://ip-api.com/json/{ip}?fields=status,city,zip,lat,lon,isp,query
_host = ipaddress.ip_address(req.client.host)
if not any((_host.is_loopback, _host.is_private, _host.is_reserved, _host.is_unspecified)):
@ -181,23 +154,19 @@ async def authenticate(req: Request, code: str = None, state: str = None):
f"http://ip-api.com/json/{req.client.host}?fields=status,city,zip,lat,lon,isp,query,proxy,hosting"
)
if response.status_code != 200:
raise HTTPException(
status_code=response.status_code,
detail=response.text
)
raise HTTPException(status_code=response.status_code, detail=response.text)
data = response.json()
if data["status"] != "success":
raise HTTPException(
status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
detail=f"Failed to get IP data for {req.client.host}: {data}."
detail=f"Failed to get IP data for {req.client.host}: {data}.",
)
else:
data = None
# Now we can update the student entry with this data
await student.update(ip_info=data, access_token_hash=token)
document = \
f"""
document = f"""
<!DOCTYPE html>
<html>
<head>
@ -217,12 +186,7 @@ f"""
"""
# And set it as a cookie
response = HTMLResponse(
document,
status_code=200,
headers={
"Location": GENERAL,
"Cache-Control": "max-age=604800"
}
document, status_code=200, headers={"Location": GENERAL, "Cache-Control": "max-age=604800"}
)
# set the cookie for at most 604800 seconds - expire after that
response.set_cookie(
@ -239,36 +203,25 @@ f"""
async def verify(code: str):
guild = app.state.bot.get_guild(guilds[0])
if not guild:
raise HTTPException(
status_code=503,
detail="Not ready."
)
raise HTTPException(status_code=503, detail="Not ready.")
# First, we need to fetch the code from the database
verify_code = await get_or_none(VerifyCode, code=code)
if not verify_code:
raise HTTPException(
status_code=404,
detail="Code not found."
)
raise HTTPException(status_code=404, detail="Code not found.")
# Now we need to fetch the student from the database
student = await get_or_none(Student, user_id=verify_code.bind)
if student:
raise HTTPException(
status_code=400,
detail="Already verified."
)
raise HTTPException(status_code=400, detail="Already verified.")
ban = await get_or_none(BannedStudentID, student_id=verify_code.student_id)
if ban is not None:
return await guild.kick(
reason=f"Attempted to verify with banned student ID {ban.student_id}"
f" (originally associated with account {ban.associated_account})"
f" (originally associated with account {ban.associated_account})"
)
await Student.objects.create(
id=verify_code.student_id, user_id=verify_code.bind, name=verify_code.name
)
await Student.objects.create(id=verify_code.student_id, user_id=verify_code.bind, name=verify_code.name)
await verify_code.delete()
role = discord.utils.find(lambda r: r.name.lower() == "verified", guild.roles)
member = await guild.fetch_member(verify_code.bind)
@ -284,10 +237,7 @@ async def verify(code: str):
console.log(f"[green]{verify_code.bind} verified ({verify_code.bind}/{verify_code.student_id})")
return RedirectResponse(
GENERAL,
status_code=308
)
return RedirectResponse(GENERAL, status_code=308)
@app.post("/bridge", include_in_schema=False, status_code=201)
@ -295,25 +245,17 @@ async def bridge(req: Request):
now = datetime.utcnow()
ts_diff = (now - app.state.last_sender_ts).total_seconds()
from discord.ext.commands import Paginator
body = await req.json()
if body["secret"] != app.state.bot.http.token:
raise HTTPException(
status_code=401,
detail="Invalid secret."
)
raise HTTPException(status_code=401, detail="Invalid secret.")
channel = app.state.bot.get_channel(1032974266527907901) # type: discord.TextChannel | None
if not channel:
raise HTTPException(
status_code=404,
detail="Channel does not exist."
)
raise HTTPException(status_code=404, detail="Channel does not exist.")
if len(body["message"]) > 6000:
raise HTTPException(
status_code=400,
detail="Message too long."
)
raise HTTPException(status_code=400, detail="Message too long.")
paginator = Paginator(prefix="", suffix="", max_size=1990)
for line in body["message"].splitlines():
try:
@ -323,9 +265,7 @@ async def bridge(req: Request):
if len(paginator.pages) > 1:
msg = None
if app.state.last_sender != body["sender"] or ts_diff >= 600:
msg = await channel.send(
f"**{body['sender']}**:"
)
msg = await channel.send(f"**{body['sender']}**:")
m = len(paginator.pages)
for n, page in enumerate(paginator.pages, 1):
await channel.send(
@ -333,31 +273,23 @@ async def bridge(req: Request):
allowed_mentions=discord.AllowedMentions.none(),
reference=msg,
silent=True,
suppress=n != m
suppress=n != m,
)
app.state.last_sender = body["sender"]
else:
content = f"**{body['sender']}**:\n>>> {body['message']}"
if app.state.last_sender == body["sender"] and ts_diff < 600:
content = f">>> {body['message']}"
await channel.send(
content,
allowed_mentions=discord.AllowedMentions.none(),
silent=True,
suppress=False
)
await channel.send(content, allowed_mentions=discord.AllowedMentions.none(), silent=True, suppress=False)
app.state.last_sender = body["sender"]
app.state.last_sender_ts = now
return {"status": "ok", "pages": len(paginator.pages)}
@app.websocket('/bridge/recv')
@app.websocket("/bridge/recv")
async def bridge_recv(ws: WebSocket, secret: str = Header(None)):
if secret != app.state.bot.http.token:
raise HTTPException(
status_code=401,
detail="Invalid secret."
)
raise HTTPException(status_code=401, detail="Invalid secret.")
queue: asyncio.Queue = app.state.bot.bridge_queue
await ws.accept()