ryg/royalnet
ryg/royalnet
1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-23 19:44:20 +00:00
Code Activity

Add summon command!

Browse source
This commit is contained in:
Steffo 2019-11-30 14:49:23 +01:00
parent a55e71d4f7
commit 4c82a7bf1b
6 changed files with 97 additions and 67 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
royalpack/commands/__init__.py
View file

@ -15,7 +15,7 @@ from .videochannel import VideochannelCommand
# from .playmode import PlaymodeCommand
# from .queue import QueueCommand
# from .skip import SkipCommand
# from .summon import SummonCommand
from .summon import SummonCommand
# from .youtube import YoutubeCommand
# from .soundcloud import SoundcloudCommand
# from .zawarudo import ZawarudoCommand
@ -43,7 +43,7 @@ available_commands = [
# PlaymodeCommand,
# QueueCommand,
# SkipCommand,
# SummonCommand,
SummonCommand,
# YoutubeCommand,
# SoundcloudCommand,
# ZawarudoCommand,

6
royalpack/commands/leagueoflegends.py
View file

@ -121,7 +121,7 @@ class LeagueoflegendsCommand(Command):
log.info(f"Updating...")
session = self.alchemy.Session()
log.info("")
lols = session.query(self.alchemy.LeagueOfLegends).all()
lols = session.query(self.alchemy.get(LeagueOfLegends)).all()
for lol in lols:
try:
await self._update(lol)
@ -161,7 +161,7 @@ class LeagueoflegendsCommand(Command):
summoner = self._riotwatcher.summoner.by_name(region=self.config["Lol"]["region"], summoner_name=name)
# Ensure the account isn't already connected to something else
leagueoflegends = await asyncify(
data.session.query(self.alchemy.LeagueOfLegends).filter_by(summoner_id=summoner["id"]).one_or_none)
data.session.query(self.alchemy.get(LeagueOfLegends)).filter_by(summoner_id=summoner["id"]).one_or_none)
if leagueoflegends:
raise CommandError(f"L'account {leagueoflegends} è già registrato su Royalnet.")
# Get rank information
@ -186,7 +186,7 @@ class LeagueoflegendsCommand(Command):
mastery = self._riotwatcher.champion_mastery.scores_by_summoner(region=self.config["Lol"]["region"],
encrypted_summoner_id=summoner["id"])
# Create database row
leagueoflegends = self.alchemy.LeagueOfLegends(
leagueoflegends = self.alchemy.get(LeagueOfLegends)(
region=self.config["Lol"]["region"],
user=author,
profile_icon_id=summoner["profileIconId"],

6
royalpack/commands/peertube.py
View file

@ -28,7 +28,7 @@ class PeertubeCommand(Command):
async def _get_json(self):
log.debug("Getting jsonfeed")
async with aiohttp.ClientSession() as session:
async with session.get(self.interface.cfg["Peertube"]["feed_url"]) as response:
async with session.get(self.config["Peertube"]["feed_url"]) as response:
log.debug("Parsing jsonfeed")
j = await response.json()
log.debug("Jsonfeed parsed successfully")
@ -37,7 +37,7 @@ class PeertubeCommand(Command):
async def _send(self, message):
client = self.interface.bot.client
await self.interface.bot.safe_api_call(client.send_message,
chat_id=self.interface.cfg["Telegram"]["main_group_id"],
chat_id=self.config["Telegram"]["main_group_id"],
text=escape(message),
parse_mode="HTML",
disable_webpage_preview=True)
@ -67,7 +67,7 @@ class PeertubeCommand(Command):
await self._send(f"🆕 Nuovo video su RoyalTube!\n"
f"[b]{video['title']}[/b]\n"
f"{video['url']}")
await asyncio.sleep(self.interface.cfg["Peertube"]["feed_update_timeout"])
await asyncio.sleep(self.config["Peertube"]["feed_update_timeout"])
async def run(self, args: CommandArgs, data: CommandData) -> None:
if self.interface.name != "telegram":

70
royalpack/commands/summon.py
View file

@ -1,7 +1,6 @@
import typing
import discord
from royalnet.commands import *
from royalnet.bots import DiscordBot
class SummonCommand(Command):
@ -13,65 +12,18 @@ 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)
if len(channels) < 1:
raise CommandError(f"Nessun canale vocale con il nome [c]{channel_name}[/c] trovato.")
channel = channels[0]
if not isinstance(channel, discord.VoiceChannel):
raise CommandError(f"Il canale [c]{channel}[/c] non è un canale vocale.")
bot.loop.create_task(bot.client.vc_connect_or_move(channel))
return {}
_event_name = "_legacy_summon"
def __init__(self, interface: CommandInterface):
super().__init__(interface)
if interface.name == "discord":
interface.register_herald_action(self._event_name, self._legacy_summon_handler)
async def run(self, args: CommandArgs, data: CommandData) -> None:
channel_name = args.joined()
if self.interface.name == "discord":
bot = self.interface.bot.client
message: discord.Message = data.message
channel_name: str = args.optional(0)
if channel_name:
guild: typing.Optional[discord.Guild] = message.guild
if guild is not None:
channels: typing.List[discord.abc.GuildChannel] = guild.channels
else:
channels = bot.get_all_channels()
matching_channels: typing.List[discord.VoiceChannel] = []
for channel in channels:
if isinstance(channel, discord.VoiceChannel):
if channel.name == channel_name:
matching_channels.append(channel)
if len(matching_channels) == 0:
await data.reply("⚠️ Non esiste alcun canale vocale con il nome specificato.")
return
elif len(matching_channels) > 1:
await data.reply("⚠️ Esiste più di un canale vocale con il nome specificato.")
return
channel = matching_channels[0]
else:
author: discord.Member = message.author
try:
voice: typing.Optional[discord.VoiceState] = author.voice
except AttributeError:
await data.reply("⚠️ Non puoi evocare il bot da una chat privata!")
return
if voice is None:
await data.reply("⚠️ Non sei connesso a nessun canale vocale!")
return
channel = voice.channel
await bot.vc_connect_or_move(channel)
await data.reply(f"✅ Mi sono connesso in [c]#{channel.name}[/c].")
guild_id = message.guild.id
user_id = message.author.id
else:
channel_name: str = args[0].lstrip("#")
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].")
guild_id = None
user_id = None
response = await self.interface.call_herald_event("discord", "discord_summon",
channel_name=channel_name, guild_id=guild_id, user_id=user_id)
if self.interface.name == "discord":
await data.reply(f"✅ Mi sono connesso in <#{response['channel']['id']}>!")
else:
await data.reply(f"✅ Mi sono connesso in [b]#{response['channel']['name']}[/b]!")

