Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

docs: add docstring to functions #117

Merged
merged 1 commit into from
Sep 2, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 27 additions & 9 deletions database/seeds/owner_table_seeder.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,14 +9,32 @@
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))


class OwnerTableSeeder(Seeder):
def run(self):
"""Run the database seeds."""
def create_owner_if_does_not_exist(owner_id: int) -> None:
"""
Checks if an owner with a specified ``owner_id`` exists in the database. If it's not, it creates it using
:func:`create_owner`.

:param owner_id: int: The ``id`` of the owner to be checked and saved
"""
if Admin.where('admin_user_id', owner_id).first():
return

create_owner(owner_id=owner_id)


if Admin.where('admin_user_id', OWNER_USER_ID_INT).first():
return
def create_owner(owner_id: int) -> None:
"""
Creates an owner with a specified ``owner_id`` and stores in the database.

Admin.create({
'admin_user_id': OWNER_USER_ID_INT,
'is_owner': True,
})
:param owner_id: int: The `id` of the owner to be saved
"""
Admin.create({
'admin_user_id': owner_id,
'is_owner': True,
})


class OwnerTableSeeder(Seeder):
@staticmethod
def run():
create_owner_if_does_not_exist(owner_id=OWNER_USER_ID_INT)
78 changes: 69 additions & 9 deletions modules/admin.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,17 +13,36 @@


def parse_and_normalize_user_id(message: str) -> int:
"""
Parses, converts and returns the user ``id`` of `/(add/del)admin` commands.

The `message` is expected to look like ``/addadmin <user_id>``.

:param message: str: The ``/(add/del)admin`` command containing a user ``id``
:return: int: The normalized user's ``id``
"""
return int(message.partition(' ')[2])


def add_admin_if_user_is_owner(update: Update, _context: CallbackContext):
def add_admin_if_user_is_owner(update: Update, _context: CallbackContext) -> None:
"""
Checks if the user who sent the message is an owner of the bot. If so, calls :func:`add_admin`.

:param update: Update: The ``update`` object
:param _context: CallbackContext: Unused
"""
if not is_admin_owner(get_effective_user_id(update)):
return

add_admin(update)


def add_admin(update: Update):
def add_admin(update: Update) -> None:
"""
Adds a user ``id`` to the ``admins`` table. If succeeds, sends a success message.

:param update: Update: The ``update`` object
"""
admin_id_to_add = parse_and_normalize_user_id(get_message_text(update))

admin = Admin()
Expand All @@ -34,14 +53,25 @@ def add_admin(update: Update):
update.message.reply_text(f"User {admin_id_to_add} has been added as admins.")


def del_admin_if_user_is_owner(update: Update, _context: CallbackContext):
def del_admin_if_user_is_owner(update: Update, _context: CallbackContext) -> None:
"""
Checks if the user who sent the message is an owner of the bot. If so, call :func:`del_admin`.

:param update: Update: The ``update`` object
:param _context: CallbackContext: Unused
"""
if not is_admin_owner(get_effective_user_id(update)):
return

del_admin(update)


def del_admin(update: Update):
def del_admin(update: Update) -> None:
"""
Deletes a user ``id`` from the ``admins`` table. Then sends a message accordingly.

:param update: Update: The ``update`` object
"""
# TODO: Check if the value is of type `int`
admin_id_to_delete = parse_and_normalize_user_id(get_message_text(update))

Expand All @@ -50,17 +80,32 @@ def del_admin(update: Update):

update.message.reply_text(f"User {admin_id_to_delete} is no longer an admin")
else:
update.message.reply_text(f"User {admin_id_to_delete} is not admin")
update.message.reply_text(f"User {admin_id_to_delete} is not an admin")


def show_stats_if_user_is_admin(update: Update, _context: CallbackContext) -> None:
"""
Checks if the user is an admin. If they are, it calls :func:`show_stats` to display the stats of the bot.

def show_stats_if_user_is_admin(update: Update, _context: CallbackContext):
:param update: Update: The ``update`` object
:param _context: CallbackContext: Unused
"""
if not is_user_admin(get_effective_user_id(update)):
return

show_stats(update)


def show_stats(update: Update):
def show_stats(update: Update) -> None:
"""
Displays a summary about how the bot is being used:
- The number of users using this bot
- The number of English and Persian users
- The number & size of the files on the disk
- How much disk space is occupied.

:param update: Update: The ``update`` object
"""
persian_users = User.all().where('language', 'fa')
english_users = User.all().where('language', 'en')

Expand All @@ -84,14 +129,25 @@ def show_stats(update: Update):
)


