1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-23 19:44:20 +00:00
This commit is contained in:
Steffo 2019-04-23 20:08:37 +02:00
parent b6b8c63f88
commit 7583f1330f
11 changed files with 72 additions and 24 deletions

View file

@ -27,7 +27,10 @@ author = 'Stefano Pigozzi'
# Add any Sphinx extension module names here, as strings. They can be # Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones. # ones.
extensions = ["sphinx.ext.autodoc"] extensions = ["sphinx.ext.autodoc", "sphinx.ext.napoleon", "sphinx.ext.intersphinx"]
intersphinx_mapping = {'python': ('https://docs.python.org/3.7', None),
'discord': ('https://discordpy.readthedocs.io/en/latest/', None)}
# Add any paths that contain templates here, relative to this directory. # Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates'] templates_path = ['_templates']

View file

@ -7,3 +7,10 @@ royalnet.database
.. automodule:: royalnet.database .. automodule:: royalnet.database
:members: :members:
Tables
------------------------------------
.. automodule:: royalnet.database.tables
:members:

View file

@ -5,43 +5,50 @@ from .royalpcmaudio import RoyalPCMAudio
class PlayMode: class PlayMode:
"""The base class for a PlayMode, such as :py:class:`royalnet.audio.Playlist`. Inherit from this class if you want to create a custom PlayMode."""
def __init__(self): def __init__(self):
"""Create a new PlayMode and initialize the generator inside."""
self.now_playing: typing.Optional[RoyalPCMAudio] = None self.now_playing: typing.Optional[RoyalPCMAudio] = None
self.generator: typing.AsyncGenerator = self._generator() self.generator: typing.AsyncGenerator = self.generate_generator()
async def next(self): async def next(self):
return await self.generator.__anext__() return await self.generator.__anext__()
def videos_left(self): def videos_left(self) -> typing.Union[int, float]:
"""Return the number of videos left in the PlayMode.
Usually returns an :py:class:`int`, but may return :py:obj:`math.inf` if the PlayMode is infinite."""
raise NotImplementedError() raise NotImplementedError()
async def _generator(self): async def generate_generator(self):
"""Get the next RPA from the list and advance it.""" """Get the next :py:class:`royalnet.audio.RoyalPCMAudio` from the list and advance it."""
raise NotImplementedError() raise NotImplementedError()
# This is needed to make the coroutine an async generator # This is needed to make the coroutine an async generator
# noinspection PyUnreachableCode # noinspection PyUnreachableCode
yield NotImplemented yield NotImplemented
def add(self, item): def add(self, item):
"""Add a new RPA to the PlayMode.""" """Add a new :py:class:`royalnet.audio.RoyalPCMAudio` to the PlayMode."""
raise NotImplementedError() raise NotImplementedError()
def delete(self): def delete(self):
"""Delete all RPAs contained inside this PlayMode.""" """Delete all :py:class:`royalnet.audio.RoyalPCMAudio` contained inside this PlayMode."""
raise NotImplementedError()
class Playlist(PlayMode): class Playlist(PlayMode):
"""A video list. RPAs played are removed from the list.""" """A video list. :py:class:`royalnet.audio.RoyalPCMAudio` played are removed from the list."""
def __init__(self, starting_list: typing.List[RoyalPCMAudio] = None): def __init__(self, starting_list: typing.List[RoyalPCMAudio] = None):
super().__init__() super().__init__()
if starting_list is None: if starting_list is None:
starting_list = [] starting_list = []
self.list: typing.List[RoyalPCMAudio] = starting_list self.list: typing.List[RoyalPCMAudio] = starting_list
def videos_left(self): def videos_left(self) -> typing.Union[int, float]:
return len(self.list) return len(self.list)
async def _generator(self): async def generate_generator(self):
while True: while True:
try: try:
next_video = self.list.pop(0) next_video = self.list.pop(0)
@ -63,7 +70,7 @@ class Playlist(PlayMode):
class Pool(PlayMode): class Pool(PlayMode):
"""A RPA pool. RPAs played are played back in random order, and they are kept in the pool.""" """A :py:class:`royalnet.audio.RoyalPCMAudio` pool. :py:class:`royalnet.audio.RoyalPCMAudio` are selected in random order and are not repeated until every song has been played at least once."""
def __init__(self, starting_pool: typing.List[RoyalPCMAudio] = None): def __init__(self, starting_pool: typing.List[RoyalPCMAudio] = None):
super().__init__() super().__init__()
if starting_pool is None: if starting_pool is None:
@ -71,10 +78,10 @@ class Pool(PlayMode):
self.pool: typing.List[RoyalPCMAudio] = starting_pool self.pool: typing.List[RoyalPCMAudio] = starting_pool
self._pool_copy: typing.List[RoyalPCMAudio] = [] self._pool_copy: typing.List[RoyalPCMAudio] = []
def videos_left(self): def videos_left(self) -> typing.Union[int, float]:
return math.inf return math.inf
async def _generator(self): async def generate_generator(self):
while True: while True:
if not self.pool: if not self.pool:
self.now_playing = None self.now_playing = None