2
royalpack/events/__init__.py
View file

@ -1,9 +1,11 @@
# Imports go here!
from .discord_cv import DiscordCvEvent
from .discord_summon import DiscordSummonEvent
# Enter the commands of your Pack here!
available_events = [
DiscordCvEvent,
DiscordSummonEvent,
]
# noinspection PyUnreachableCode

76
royalpack/events/discord_summon.py Normal file
View file

@ -0,0 +1,76 @@
from typing import *
from royalnet.commands import *
from royalnet.serf.discord import *
from royalnet.serf.discord.errors import *
import discord
class DiscordSummonEvent(Event):
name = "discord_summon"
async def run(self, *,
channel_name: Optional[str] = None,
channel_id: Optional[int] = None,
guild_id: Optional[int] = None,
user_id: Optional[int] = None,
**kwargs):
if not isinstance(self.serf, DiscordSerf):
raise UnsupportedError()
# Find the guild
if guild_id is not None:
guild: Optional[discord.Guild] = self.serf.client.get_guild(guild_id)
else:
guild = None
# Find the member
if user_id is not None and guild is not None:
member: Optional[Union[discord.User, discord.Member]] = guild.get_member(user_id=user_id)
else:
member = None
# From channel id
if channel_id:
client: discord.Client = self.serf.client
channel = client.get_channel(channel_id)
# Find channel
elif channel_name is not None:
# Find accessible_to
accessible_to = [self.serf.client.user]
if member is not None:
accessible_to.append(member)
# Find the channel
channel: Optional["discord.VoiceChannel"] = self.serf.find_channel(channel_type=discord.VoiceChannel,
name=channel_name,
guild=guild,
accessible_to=accessible_to,
required_permissions=["connect", "speak"]
)
elif not (guild_id is None and user_id is None):
if member.voice is not None:
channel = member.voice.channel
else:
channel = None
else:
raise InvalidInputError("Non so a cosa devo connettermi! Specifica il nome di un canale, o entra tu stesso"
" in un canale vocale prima di invocare [c]summon[/c]!")
if channel is None:
raise InvalidInputError("Non ho trovato nessun canale in cui connettermi.")
# Create a new VoicePlayer
vp = VoicePlayer(loop=self.loop)
vp.playing = await PlayableYTDQueue.create()
# Connect to the channel
try:
await vp.connect(channel)
except GuildAlreadyConnectedError:
raise UserError("Il bot è già connesso in un canale vocale nel Server!\n"
"Spostalo manualmente, o disconnettilo e riinvoca [c]summon[/c]!")
# Add the created VoicePlayer to the list
self.serf.voice_players.append(vp)
# Reply to the request
return {
"channel": {
"id": channel.id,
"name": channel.name,
"guild": {
"name": channel.guild.name,
},
}
}