ryg/royalnet
ryg/royalnet
1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-27 13:34:28 +00:00
Code Activity

Rewrite all network code and make it work somehow

Browse source
This commit is contained in:
Steffo 2019-10-19 00:24:35 +02:00
parent 9859d422f2
commit ecd4c92bb2
15 changed files with 272 additions and 270 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
royalnet/bots/generic.py
View file

@ -28,7 +28,8 @@ class GenericBot:
self._Interface = self._interface_factory()
self._Data = self._data_factory()
self.commands = {}
self.network_handlers: typing.Dict[str, typing.Callable[[typing.Any], typing.Awaitable[typing.Dict]]] = {}
self.network_handlers: typing.Dict[str, typing.Callable[["GenericBot", typing.Any],
typing.Awaitable[typing.Dict]]] = {}
for SelectedCommand in self.uninitialized_commands:
interface = self._Interface()
try:
@ -69,15 +70,17 @@ class GenericBot:
async def call_herald_action(ci, destination: str, event_name: str, args: typing.Dict) -> typing.Dict:
if self.network is None:
raise UnsupportedError("royalherald is not enabled on this bot")
raise UnsupportedError("Herald is not enabled on this bot")
request: rh.Request = rh.Request(handler=event_name, data=args)
response: rh.Response = await self.network.request(destination=destination, request=request)
if isinstance(response, rh.ResponseFailure):
raise CommandError(f"royalherald action code failed: {response}")
raise CommandError(f"Herald action call failed:\n"
f"[p]{response}[/p]")
elif isinstance(response, rh.ResponseSuccess):
return response.data
else:
raise TypeError(f"royalherald returned unknown response: {response}")
raise TypeError(f"Other Herald Link returned unknown response:\n"
f"[p]{response}[/p]")
return GenericInterface
@ -103,7 +106,7 @@ class GenericBot:
else:
log.debug(f"Using {network_handler} as handler for {request.handler}")
try:
response_data = await network_handler(**request.data)
response_data = await network_handler(self, **request.data)
return rh.ResponseSuccess(data=response_data)
except Exception as e:
sentry_sdk.capture_exception(e)

41
royalnet/packs/royal/commands/cv.py
View file

