Бот - для работы с репозиторием проекта в BitBucket.

Приветствую любителей ботов :-) Позвольте представить вашему вниманию бота для работы с репозиторием проекта в BitBucket.

Краткое описание моих скромных трудов.

Схема:

Пример сообщений в телеграм:

Сразу оговорюсь, я являюсь автоматизатором на Java. Начинал изучение с Python, поэтому и бота написал на нем, так как моя работа происходит на Java, мне захотелось вспомнить Python), поддерживать и углублять свои знания этого языка.

И вот мне пришла в голову идея создания бота. Который к слову на проекте используется в нашем командном телеграмм чате, состоящим из 9 человек. Я обдумал основные задачи для которых он был бы необходим, их три (по результатм опроса, позже, выяснилось этого достаточно, но можно расширять функционал по своему усмотрению и необходимости):

Информировать о новых пулл-реквестах
Напоминать о текущих, что бы их проверили и поставили апрув (подтверждение на мерж)
Информировать о комментах в реквесте, если их более 1

Задачи исходили в первую очередь из потребностей проекта. У нас бывало по долгу не проверялись реквесты. Их либо забывали проверять, либо не знали что реквест создан (у нас есть информирование по реквестам через корпоративную почту, но там бывают задержки, да и не всегда она читается).

Я понимаю, бота можно расширить до того, что он будет информировать о любых измен��ниях реквеста (аппрув, изменение названия, событие мержа реквеста), но тогда наш чат превратиться в чат бесконечного спама от бота, будет неудобно переписываться по рабочим вопросам между собой, поэтому я ограничился задачами, описанными, выше. Этого как оказалось вполне хватает, реквесты стали более быстро проверяться, комментироваться и отслеживаться, что не может не радовать).

Да бот оказался весьма кстати и как мне кажется полезен.

Технологии и библиотеки

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

Я хотел сделать его максимально понятным и удобным в использовании. Разумеется так как он работает с конфидециальным корпоративным ресурсом я не могу деплоить его в облаках или на сторонних серверах, он работает у меня локально, меня и всех это устраивает. Будьте осторожны, если захотите использовать этого бота, с вашими конфидециальными данными (конечно если таковые есть), используйте бота в защищенных корпоративных сетях либо локально.

Я выбрал популярную библиотеку Aiogram для работы с телеграмм. Примеров ее использования достаточно много можно найти в совбодном доступе, роликов на ютубе достаточно.

Работу с API BitBucket была использована библиотека Atlassian.

Пример команды, можно подробнее посмотреть на GitHub:

from aiogram import Bot, Dispatcher, executor, types
from aiogram.types import ParseMode, Message
from aiogram.utils.markdown import text, bold, code


@dp.message_handler(commands=['start', 'help'])
async def start_bot(message: types.Message):
    """
    Стартовая команда, для приветствия и инфо о боте
    """
    try:
        msg = text(code("Привет, это бот BitBucket\n") +
                    code("Для проекта:  ") + bold(project.upper() + "\n") +
                    code("Репозитория:  ") + bold(repository + "\n") +
                    code("Нажмите, просмотреть:   ") + bold("/requests") + "\n" +
                    code("Посмотреть курсы валют:  ") + bold("/currency"))

        await bot.send_message(message.from_user.id, msg, parse_mode=ParseMode.MARKDOWN, reply_markup=kb_client)
    except:
        await message.reply("ВАЖНО: Вам нужно добавиться к боту, все уведомления идут в ЛС!!!")


@dp.message_handler(commands=['requests'])
async def send_open_pull_requests(message: Message):
    """
    Команда для вывода Pull-requests в личку пользователю

    :param message: класс Message от которого можно использовать
                    разные методы отправки сообщений
    """
    size = bitBacketUtils.get_size_pull_requests()
    try:
        msg = text("? У вас есть - " + bold(str(size)) + " Pull requests в статусе OPEN:\n")
        await bot.send_message(message.from_user.id, msg, parse_mode=ParseMode.MARKDOWN)
    except:
        await message.reply("❗ Возникла ошибка, повторите команду! Или добавьтесь к боту!")

    try:
        if size == 0:
            msg = text("Pull requests в статусе OPEN нет")
            await bot.send_message(message.from_user.id, msg, parse_mode=ParseMode.MARKDOWN)
        else:
            for request in bitBacketUtils.get_open_pull_requests():
                mess = text("? Имя автора:  " + bold(request["Имя автора"]) + "\n"
                            "? Имя ветки:  " + request["Commit branch"].replace("_", " ") + "\n"                                   
                            "Состояние:  " + code(request["Состояние"]) + "\n"                         
                            "Роль:  " + code(request["Роль"]) + "\n"
                            "Статус проверки:  " + code(request["Статус проверки"]) + "\n"
                            "? Кол-во проверок:  " + bold(str(request["Кол-во проверок"])) + "\n"
                            "? Кол-во комментариев:  " + bold(str(request["Кол-во комментариев"])) + "\n"
                            "? Проверьте меня): ", request["Проверьте меня)"])
                await bot.send_message(message.from_user.id, mess, parse_mode=ParseMode.MARKDOWN)
    except:
        await message.reply("❗ Возникла ошибка, повторите команду! Или добавьтесь к боту!")

