null

Разработка Telegram бот-а или теперь не нужно напоминать про time-management вручную

10 июня 2023 0 Алексей Егошин

Оглавление

Часть 2. Интеграция с GitLab

 

Предыстория

Команды разработки, которые работают на аутсорсинге, обязаны учитывать потраченное на проект время, чтобы можно было выставить счет заказчику "не из воздуха". Принципиально можно выделить два пути решения этой проблемы:

  • с каждого члена команды генерируем фиксированное число часов, который уходят в отчет заказчику;
  • каждый член команды ведет учет потраченного времени в часах и именно оно выставляется в счет заказчику.

Очевидно, что первый путь для самих членов команды проще - "работай себе и ни о чем не думай". Однако этот путь реализуем только в том случае, если человек работает фиксированное количество часов в неделю и только на одном проекте. На практике же, по крайней мере в нашей компании, используется путь под номером два, так как люди часто переключаются между проектами.

Следовательно, возникает необходимость трекинга потраченных часов и сделать это, очевидно, может только сам работник. Но есть одно но. Никому не нравится вести time-management и учитывать потраченные часы. Скорее всего дело в банальной человеческой лени, которая присуща всем людям. Из-за этого менеджеры на проектах вынуждены периодически напоминать команде о необходимости трекинга часов.

Но мы же великие программисты-автоматизаторы! Давайте автоматизируем напоминания?

 

Выбор технологий

В принципе выбора большого нет. Хочется заиспользовать инструмент, для которого не потребуется внедрения дополнительных систем учета времени, напоминалок или чего-то подобного. С учетом этого требования есть только два варианта:

  1. учет времени мы ведем в отдельных тикетах на корпоративном Gitlab-е - можно сделать демона, который будет автоматически писать комментарии-напоминалки. При появлении новых комментариев Gitlab "из коробки" отправляет уведомления на привязанную почту;
  2. внутреннее общение внутри команды у нас организовано через чаты в Telegram - значит, можно сделать Telegram бота, который будет рассылать людям уведомления.

Выбор пал на второй вариант, так как в случае первого напоминание по сути происходит по почте, а почту "современная молодеж" читать не любит.

 

Функции бота

Попробуем формализовать список функций, которыми должен обладать бот:

  • подписка\отписка на напоминания;
  • напоминание утром и вечером каждого рабочего дня;
  • возможность настройки обращения от бота (чисто забавы ради).

Следовательно, нам понадобится где-то хранить информацию о подписавшихся пользователях, чтобы было возможно их уведомлять. По идее для этого подойдут любые базы данных, в том числе sqlite, но, кажется, для нашей задаче это overhead. Будем использовать хранение информации в простом json-файле. Заодно получим простую возможность редактирования информации о пользователях из консольного редактора.

Для реализации напоминаний по расписанию идеально подойдет crontab.

Ну и, конечно, самый главный вопрос - язык программирования. Задача простая, сделать хочется быстро - идеальный случай для ненавистного Python. Плюсом он есть "из коробки" в (вероятно во многих других в том числе) Ubuntu дистрибутиве Linux-а.

 

Арендуем ВМ

Автор статьи предпочитает использовать для подобных нересурсоемких задач дешевые VPS-сервера от компании RuVDS.

Очевидно, нам подойдет даже самый дешевый сервер. Конфигурация представлена на изображении, ОС можно выбрать из широкого списка, мы же будем использовать Ubuntu 20.04 LTS. Однако автор статьи принял решение выбрать сервер слегка подороже, так как планирует его использовать и для других личных целей, а 512 Мб RAM все же слишком мало в "современном мире веб-технологий". Итоговый вариант представлен ниже

 

Предварительная настройка ВМ

Нам потребуется дополнительно установить следующие пакеты из стандартного репозитория: tmux, python3.8-venv, git (на удивление чистые сервера RuVDS запускаются без git-а). Tmux будем использовать из личных предпочтений, python3.8-venv потребуется для установки питоно-пакетов. Ну и понятно дело, кладем свой публичный ssh-ключ и прописываем в ssh-конфиг метод подключения к только что купленному серверу.

 

Реализация бота

Вообще говоря Telegram-боты могут быть реализованы на любых языках, в том числе возможно создать простого бота в принципе без написания кода, но это не наш случай. Как уже упоминалось, в качестве ЯП был выбран ненавистный Python, просто потому, что - надо отдать ему должное - подобные задачи решать с использованием питона куда быстрее, чем с использованием той же Java. Плюс у нас исчезает необходимость сборки как таковой. 

