Статья Шлем важные Exception себе в Telegram

Всем привет. Пишу код в свободное время, для меня python - это хобби.
Прошу дать обратную связь с высоты своего опыта, если я допустил какие-то грубые ошибки.

Сервис получился простой, но эффективный. Реализовал все за пару часов, а работает без сбоев уже год. То есть, реализовал и забыл.

Итак, написал я одну программу, потом еще один скрипт, потом объединил их в проект, а потом еще один скрипт и так далее.
В какой-то момент возникла проблема: постоянно в фоне в разных виртуальных окружениях эти скрипты падают, вылетают какие-то непредусмотренные Exceptions и я не всегда узнаю об этом моментально. Тем временем проходит время, а нужный сборщик данных молча сидит и нифига не собирает уже пару дней. Но я бы хотел получать важные уведомления от моих проектов моментально.

Схема для решения нарисовалась вот такая:

Таким образом нам понадобится поднять специальный Webhook сервер, в который мы будем кидать POST запросом с заранее продуманным содержимым, чтоб это сразу транслировалось в личку в телеграм.

Реализуем Webhook server

Реализаций вебсервера вокруг очень много, я выбирал framework под свои цели:
- Поднять как daemon на своем VPS
- Простой в конфиге
- Не требовательный
- Без лишних понтов

Выбор пал на библиотеку cherrypy (

Ссылка скрыта от гостей

)

Для его работы потребуется создать бота в telegram (

Ссылка скрыта от гостей

), а так же сделать самоподписанный сертификат, т.к. api телеграмма работает только через https.

Спойлер: Код

import cherrypy
import telebot
import config

webhook_host = config.webhook_host
webhook_port = config.webhook_port
webhook_listen = '0.0.0.0'

webhook_ssl_cert = 'webhook_cert.pem_PATH'  # Файл сертификата
webhook_ssl_priv = 'webhook_pkey.pem_PATH'  # Файл приватного ключа сертификата

webhook_url_base = f'https://{webhook_host}:{webhook_port}'
webhook_url_path = f'/{config.webhook_bot_token}/'

bot = telebot.TeleBot(config.webhook_bot_token)

# вебхук-сервер
class WebhookServer(object):
    @cherrypy.expose
    def index(self):
        if 'content-length' in cherrypy.request.headers and \
                        'content-type' in cherrypy.request.headers and \
                        cherrypy.request.headers['content-type'] == 'application/json':
            length = int(cherrypy.request.headers['content-length'])
            json_string = cherrypy.request.body.read(length).decode("utf-8")
            #print(f'Json-string={json_string}')
            update = telebot.types.Update.de_json(json_string)
            #print(f'update={update}')
            # Проверка входящего сообщения
            bot.process_new_updates([update])
            return ''
        else:
            raise cherrypy.HTTPError(403)

# Хэндлер на все текстовые сообщения
@bot.message_handler(content_types=['text'])
def send_message(message):
    bot.send_message(chat_id=config.admin_id, text=message.text)

# Снимаем вебхук перед повторной установкой (избавляет от некоторых проблем)
bot.remove_webhook()

# Ставим заново вебхук
bot.set_webhook()
bot.set_webhook(url=webhook_url_base + webhook_url_path, certificate=open(webhook_ssl_cert, 'r'))

cherrypy.config.update({
    'server.socket_host': webhook_listen,
    'server.socket_port': webhook_port,
    'server.ssl_module': 'builtin',
    'server.ssl_certificate': webhook_ssl_cert,
    'server.ssl_private_key': webhook_ssl_priv
})

# запуск
cherrypy.quickstart(WebhookServer(), webhook_url_path, {'/': {}})

Чтоб код заработал, рядом кладу файл config.py и оставляю там чувствительные данные.

Спойлер: Код

webhook_bot_token = 'webhook_bot_token'
admin_id = 'my_telegram_id'
webhook_host = 'webhook_host'
webhook_port = webhook_port

Ну в общем вот и все. Не могу сказать, что по пути реализации тут встречаются какие-то сложные проблемы. Есть только такой нюанс: чтоб узнать свой id в телеграм, надо написать боту @ShowJsonBot
Данный скрипт я превратил в службу на своем vps но это уже совсем другая история

Добавляем код в важные файлы/функции в своем проекте

Теперь нужно собрать Exception в строчку и отправить через webhook server себе в личку.
Чтобы отправить сообщение себе в лс, смотрим в документации telegram api метод

Ссылка скрыта от гостей

Все, что нужно - это указать заголовок Content-type, вложить в запрос text и chat_id.
Теперь в любой важной программе добавляю универсальную функцию по отправке выскочившего Exception мне в телеграмм:

Спойлер: код

def do_alarm(t_alarmtext):
    headers = {"Content-type": "application/json"}
    payload = {"text": f"{t_alarmtext}", "chat_id": f"{admin_id}"}
    requests.post(url=webhook_url, data=json.dumps(payload), headers=headers)

На входе функция принимает текст, на выходе отправляет POST запрос.

Сам я этим способом, например, смотрю исключения, которые вылетают в коде после тестов через день/неделю/год/поколение. По этому в блок main добавляю код:

Спойлер: код

    except Exception as e:
        t_alarmtext = f'Project_name (app.py): {str(e)}'  # название файла/функции/проекта
        do_alarm(t_alarmtext)
        logger.error(f'Other except error Exception', exc_info=True)

Переводя на человеческий язык: Ну ты это... Пиши если че. + запись ошибки в лог.

Выводы:

• Поставленную проблему данное решение закрывает
• По-хорошему надо засунуть это в докер
• Потенциально это можно превратить в простейший сервис мониторинга но лучше не стоит, юзай графану, не собирай шаньги с могил

Всем кто осилил спасибо за внимание!