Она позволяет работать со всеми продуктами Atlassian (Confluense, ButBucket, Jira и т.д.). Достаточно понятная документация, которая описывает все методы, которые могут быть использованы с приммерами ответов.

Так же я использовал SQLite3 базу (присутствует по умолчанию) для записи комментрариев по реквесту и последующего сравниния информации сохранненой в базе с новой (которая в реквесте). Последующий вывод, если есть изменения.

Пример начала класса, для работы с базой SQLite3 , можно подробнее посмотреть на GitHub:

import sqlite3 as sq

class DataBase:
    comment = int()
    requests_id = list()

    def connect_to_db(self) -> sq.Connection:
        """
        Подключение к базе "bitbucket.db", если ее нет она будет создана

        :return: connection Соединение с базой
        """
        with sq.connect("bitbucket.db") as connection:
            return connection

    def create_cursor(self, connection) -> sq.Cursor:
        """
        Создает и возвращает курсор для работы с базой

        :param connection: Соединение с базой
        :return: cursor Курсор
        """
        cursor = connection.cursor()
        return cursor

    def close_db(self, connection):
        """
        Закрывает соединение с базой

        :param connection: Соединение с базой
        :return: Закрывает соединение с базой
        """
        connection.close()

    def create_table(self, connection, cursor):
        """
        Создает таблицу, куда будут писаться:
        userId: id реквеста
        userName: имя пользователя
        countCm: кол-во комментариев к реквесту

        :param connection: Соединение с базой
        :param cursor: Курсор для работы с базой
        :return: Создает таблицу
        """
        cursor.execute("""CREATE TABLE IF NOT EXISTS comments(
        userId INTEGER DEFAULT 0,
        userName TEXT NOT NULL,
        countCm INTEGER DEFAULT 0)""")
        connection.commit()

Для отслеживания кол-ва реквестов, я записывал просто в (file.json) файл константу, которая сравнивалась с актуальным результатом кол-ва реквестов, если он был больше, значит создан новый реквест и выводил сообщение в телеграмм.

Для доп.функции сделал вывод курса валют библиотекой Pycbrf , с помощью нее можно получить актуальные курсы валют на необходимые даты.

Пример:

from datetime import datetime
from decimal import Decimal

from pytz import timezone
from pycbrf.toolbox import ExchangeRates


class ExchangeCurrency:

    _usd = str()
    _eur = str()

    def get_date_now(self) -> str:
        saratov_tz = timezone('Europe/Moscow')
        now = saratov_tz.localize(datetime.now()).strftime("%Y-%m-%d")
        return now

    def get_usd_currency(self) -> Decimal:
        rates = ExchangeRates(self.get_date_now())
        return rates['USD'].value

    def get_eur_currency(self) -> Decimal:
        rates = ExchangeRates(self.get_date_now())
        return rates['EUR'].value

    def get_currency_info(self) -> str:
        usd = self.get_usd_currency()
        eur = self.get_eur_currency()
        return 'Курс Доллара: ' + str(usd) + ' RUB\n' + \
               'Курс Евро: ' + str(eur) + ' RUB'

Использование

Что бы использовать бота, предварительно, нужно его создать в телеграмме. Информации много, поэтому не буду рассказывать как это сделать.

Для работы бота в чате понадобиться chat_id, что бы его унать можно вывести его в консоль, отправив сообщение ботом в чат, привязав chat_id на вывод сообщения. Либо можно привязать вывод к выполнению команды в чат.

Для работы бота в личке, он будет получать user_id из контекста. Поэтому заранее узнавать его не надо.

Для того, чтобы запустить бота нужно предварительно указать в файле Utils.py следующие параметры:

Токен созданного бота
Адрес расположения проектов в BitBucket: "https://bitbucket.corporativ_adress.ru”
Логин
Пароль
Имя проекта в BitBucket
Название репозитория в проекте

После запуска бота, нужно его добавить в чат, далее можно узнать chat_id, как я писал уже выше.

В личке будет работать сразу.

Милости прошу :-) Спасибо за внимание!!!

Ссылка на GitHub

Телеграм бот на Python, для напоминаний о реквестах на BitBucket