Для обработки входящих сообщений от Telegram можно использовать два подхода: long polling и webhook. В принципе обе технологии простые и понятные, мы же выберем long polling просто по причине того, что его писать быстрее. У нас все равно не предполагается сотен пользователей, а значит и поллинга нам хватит. Также заиспользуем библиотеку python-telegram-bot, которая предоставляет программное API для супер-простого взаимодействия с Telegram Bot API.

Итого код самого бота получается таким:

from typing import List, Tuple, Callable, Optional

from telegram import Update, Bot
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters

from personal_notifier.dialogue import get_answer, \
    already_registered_message, \
    new_user_message, \
    bye_existed_user_message, \
    bye_not_found_user_message, nickname_updated_message, nickname_user_not_found_message
from personal_notifier.envs import TELEGRAM_BOT_TOKEN_ENV_VARIABLE, get_env
from personal_notifier.model.user import User
from personal_notifier.storage.json_file_storage import JsonFileStorage


def _update_to_user(
    update: Update
) -> User:
    return User(
        id=update.message.chat_id,
        name=update.message.chat.first_name,
        nickname=''
    )


def _search_author_user_template(
    update: Update,
    func: Optional[Callable[[bool, int, User, List[User], JsonFileStorage], None]]
) -> Tuple[bool, User]:
    author_user: User = _update_to_user(update)
    json_file_storage: JsonFileStorage = JsonFileStorage.default()
    all_users = json_file_storage.load_users()

    is_user_found: bool = False
    found_user: Optional[User] = None
    user_index: int = -1
    for i in range(len(all_users)):
        user = all_users[i]
        if user.id == author_user.id:
            is_user_found = True
            found_user = user
            user_index = i

    if is_user_found:
        author_user.nickname = found_user.nickname

    if func:
        func(is_user_found, user_index, author_user, all_users, json_file_storage)

    return is_user_found, author_user


async def _start_command_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    def _save_user_if_new(
            is_user_found: bool,
            user_index: int,
            author_user: User,
            all_users: List[User],
            json_file_storage: JsonFileStorage
    ) -> None:
        if not is_user_found:
            all_users.append(author_user)
            json_file_storage.save_users(all_users)

    (is_user_found, author_user) = _search_author_user_template(
        update=update,
        func=_save_user_if_new
    )

    if is_user_found:
        await update.message.reply_text(already_registered_message(author_user))
        return

    await update.message.reply_text(new_user_message(author_user))


async def _bye_command_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    def _remove_user_if_found(
        is_user_found: bool,
        user_index: int,
        author_user: User,
        all_users: List[User],
        json_file_storage: JsonFileStorage
    ) -> None:
        if is_user_found:
            del all_users[user_index]
            json_file_storage.save_users(all_users)

    (is_user_found, author_user) = _search_author_user_template(
        update=update,
        func=_remove_user_if_found
    )

    if is_user_found:
        await update.message.reply_text(bye_existed_user_message(author_user))
        return

    await update.message.reply_text(bye_not_found_user_message(author_user))


async def _nickname_command_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if len(context.args) != 1:
        await update.message.reply_text("Неправильное количество аргументов")
        return

    def _update_nickname_if_user_found(
        is_user_found: bool,
        user_index: int,
        author_user: User,
        all_users: List[User],
        json_file_storage: JsonFileStorage
    ) -> None:
        if is_user_found:
            all_users[user_index].nickname = context.args[0]
            json_file_storage.save_users(all_users)

    (is_user_found, author_user) = _search_author_user_template(
        update=update,
        func=_update_nickname_if_user_found
    )

    if is_user_found:
        await update.message.reply_text(nickname_updated_message(author_user))
        return

    await update.message.reply_text(nickname_user_not_found_message(author_user))


async def _help_command_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    help: str = "/start - подписаться на напоминания\n/bye - отписаться от напоминаний\n/nickname <nickname> - установить обращение"
    await update.message.reply_text(help)


async def _random_message_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    author_user: User = _update_to_user(update)
    all_users: List[User] = JsonFileStorage.default().load_users()

    is_user_found: bool = False
    for user in all_users:
        if user.id == author_user.id:
            is_user_found = True
            break

    if not is_user_found:
        await update.message.reply_text(text="Хто ты, путник? Напиши '/start', чтобы начать")
        return

    answer: str = get_answer(
        user=author_user,
        message=update.message.text
    )
    await update.message.reply_text(answer)