def list_users_if_user_is_admin(update: Update, _context: CallbackContext):
def list_users_if_user_is_admin(update: Update, _context: CallbackContext) -> None:
"""
Checks if the user who sent the message is an owner of the bot. If so, calls :func:`list_users`.

:param update: Update: The ``update`` object
:param _context: CallbackContext: Unused
"""
if not is_user_admin(get_effective_user_id(update)):
return

list_users(update)


def list_users(update: Update):
def list_users(update: Update) -> None:
"""
Displays a list of all users in the form of ``user_id:username``.

:param update: Update: The ``update`` object
"""
users = User.all()

reply_message = ''
Expand All @@ -113,6 +169,10 @@ def send_to_all():
class AdminModule:
@staticmethod
def register():
"""
Registers all the handlers that are defined in ``Admin`` module, so that they can be used to respond to messages
sent to the bot.
"""
add_handler(CommandHandler('addadmin', add_admin_if_user_is_owner))
add_handler(CommandHandler('deladmin', del_admin_if_user_is_owner))
add_handler(CommandHandler('stats', show_stats_if_user_is_admin))
Expand Down
69 changes: 52 additions & 17 deletions modules/bitrate_changer.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,25 +14,70 @@
reply_default_message, reset_user_data_context, set_current_module, t


def parse_bitrate_number(input_string: str) -> int | None:
def parse_bitrate_number(message: str) -> int | None:
"""
Parses, converts and returns a bitrate in a text.

The ``message`` is expected to look like `320 kb/s`.

:param message: str: A message text containing a bitrate
:return: int | None: The extracted bitrate
"""
number_pattern = r'^\d+'

matches = re.findall(number_pattern, input_string)
matches = re.findall(number_pattern, message)

if matches:
return int(matches[0])
else:
return None


def convert_bitrate(input_path: str, output_bitrate: int, output_path: str):
def convert_bitrate(input_path: str, output_bitrate: int, output_path: str) -> None:
"""
Creates a new audio file with a specified bitrate.

:param input_path: str: The path of the input file
:param output_bitrate: int: The bitrate of the output file
:param output_path: str: The output path of the converted file
"""
os.system(
f"ffmpeg -v debug -i \"{input_path}\" -c:a libmp3lame \
-b:a {output_bitrate}k -ac 2 -ar 44100 -vn \"{output_path}\""
)


def show_bitrate_changer_keyboard(update: Update, context: CallbackContext) -> None:
"""
sets the current module to `Module.BITRATE_CHANGER`, and displays a keyboard with buttons for each bitrate option.

:param update: Update: The `update` object
:param context: CallbackContext: The `context` object
"""
user_data = get_user_data(context)

lang = get_user_language_or_fallback(user_data)
bitrate_selector_keyboard = generate_bitrate_selector_keyboard(lang)

set_current_module(user_data, Module.BITRATE_CHANGER)

update.message.reply_text(
f"{t(lp.BITRATE_CHANGER_HELP, lang)}\n",
reply_markup=bitrate_selector_keyboard
)


def change_bitrate(update: Update, context: CallbackContext) -> None:
"""
Handles the change bitrate functionality.

This is the main function of the ``bitrate_changer`` module that accepts the desired bitrate, generates a new file,
and sends it to the user. It sends a default message if user has not started the bot.

:param update: Update: The ``update`` object
:param update: Update: The ``context`` object
:raises TelegramError | BaseException
"""
user_data = get_user_data(context)
message = get_message(update)
lang = get_user_language_or_fallback(user_data)
Expand Down Expand Up @@ -73,23 +118,13 @@ def change_bitrate(update: Update, context: CallbackContext) -> None:
reset_user_data_context(get_effective_user_id(update), user_data)


def show_bitrate_changer_keyboard(update: Update, context: CallbackContext) -> None:
user_data = get_user_data(context)

lang = get_user_language_or_fallback(user_data)
bitrate_selector_keyboard = generate_bitrate_selector_keyboard(lang)

set_current_module(user_data, Module.BITRATE_CHANGER)

update.message.reply_text(
f"{t(lp.BITRATE_CHANGER_HELP, lang)}\n",
reply_markup=bitrate_selector_keyboard
)


class BitrateChangerModule:
@staticmethod
def register():
"""
Registers all the handlers that are defined in ``BitrateChanger`` module, so that they can be used to respond to
messages sent to the bot.
"""
add_handler(MessageHandler(
Filters.regex(r'^(\d{3}\s{1}kb/s)$'),
change_bitrate)
Expand Down
Loading