View file

@ -5,17 +5,29 @@ from .royalpcmfile import RoyalPCMFile
class RoyalPCMAudio(AudioSource): class RoyalPCMAudio(AudioSource):
"""A discord-compatible :py:class:`discord.AudioSource` that keeps data in a file instead of in memory."""
def __init__(self, rpf: "RoyalPCMFile"): def __init__(self, rpf: "RoyalPCMFile"):
"""Create a :py:class:`discord.audio.RoyalPCMAudio` from a :ref:`discord.audio.RoyalPCMFile`. Not recommended, use """
self.rpf: "RoyalPCMFile" = rpf self.rpf: "RoyalPCMFile" = rpf
self._file = open(self.rpf.audio_filename, "rb") self._file = open(self.rpf.audio_filename, "rb")
@staticmethod @staticmethod
def create_from_url(url: str) -> typing.List["RoyalPCMAudio"]: def create_from_url(url: str) -> typing.List["RoyalPCMAudio"]:
"""Download a file with youtube_dl and create a list of :py:class:`discord.audio.RoyalPCMAudio`.
Parameters:
url: The url of the file to download."""
rpf_list = RoyalPCMFile.create_from_url(url) rpf_list = RoyalPCMFile.create_from_url(url)
return [RoyalPCMAudio(rpf) for rpf in rpf_list] return [RoyalPCMAudio(rpf) for rpf in rpf_list]
@staticmethod @staticmethod
def create_from_ytsearch(search: str, amount: int = 1) -> typing.List["RoyalPCMAudio"]: def create_from_ytsearch(search: str, amount: int = 1) -> typing.List["RoyalPCMAudio"]:
"""Search a string on YouTube and download the first ``amount`` number of videos, then download those with youtube_dl and create a list of :py:class:`discord.audio.RoyalPCMAudio`.
Parameters:
search: The string to search on YouTube.
amount: The number of videos to download."""
rpf_list = RoyalPCMFile.create_from_ytsearch(search, amount) rpf_list = RoyalPCMFile.create_from_ytsearch(search, amount)
return [RoyalPCMAudio(rpf) for rpf in rpf_list] return [RoyalPCMAudio(rpf) for rpf in rpf_list]
@ -36,6 +48,7 @@ class RoyalPCMAudio(AudioSource):
return data return data
def delete(self): def delete(self):
"""Permanently delete the downloaded file."""
self._file.close() self._file.close()
self.rpf.delete_audio_file() self.rpf.delete_audio_file()

View file

@ -46,13 +46,17 @@ class RoyalPCMFile(YtdlFile):
return f"<RoyalPCMFile {self.audio_filename}>" return f"<RoyalPCMFile {self.audio_filename}>"
@staticmethod @staticmethod
def create_from_url(url, **ytdl_args) -> typing.List["RoyalPCMFile"]: def create_from_url(url: str, **ytdl_args) -> typing.List["RoyalPCMFile"]:
"""Download a file with youtube_dl and create a list of :py:class:`discord.audio.RoyalPCMFile`.
Parameters:
url: The url of the file to download."""
info_list = YtdlInfo.create_from_url(url) info_list = YtdlInfo.create_from_url(url)
return [RoyalPCMFile(info, **ytdl_args) for info in info_list] return [RoyalPCMFile(info, **ytdl_args) for info in info_list]
@staticmethod @staticmethod
def create_from_ytsearch(search: str, amount: int = 1, **ytdl_args) -> typing.List["RoyalPCMFile"]: def create_from_ytsearch(search: str, amount: int = 1, **ytdl_args) -> typing.List["RoyalPCMFile"]:
"""Search a string on YouTube and download the first amount videos found.""" """Search a string on YouTube and download the first ``amount`` number of videos, then download those with youtube_dl and create a list of :py:class:`discord.audio.RoyalPCMFile`."""
url = f"ytsearch{amount}:{search}" url = f"ytsearch{amount}:{search}"
info_list = YtdlInfo.create_from_url(url) info_list = YtdlInfo.create_from_url(url)
return [RoyalPCMFile(info, **ytdl_args) for info in info_list] return [RoyalPCMFile(info, **ytdl_args) for info in info_list]