@ -1,19 +1,22 @@
import discord
import typing
from royalnet.commands import *
from royalnet.utils import NetworkHandler, andformat
from royalnet.utils import andformat
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class CvNH(NetworkHandler):
message_type = "discord_cv"
class CvCommand(Command):
name: str = "cv"
@classmethod
async def discord(cls, bot: DiscordBot, data: dict):
description: str = "Elenca le persone attualmente connesse alla chat vocale."
syntax: str = "[guildname] ['all']"
@staticmethod
async def _legacy_cv_handler(bot: DiscordBot, guild_name: typing.Optional[str], everyone: bool):
# Find the matching guild
if data["guild_name"]:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(data["guild_name"])
if guild_name:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(guild_name)
else:
guilds = bot.client.guilds
if len(guilds) == 0:
@ -47,7 +50,7 @@ class CvNH(NetworkHandler):
for member in members_in_channels[channel]:
member: typing.Union[discord.User, discord.Member]
# Ignore not-connected non-notable members
if not data["everyone"] and channel == 0 and len(member.roles) < 2:
if not everyone and channel == 0 and len(member.roles) < 2:
continue
# Ignore offline members
if member.status == discord.Status.offline and member.voice is None:
@ -108,24 +111,20 @@ class CvNH(NetworkHandler):
message += f" | ❓ Unknown activity"
message += "\n"
message += "\n"
return ResponseSuccess({"response": message})
return {"response": message}
class CvCommand(Command):
name: str = "cv"
description: str = "Elenca le persone attualmente connesse alla chat vocale."
syntax: str = "[guildname] ['all']"
_event_name = "_legacy_cv"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler("discord_cv", CvNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_cv_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
# noinspection PyTypeChecker
guild_name, everyone = args.match(r"(?:\[(.+)])?\s*(\S+)?\s*")
response = await self.interface.net_request(Request("discord_cv", {"guild_name": guild_name,
"everyone": bool(everyone)}),
destination="discord")
response = await self.interface.call_herald_action("discord", self._event_name, {
"guild_name": guild_name,
"everyone": everyone
})
await data.reply(response["response"])

3
royalnet/packs/royal/commands/dndspell.py
View file

@ -41,7 +41,8 @@ class DndspellCommand(Command):
for spell in j["spell"]:
self._dnddata.add(spell)
def _parse_spell(self, spell: dict) -> str:
@staticmethod
def _parse_spell(spell: dict) -> str:
string = f'✨ [b]{spell["name"]}[/b]\n'
if "source" in spell:
string += f'[i]{spell["source"]}, page {spell["page"]}[/i]\n'

38
royalnet/packs/royal/commands/pause.py
View file

@ -1,20 +1,21 @@
import typing
import discord
from royalnet.commands import *
from royalnet.utils import NetworkHandler
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class PauseNH(NetworkHandler):
message_type = "music_pause"
class PauseCommand(Command):
name: str = "pause"
# noinspection PyProtectedMember
@classmethod
async def discord(cls, bot: DiscordBot, data: dict):
description: str = "Mette in pausa o riprende la riproduzione della canzone attuale."
syntax = "[ [guild] ]"
@staticmethod
async def _legacy_pause_handler(bot: DiscordBot, guild_name: typing.Optional[str]):
# Find the matching guild
if data["guild_name"]:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(data["guild_name"])
if guild_name:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(guild_name)
else:
guilds = bot.client.guilds
if len(guilds) == 0:
@ -32,23 +33,20 @@ class PauseNH(NetworkHandler):
voice_client._player.resume()
else:
voice_client._player.pause()
return ResponseSuccess({"resumed": resume})
return {"resumed": resume}
class PauseCommand(Command):
name: str = "pause"
description: str = "Mette in pausa o riprende la riproduzione della canzone attuale."
syntax = "[ [guild] ]"
_event_name = "_legacy_pause"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler(PauseNH.message_type, PauseNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_pause_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
guild, = args.match(r"(?:\[(.+)])?")
response = await self.interface.net_request(Request("music_pause", {"guild_name": guild}), "discord")
guild_name, = args.match(r"(?:\[(.+)])?")
response = await self.interface.call_herald_action("discord", self._event_name, {
"guild_name": guild_name
})
if response["resumed"]:
await data.reply(f"▶️ Riproduzione ripresa.")
else:

44
royalnet/packs/royal/commands/play.py
View file

@ -3,21 +3,26 @@ import pickle
import datetime
import discord
from royalnet.commands import *
from royalnet.utils import NetworkHandler, asyncify
from royalnet.utils import asyncify
from royalnet.audio import YtdlDiscord
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class PlayNH(NetworkHandler):
message_type = "music_play"
class PlayCommand(Command):
name: str = "play"
@classmethod
async def discord(cls, bot: "DiscordBot", data: dict):
aliases = ["p"]
description: str = "Aggiunge un url alla coda della chat vocale."
syntax = "[ [guild] ] (url)"
@staticmethod
async def _legacy_play_handler(bot: "DiscordBot", guild_name: typing.Optional[str], url: str):
"""Handle a play Royalnet request. That is, add audio to a PlayMode."""
# Find the matching guild
if data["guild_name"]:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(data["guild_name"])
if guild_name:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(guild_name)
else:
guilds = bot.client.guilds
if len(guilds) == 0:
@ -35,7 +40,7 @@ class PlayNH(NetworkHandler):
"outtmpl": f"./downloads/{datetime.datetime.now().timestamp()}_%(title)s.%(ext)s"
}
# Start downloading
dfiles: typing.List[YtdlDiscord] = await asyncify(YtdlDiscord.create_from_url, data["url"], **ytdl_args)
dfiles: typing.List[YtdlDiscord] = await asyncify(YtdlDiscord.create_from_url, url, **ytdl_args)
await bot.add_to_music_data(dfiles, guild)
# Create response dictionary
response = {
@ -44,21 +49,14 @@ class PlayNH(NetworkHandler):
"discord_embed_pickle": str(pickle.dumps(dfile.info.to_discord_embed()))
} for dfile in dfiles]
}
return ResponseSuccess(response)
return response
class PlayCommand(Command):
name: str = "play"
aliases = ["p"]
description: str = "Aggiunge un url alla coda della chat vocale."
syntax = "[ [guild] ] (url)"
_event_name = "_legacy_play"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler(PlayNH.message_type, PlayNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_play_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
guild_name, url = args.match(r"(?:\[(.+)])?\s*<?(.+)>?")
@ -66,8 +64,10 @@ class PlayCommand(Command):
raise CommandError("PlayCommand only accepts URLs.\n"
"If you want to search a song on YouTube or Soundcloud, please use YoutubeCommand"
" or SoundcloudCommand!")
response = await self.interface.net_request(Request("music_play", {"url": url, "guild_name": guild_name}),
"discord")
response: dict = await self.interface.call_herald_action("discord", self._event_name, {
"guild_name": guild_name,
"url": url
})
if len(response["videos"]) == 0:
raise CommandError(f"Nessun file trovato.")
for video in response["videos"]:

73
royalnet/packs/royal/commands/playmode.py
View file

@ -1,42 +1,8 @@
import typing
import pickle
import discord
from royalnet.commands import *
from royalnet.utils import NetworkHandler
from royalnet.audio.playmodes import Playlist, Pool, Layers
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class PlaymodeNH(NetworkHandler):
message_type = "music_playmode"
@classmethod
async def discord(cls, bot: "DiscordBot", data: dict):
"""Handle a playmode Royalnet request. That is, change current PlayMode."""
# Find the matching guild
if data["guild_name"]:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(data["guild_name"])
else:
guilds = bot.client.guilds
if len(guilds) == 0:
raise CommandError("No guilds with the specified name found.")
if len(guilds) > 1:
raise CommandError("Multiple guilds with the specified name found.")
guild = list(bot.client.guilds)[0]
# Delete the previous PlayMode, if it exists
if bot.music_data[guild] is not None:
bot.music_data[guild].delete()
# Create the new PlayMode
if data["mode_name"] == "playlist":
bot.music_data[guild] = Playlist()
elif data["mode_name"] == "pool":
bot.music_data[guild] = Pool()
elif data["mode_name"] == "layers":
bot.music_data[guild] = Layers()
else:
raise CommandError("Unknown PlayMode specified.")
return ResponseSuccess()
class PlaymodeCommand(Command):
@ -48,13 +14,44 @@ class PlaymodeCommand(Command):
syntax = "[ [guild] ] (mode)"
@staticmethod
async def _legacy_playmode_handler(bot: "DiscordBot", guild_name: typing.Optional[str], mode_name: str):
"""Handle a playmode Royalnet request. That is, change current PlayMode."""
# Find the matching guild
if guild_name:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(guild_name)
else:
guilds = bot.client.guilds
if len(guilds) == 0:
raise CommandError("No guilds with the specified name found.")
if len(guilds) > 1:
raise CommandError("Multiple guilds with the specified name found.")
guild = list(bot.client.guilds)[0]
# Delete the previous PlayMode, if it exists
if bot.music_data[guild] is not None:
bot.music_data[guild].delete()
# Create the new PlayMode
if mode_name == "playlist":
bot.music_data[guild] = Playlist()
elif mode_name == "pool":
bot.music_data[guild] = Pool()
elif mode_name == "layers":
bot.music_data[guild] = Layers()
else:
raise CommandError("Unknown PlayMode specified.")
return {}
_event_name = "_legacy_playmode"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler(PlaymodeNH.message_type, PlaymodeNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_playmode_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
guild_name, mode_name = args.match(r"(?:\[(.+)])?\s*(\S+)\s*")
await self.interface.net_request(Request(PlaymodeNH.message_type, {"mode_name": mode_name,
"guild_name": guild_name}),
"discord")
await self.interface.call_herald_action("discord", self._event_name, {
"guild_name": guild_name,
"mode_name": mode_name
})
await data.reply(f"🔃 Impostata la modalità di riproduzione a: [c]{mode_name}[/c].")

82
royalnet/packs/royal/commands/queue.py
View file

@ -2,46 +2,8 @@ import typing
import pickle
import discord
from royalnet.commands import *
from royalnet.utils import NetworkHandler, numberemojiformat
from royalnet.utils import numberemojiformat
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class QueueNH(NetworkHandler):
message_type = "music_queue"
@classmethod
async def discord(cls, bot: "DiscordBot", data: dict):
# Find the matching guild
if data["guild_name"]:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(data["guild_name"])
else:
guilds = bot.client.guilds
if len(guilds) == 0:
raise CommandError("No guilds with the specified name found.")
if len(guilds) > 1:
raise CommandError("Multiple guilds with the specified name found.")
guild = list(bot.client.guilds)[0]
# Check if the guild has a PlayMode
playmode = bot.music_data.get(guild)
if not playmode:
return ResponseSuccess({
"type": None
})
try:
queue = playmode.queue_preview()
except NotImplementedError:
return ResponseSuccess({
"type": playmode.__class__.__name__
})
return ResponseSuccess({
"type": playmode.__class__.__name__,
"queue":
{
"strings": [str(dfile.info) for dfile in queue],
"pickled_embeds": str(pickle.dumps([dfile.info.to_discord_embed() for dfile in queue]))
}
})
class QueueCommand(Command):
@ -53,13 +15,49 @@ class QueueCommand(Command):
syntax = "[ [guild] ]"
@staticmethod
async def _legacy_queue_handler(bot: "DiscordBot", guild_name: typing.Optional[str]):
# Find the matching guild
if guild_name:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(guild_name)
else:
guilds = bot.client.guilds
if len(guilds) == 0:
raise CommandError("No guilds with the specified name found.")
if len(guilds) > 1:
raise CommandError("Multiple guilds with the specified name found.")
guild = list(bot.client.guilds)[0]
# Check if the guild has a PlayMode
playmode = bot.music_data.get(guild)
if not playmode:
return {
"type": None
}
try:
queue = playmode.queue_preview()
except NotImplementedError:
return {
"type": playmode.__class__.__name__
}
return {
"type": playmode.__class__.__name__,
"queue":
{
"strings": [str(dfile.info) for dfile in queue],
"pickled_embeds": str(pickle.dumps([dfile.info.to_discord_embed() for dfile in queue]))
}
}
_event_name = "_legacy_queue"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler(QueueNH.message_type, QueueNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_queue_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
guild, = args.match(r"(?:\[(.+)])?")
response = await self.interface.net_request(Request(QueueNH.message_type, {"guild_name": guild}), "discord")
guild_name, = args.match(r"(?:\[(.+)])?")
response = await self.interface.call_herald_action("discord", self._event_name, {"guild_name": guild_name})
if response["type"] is None:
await data.reply("ℹ️ Non c'è nessuna coda di riproduzione attiva al momento.")
return

41
royalnet/packs/royal/commands/skip.py
View file

@ -1,19 +1,23 @@
import typing
import discord
from royalnet.commands import *
from royalnet.utils import NetworkHandler
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class SkipNH(NetworkHandler):
message_type = "music_skip"
class SkipCommand(Command):
name: str = "skip"
@classmethod
async def discord(cls, bot: "DiscordBot", data: dict):
aliases = ["s", "next", "n"]
description: str = "Salta la canzone attualmente in riproduzione in chat vocale."
syntax: str = "[ [guild] ]"
@staticmethod
async def _legacy_skip_handler(bot: "DiscordBot", guild_name: typing.Optional[str]):
# Find the matching guild
if data["guild_name"]:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(data["guild_name"])
if guild_name:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(guild_name)
else:
guilds = bot.client.guilds
if len(guilds) == 0:
@ -27,23 +31,18 @@ class SkipNH(NetworkHandler):
raise CommandError("Nothing to skip")
# noinspection PyProtectedMember
voice_client._player.stop()
return ResponseSuccess()
return {}
class SkipCommand(Command):
name: str = "skip"
aliases = ["s", "next", "n"]
description: str = "Salta la canzone attualmente in riproduzione in chat vocale."
syntax: str = "[ [guild] ]"
_event_name = "_legacy_skip"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler(SkipNH.message_type, SkipNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_skip_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
guild, = args.match(r"(?:\[(.+)])?")
await self.interface.net_request(Request(SkipNH.message_type, {"guild_name": guild}), "discord")
guild_name, = args.match(r"(?:\[(.+)])?")
await self.interface.call_herald_action("discord", self._event_name, {
"guild_name": guild_name
})
await data.reply(f"⏩ Richiesto lo skip della canzone attuale.")

4
royalnet/packs/royal/commands/smecds.py
View file

@ -60,7 +60,9 @@ class SmecdsCommand(Command):
"dei Sims", "di Sid", "delle Skullgirls", "di Sonic", "di Spiral (Knights)", "di Spore", "di Starbound",
"di SimCity", "di Sensei", "di Ssssssssssssss... Boom! E' esploso il dizionario", "della scala",
"di Sakura", "di Suzie", "di Shinji", "del senpai", "del support", "di Superman", "di Sekiro",
"dello Slime God", "del salassato", "della salsa"]
"dello Slime God", "del salassato", "della salsa", "di Senjougahara", "di Sugar", "della Stampa",
"della Stampante"]
SMECDS = "🤔 Secondo me, è colpa {ds}."
async def run(self, args: CommandArgs, data: CommandData) -> None:

47
royalnet/packs/royal/commands/soundcloud.py
View file

@ -3,21 +3,25 @@ import pickle
import datetime
import discord
from royalnet.commands import *
from royalnet.utils import NetworkHandler, asyncify
from royalnet.utils import asyncify
from royalnet.audio import YtdlDiscord
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class SoundcloudNH(NetworkHandler):
message_type = "music_soundcloud"
class SoundcloudCommand(Command):
name: str = "soundcloud"
@classmethod
async def discord(cls, bot: "DiscordBot", data: dict):
"""Handle a play Royalnet request. That is, add audio to a PlayMode."""
aliases = ["sc"]
description: str = "Cerca una canzone su Soundcloud e la aggiunge alla coda della chat vocale."
syntax = "[ [guild] ] (url)"
@staticmethod
async def _legacy_soundcloud_handler(bot: "DiscordBot", guild_name: typing.Optional[str], search: str):
# Find the matching guild
if data["guild_name"]:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(data["guild_name"])
if guild_name:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(guild_name)
else:
guilds = bot.client.guilds
if len(guilds) == 0:
@ -34,40 +38,33 @@ class SoundcloudNH(NetworkHandler):
"outtmpl": f"./downloads/{datetime.datetime.now().timestamp()}_%(title)s.%(ext)s"
}
# Start downloading
dfiles: typing.List[YtdlDiscord] = await asyncify(YtdlDiscord.create_from_url, f'scsearch:{data["search"]}',
dfiles: typing.List[YtdlDiscord] = await asyncify(YtdlDiscord.create_from_url, f'scsearch:{search}',
**ytdl_args)
await bot.add_to_music_data(dfiles, guild)
# Create response dictionary
response = {
return {
"videos": [{
"title": dfile.info.title,
"discord_embed_pickle": str(pickle.dumps(dfile.info.to_discord_embed()))
} for dfile in dfiles]
}
return ResponseSuccess(response)
class SoundcloudCommand(Command):
name: str = "soundcloud"
aliases = ["sc"]
description: str = "Cerca una canzone su Soundcloud e la aggiunge alla coda della chat vocale."
syntax = "[ [guild] ] (url)"
_event_name = "_legacy_soundcloud"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler(SoundcloudNH.message_type, SoundcloudNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_soundcloud_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
guild_name, search = args.match(r"(?:\[(.+)])?\s*<?(.+)>?")
if search.startswith("http://") or search.startswith("https://"):
raise CommandError("SoundcloudCommand only accepts search queries, and you've sent an URL.\n"
"If you want to add a song from an url, please use PlayCommand!")
response = await self.interface.net_request(Request("music_soundcloud", {"search": search,
"guild_name": guild_name}),
"discord")
response = await self.interface.call_herald_action("discord", self._event_name, {
"guild_name": guild_name,
"search": search
})
if len(response["videos"]) == 0:
raise CommandError(f"Il video non può essere scaricato a causa di un blocco imposto da Soundcloud.")
for video in response["videos"]:

37
royalnet/packs/royal/commands/summon.py
View file

@ -1,24 +1,7 @@
import typing
import discord
from royalnet.commands import *
from royalnet.utils import NetworkHandler
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class SummonNH(NetworkHandler):
message_type = "music_summon"
@classmethod
async def discord(cls, bot: "DiscordBot", data: dict):
"""Handle a summon Royalnet request.
That is, join a voice channel, or move to a different one if that is not possible."""
channels = bot.client.find_channel_by_name(data["channel_name"])
channel = channels[0]
if not isinstance(channel, discord.VoiceChannel):
raise CommandError("Channel is not a voice channel")
bot.loop.create_task(bot.client.vc_connect_or_move(channel))
return ResponseSuccess()
class SummonCommand(Command):
@ -30,9 +13,23 @@ class SummonCommand(Command):
syntax: str = "[nomecanale]"
@staticmethod
async def _legacy_summon_handler(bot: "DiscordBot", channel_name: str):
"""Handle a summon Royalnet request.
That is, join a voice channel, or move to a different one if that is not possible."""
channels = bot.client.find_channel_by_name(channel_name)
channel = channels[0]
if not isinstance(channel, discord.VoiceChannel):
raise CommandError("Channel is not a voice channel")
bot.loop.create_task(bot.client.vc_connect_or_move(channel))
return {}
_event_name = "_legacy_summon"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler(SummonNH.message_type, SummonNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_summon_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
if self.interface.name == "discord":
@ -72,5 +69,7 @@ class SummonCommand(Command):
await data.reply(f"✅ Mi sono connesso in [c]#{channel.name}[/c].")
else:
channel_name: str = args[0].lstrip("#")
await self.interface.net_request(Request(SummonNH.message_type, {"channel_name": channel_name}), "discord")
response = await self.interface.call_herald_action("discord", self._event_name, {
"channel_name": channel_name
})
await data.reply(f"✅ Mi sono connesso in [c]#{channel_name}[/c].")

7
royalnet/packs/royal/commands/videochannel.py
View file

@ -40,8 +40,11 @@ class VideochannelCommand(Command):
raise CommandError("Non sei connesso a nessun canale vocale.")
channel = voice.channel
if author.is_on_mobile():
await data.reply(f"📹 Per entrare in modalità video, clicca qui: <https://discordapp.com/channels/{channel.guild.id}/{channel.id}>\n[b]Attenzione: la modalità video non funziona su Discord per Android e iOS![/b]")
await data.reply(f"📹 Per entrare in modalità video, clicca qui:\n"
f"<https://discordapp.com/channels/{channel.guild.id}/{channel.id}>\n"
f"[b]Attenzione: la modalità video non funziona su Android e iOS![/b]")
return
await data.reply(f"📹 Per entrare in modalità video, clicca qui: <https://discordapp.com/channels/{channel.guild.id}/{channel.id}>")
await data.reply(f"📹 Per entrare in modalità video, clicca qui:\n"
f"<https://discordapp.com/channels/{channel.guild.id}/{channel.id}>")
else:
raise UnsupportedError(f"This command is not supported on {self.interface.name.capitalize()}.")

45
royalnet/packs/royal/commands/youtube.py
View file

@ -3,21 +3,25 @@ import pickle
import datetime
import discord
from royalnet.commands import *
from royalnet.utils import NetworkHandler, asyncify
from royalnet.utils import asyncify
from royalnet.audio import YtdlDiscord
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class YoutubeNH(NetworkHandler):
message_type = "music_youtube"
class YoutubeCommand(Command):
name: str = "youtube"
aliases = ["yt"]
description: str = "Cerca un video su YouTube e lo aggiunge alla coda della chat vocale."
syntax = "[ [guild] ] (url)"
@classmethod
async def discord(cls, bot: "DiscordBot", data: dict):
"""Handle a play Royalnet request. That is, add audio to a PlayMode."""
async def _legacy_youtube_handler(cls, bot: "DiscordBot", guild_name: typing.Optional[str], search: str):
# Find the matching guild
if data["guild_name"]:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(data["guild_name"])
if guild_name:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(guild_name)
else:
guilds = bot.client.guilds
if len(guilds) == 0:
@ -34,39 +38,32 @@ class YoutubeNH(NetworkHandler):
"outtmpl": f"./downloads/{datetime.datetime.now().timestamp()}_%(title)s.%(ext)s"
}
# Start downloading
dfiles: typing. List[YtdlDiscord] = await asyncify(YtdlDiscord.create_from_url, f'ytsearch:{data["search"]}', **ytdl_args)
dfiles: typing. List[YtdlDiscord] = await asyncify(YtdlDiscord.create_from_url, f'ytsearch:{search}', **ytdl_args)
await bot.add_to_music_data(dfiles, guild)
# Create response dictionary
response = {
return {
"videos": [{
"title": dfile.info.title,
"discord_embed_pickle": str(pickle.dumps(dfile.info.to_discord_embed()))
} for dfile in dfiles]
}
return ResponseSuccess(response)
class YoutubeCommand(Command):
name: str = "youtube"
aliases = ["yt"]
description: str = "Cerca un video su YouTube e lo aggiunge alla coda della chat vocale."
syntax = "[ [guild] ] (url)"
_event_name = "_legacy_youtube"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler(YoutubeNH.message_type, YoutubeNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_youtube_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
guild_name, search = args.match(r"(?:\[(.+)])?\s*<?(.+)>?")
if search.startswith("http://") or search.startswith("https://"):
raise CommandError("YoutubeCommand only accepts search queries, and you've sent an URL.\n"
"If you want to add a song from an url, please use PlayCommand!")
response = await self.interface.net_request(Request("music_youtube", {"search": search,
"guild_name": guild_name}),
"discord")
response = await self.interface.call_herald_action("discord", self._event_name, {
"guild_name": guild_name,
"search": search
})
if len(response["videos"]) == 0:
raise CommandError(f"Il video non può essere scaricato a causa di un blocco imposto da YouTube.")
for video in response["videos"]:

46
royalnet/packs/royal/commands/zawarudo.py
View file

@ -3,26 +3,26 @@ import discord
import asyncio
import datetime
from royalnet.commands import *
from royalnet.utils import NetworkHandler, asyncify
from royalnet.utils import asyncify
from royalnet.audio import YtdlDiscord
from royalnet.audio.playmodes import Playlist
from royalnet.bots import DiscordBot
from royalherald import Request, ResponseSuccess
class ZawarudoNH(NetworkHandler):
message_type = "music_zawarudo"
class ZawarudoCommand(Command):
name: str = "zawarudo"
ytdl_args = {
"format": "bestaudio/best",
"outtmpl": f"./downloads/%(title)s.%(ext)s"
}
aliases = ["theworld", "world"]
@classmethod
async def discord(cls, bot: "DiscordBot", data: dict):
description: str = "Ferma il tempo!"
syntax = "[ [guild] ] [1-9]"
@staticmethod
async def _legacy_zawarudo_handler(bot: "DiscordBot", guild_name: typing.Optional[str], time: int):
# Find the matching guild
if data["guild_name"]:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(data["guild_name"])
if guild_name:
guilds: typing.List[discord.Guild] = bot.client.find_guild_by_name(guild_name)
else:
guilds = bot.client.guilds
if len(guilds) == 0:
@ -57,7 +57,7 @@ class ZawarudoNH(NetworkHandler):
if member.bot:
continue
await member.edit(mute=True)
await asyncio.sleep(data["time"])
await asyncio.sleep(time)
await bot.add_to_music_data(zw_end, guild)
for member in affected:
member: typing.Union[discord.User, discord.Member]
@ -66,21 +66,14 @@ class ZawarudoNH(NetworkHandler):
await member.edit(mute=False)
bot.music_data[guild] = old_playlist
await bot.advance_music_data(guild)
return ResponseSuccess()
return {}
class ZawarudoCommand(Command):
name: str = "zawarudo"
aliases = ["theworld", "world"]
description: str = "Ferma il tempo!"
syntax = "[ [guild] ] [1-9]"
_event_name = "_legacy_zawarudo"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
interface.register_net_handler(ZawarudoNH.message_type, ZawarudoNH)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_zawarudo_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
guild_name, time = args.match(r"(?:\[(.+)])?\s*(.+)?")
@ -93,4 +86,7 @@ class ZawarudoCommand(Command):
if time > 10:
raise InvalidInputError("The World can stop time only for 10 seconds.")
await data.reply(f"🕒 ZA WARUDO! TOKI WO TOMARE!")
await self.interface.net_request(Request(ZawarudoNH.message_type, {"time": time, "guild_name": guild_name}), "discord")
await self.interface.call_herald_action("discord", self._event_name, {
"guild_name": guild_name,
"time": time
})

21
royalnet/utils/__init__.py
View file

@ -5,10 +5,23 @@ from .escaping import telegram_escape, discord_escape
from .safeformat import safeformat
from .classdictjanitor import cdj
from .sleepuntil import sleep_until
from .networkhandler import NetworkHandler
from .formatters import andformat, plusformat, fileformat, ytdldateformat, numberemojiformat, splitstring, ordinalformat
from .parse5etoolsentry import parse_5etools_entry
__all__ = ["asyncify", "safeformat", "cdj", "sleep_until", "plusformat",
"NetworkHandler", "andformat", "plusformat", "fileformat", "ytdldateformat", "numberemojiformat",
"telegram_escape", "discord_escape", "splitstring", "parse_5etools_entry", "ordinalformat"]
__all__ = [
"asyncify",
"safeformat",
"cdj",
"sleep_until",
"plusformat",
"andformat",
"plusformat",
"fileformat",
"ytdldateformat",
"numberemojiformat",
"telegram_escape",
"discord_escape",
"splitstring",
"parse_5etools_entry",
"ordinalformat",
]