async def notify_about_time_management(
    user: User,
    message: str
) -> None:
    bot = Bot(get_env(TELEGRAM_BOT_TOKEN_ENV_VARIABLE))
    async with bot:
        await bot.send_message(
            text=message,
            chat_id=user.id,
        )


def start_personal_notifier_long_polling_bot() -> None:
    token: str = get_env(TELEGRAM_BOT_TOKEN_ENV_VARIABLE)

    application = Application.builder().token(token).build()
    application.add_handler(CommandHandler("start", _start_command_handler))
    application.add_handler(CommandHandler("bye", _bye_command_handler))
    application.add_handler(CommandHandler("help", _help_command_handler))
    application.add_handler(CommandHandler("nickname", _nickname_command_handler))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, _random_message_handler))

    application.run_polling()

Ключевые python-функции здесь start_personal_notifier_long_polling_bot - при помощи которой запускает приложение, периодически опрашивающее Telegram Bot API о новых сообщениях боту, и notify_about_time_management - при помощи этой функции обращаемся к Telegram Bot API для асинхронной отправке сообщений подписанным на напоминания пользователям.

Также нам потребуется точка входа в программу следующего вида:

import sys

from personal_notifier.envs import check_envs
from personal_notifier.notify import notify_all_users
from personal_notifier.telegram_bot import start_personal_notifier_long_polling_bot

USAGE_STR = "python main.py --mode notify m(or e) or python main.py --bot"


if __name__ == '__main__':
    if not check_envs():
        exit(1)

    if len(sys.argv) < 3 or len(sys.argv) > 4:
        print(f"Invalid params. Usage: {USAGE_STR}")
        exit(1)

    if sys.argv[1] != '--mode':
        print(f"Invalid params. Usage: {USAGE_STR}")
        exit(1)

    if sys.argv[2] != 'notify' and sys.argv[2] != 'bot':
        print(f"Invalid params. Usage: {USAGE_STR}")
        exit(1)

    mode = sys.argv[2]

    if mode == 'bot':
        if len(sys.argv) != 3:
            print(f"Invalid params. Usage: {USAGE_STR}")
            exit(1)
        start_personal_notifier_long_polling_bot()

    if mode == 'notify':
        if len(sys.argv) != 4:
            print(f"Invalid params. Usage: {USAGE_STR}")
            exit(1)

        notification_type = sys.argv[3]
        if notification_type == 'm':
            notify_all_users(is_morning=True)
        elif notification_type == 'e':
            notify_all_users(is_morning=False)
        else:
            print(f"Invalid params. Usage: {USAGE_STR}")
            exit(1)

Как можно заметить, точка входа принимает оргумент mode, в зависимости от которого либо запускается бот для получения сообщений, либо вызывается оберточная функция notify_all_users, которая рассылает всем подписанным пользователям уведомления.

 

Деплой и запуск

Скачиваем исходники бота при помощи git-а по HTTPS урлу и размещаем в файловой системе сервера. Осталось сделать:

  • создать бота через FatherBot и получить API Token;
  • запустить бота для получения сообщений;
  • настроить cron-расписание для отправки уведомлений.

Первый пункт делается супер просто - напишите FatherBot-у и далее получите понятные инструкции по созданию бота. Получив от него API Token сконфигурируем переменную окружения с токеном в .bashrc. Второой пункт с использованием tmux выглядит как простой запуск консольной программы:

python3 main.py --mode bot

Третий и последний пункт - настройка сron расписания. Используем команду crontab -e и вписываем следующий конфиг:

0 10 * * 0-4 bash -c "source /apps/personal-notifier/venv/bin/activate && source ~/.bashrc && python3 /apps/personal-notifier/main.py --mode notify m > /apps/last_morning_logs.txt 2>&1"
50 18 * * 0-4 bash -c "source /apps/personal-notifier/venv/bin/activate && source ~/.bashrc && python3 /apps/personal-notifier/main.py --mode notify e > /apps/last_evening_logs.txt 2>&1"

Таким образом, мы настроили напоминания на 10:00 каждый рабочий день и на 18:50 каждый рабочий день.

 

Что можно улучшить

Из планов на будущее можно выделить следующее:

  1. учет праздничных дней и отпусков;
  2. интеграция с Gitlab-ом для отмены напоминаний, если пользователь во время без нареканий трекает время;
  3. реализация фичи, которая бы позволила админу бота вручную триггерить дополнительное напоминание в любое время.

 

Исходный код доступен в публичном репозитории на GitHub-е: github.com/Roggired/personal-notifier

Сам бот: TuneITPersonalNotifierBot