View file

@ -1,4 +1,5 @@
from .telegram import TelegramBot, TelegramConfig from .telegram import TelegramBot, TelegramConfig
from .discord import DiscordBot, DiscordConfig from .discord import DiscordBot, DiscordConfig
from .generic import GenericBot
__all__ = ["TelegramBot", "TelegramConfig", "DiscordBot", "DiscordConfig"] __all__ = ["TelegramBot", "TelegramConfig", "DiscordBot", "DiscordConfig", "GenericBot"]

View file

@ -19,11 +19,14 @@ if not discord.opus.is_loaded():
class DiscordConfig: class DiscordConfig:
"""The specific configuration to be used for :ref:`royalnet.database.DiscordBot`."""
def __init__(self, token: str): def __init__(self, token: str):
self.token = token self.token = token
class DiscordBot(GenericBot): class DiscordBot(GenericBot):
"""A bot that connects to `Discord <https://discordapp.com/>`_."""
interface_name = "discord" interface_name = "discord"
def _init_voice(self): def _init_voice(self):

View file

@ -12,7 +12,7 @@ log = logging.getLogger(__name__)
class GenericBot: class GenericBot:
"""A generic bot class, to be used as base for the other more specific classes, such as TelegramBot and DiscordBot.""" """A generic bot class, to be used as base for the other more specific classes, such as :ref:`royalnet.bots.TelegramBot` and :ref:`royalnet.bots.DiscordBot`."""
interface_name = NotImplemented interface_name = NotImplemented
def _init_commands(self, def _init_commands(self,

View file

@ -13,16 +13,14 @@ loop = asyncio.get_event_loop()
log = _logging.getLogger(__name__) log = _logging.getLogger(__name__)
async def todo(message: Message):
log.warning(f"Skipped {message} because handling isn't supported yet.")
class TelegramConfig: class TelegramConfig:
"""The specific configuration to be used for :ref:`royalnet.database.TelegramBot`."""
def __init__(self, token: str): def __init__(self, token: str):
self.token: str = token self.token: str = token
class TelegramBot(GenericBot): class TelegramBot(GenericBot):
"""A bot that connects to `Telegram <https://telegram.org/>`_."""
interface_name = "telegram" interface_name = "telegram"
def _init_client(self): def _init_client(self):

View file

@ -12,7 +12,15 @@ loop = asyncio.get_event_loop()
class Alchemy: class Alchemy:
def __init__(self, database_uri: str, tables: typing.Optional[typing.Set] = None): """A wrapper around SQLAlchemy declarative that allows to use multiple databases at once while maintaining a single table-class for both of them."""
def __init__(self, database_uri: str, tables: typing.Set):
"""Create a new Alchemy object.
Args:
database_uri: The uri of the database, as described at https://docs.sqlalchemy.org/en/13/core/engines.html .
tables: The set of tables to be created and used in the selected database. Check the tables submodule for more details.
"""
if database_uri.startswith("sqlite"): if database_uri.startswith("sqlite"):
raise NotImplementedError("Support for sqlite databases is currently missing") raise NotImplementedError("Support for sqlite databases is currently missing")
self.engine = create_engine(database_uri) self.engine = create_engine(database_uri)
@ -20,7 +28,7 @@ class Alchemy:
self.Session = sessionmaker(bind=self.engine) self.Session = sessionmaker(bind=self.engine)
self._create_tables(tables) self._create_tables(tables)
def _create_tables(self, tables: typing.Optional[typing.List]): def _create_tables(self, tables: typing.Set):
for table in tables: for table in tables:
name = table.__name__ name = table.__name__
try: try:
@ -34,7 +42,8 @@ class Alchemy:
self.Base.metadata.create_all() self.Base.metadata.create_all()
@contextmanager @contextmanager
async def session_cm(self): def session_cm(self):
"""Use Alchemy as a context manager (to be used in with statements)."""
session = self.Session() session = self.Session()
try: try:
yield session yield session
@ -46,6 +55,7 @@ class Alchemy:
@asynccontextmanager @asynccontextmanager
async def session_acm(self): async def session_acm(self):
"""Use Alchemy as a asyncronous context manager (to be used in async with statements)."""
session = await asyncify(self.Session) session = await asyncify(self.Session)
try: try:
yield session yield session

View file

@ -2,6 +2,8 @@ import typing
class DatabaseConfig: class DatabaseConfig:
"""The configuration to be used for the :ref:`royalnet.database.Alchemy` component of :ref:`royalnet.bots.GenericBot`."""
def __init__(self, def __init__(self,
database_uri: str, database_uri: str,
master_table: typing.Type, master_table: typing.Type,