Будучи руководителем группы разработки или просто частью команды по разработке часто встает вопросы информированности коллег о релизах новых версий внутренних инструментов. И порой даже сообщения в общие чаты не помогают, да и руками писать это практика плохая скажу я вам.
И вот решил я разработать инструмент который слал бы сообщения о пуша\удаления и проверках в наш harbor. К слову harbor это open-source реестр (registry) для хранения и управления Docker-контейнерами и Helm-чартами и Docker-образами, ориентированный на корпоративное использование.
Но вернемся к сути статьи, оповещения должны были слаться в телеграм чаты, в том числе в разные топики для удобства координации наших команд, так как координация всех действий и любое взаимодействие происходит именно через него. В общем создали мы топик "Releases and Updates" и начал я думать как реализовать, первым делом подумал про API, но как я понял оно там несколько для других целей. Проще всего было это организовать через webhook'и, оно к слову в harbor из коробки, как раз для тех целей которые были нужны.
Вот следующая последовательность действий для настройки этого добра:
Откройте Telegram и найдите пользователя BotFather.
Создайте нового бота, отправив команду /newbot и следуйте инструкциям.
Запишите ваш токен бота, который вы получите в результате.
Добавьте бота в канал\группу.
У бота должны быть админские права, чтобы он мог слать оповещения.
На этом пока отложим telegram и перейдем к настройкам harbor.
Зайдите в harbor.
Перейдите в настройки репозитория.
Настройте новый Webhook, указав URL вашего сервера, который будет принимать уведомления (например, http://your-server-ip:5000/webhook).
Убедитесь, что вы выбрали необходимое событие, push.
Далее переходим к тонкостям отправки сообщений в telegram, а именно к отправки сообщений в конкретный топик. А нюанс вот в чем:
Чат id можно скопировал через ссылку на сообщение по типу - https://t.me/c/123xxxx567/17/16
(Вот расшифровка ссылки https://t.me/c/{chat_id}/{message_thread_id}/{message_id})
Для использования chat_id в API надо добавить -100, получите -100123xxxx567
Теперь переходим к разработке, я решил использовать flask для этих целей.
Примерный вид POST сообщения о пуше от harbor -
INFO:werkzeug:127.0.0.1 - - [30/Apr/2025 10:21:33] "POST /webhook HTTP/1.1" 200 -
вот тело сообщения:
{ "type": "PUSH_ARTIFACT", "occur_at": 1746008487, "operator": "sergey.akhmineev", "event_data": { "resources": [ { "digest": "sha256:9sdfgsdfgsdfgsdfgsgdfgsfgsdfgsdfgsdfgsdfgsdfgsfdgs", "tag": "1.9.0", "resource_url": "your-server-ip/your-repo/your-images:1.9.0" } ], "repository": { "date_created": 1733465413, "name": "your-images", "namespace": "your-repo", "repo_full_name": "your-repo/your-images", "repo_type": "public" } } }
таким образом все необходимое нам есть + webhook по сути убирает необходимость запуска скрипта по расписанию для чтения API, ведь мы просто ждем когда прилетит сообщение.
Переходим к разработке, вот минимум для отправки сообщения в telegram
# Извлечение токена Telegram и Chat ID из конфигурации TELEGRAM_API_URL = f'https://api.telegram.org/bot{conf_dict["telegram"]["bot_token"]}/sendMessage' CHAT_ID = conf_dict["telegram"]["chat_id"] MESSAGE_THREAD_ID = conf_dict.get("telegram", {}).get("message_thread_id") # Отправка сообщения def send_message_to_telegram(message): payload = { 'chat_id': CHAT_ID, 'text': message, 'parse_mode': 'Markdown', # Для форматирования сообщения } if MESSAGE_THREAD_ID: payload['message_thread_id'] = MESSAGE_THREAD_ID try: response = requests.post(TELEGRAM_API_URL, json=payload) response.raise_for_status() # Проверка на HTTP-ошибки logging.info("Сообщение успешно отправлено в Telegram.") except requests.exceptions.RequestException as e: logging.error(f"Ошибка при отправке сообщения: {str(e)}")
А вот и само приложение для того чтобы слушать webhook:
@app.route('/webhook', methods=['POST']) def webhook(): data = request.json logging.info(f"Получены данные вебхука: {data}") if data and data.get('type') == 'PUSH_ARTIFACT': event_data = data.get('event_data', {}) repository_info = event_data.get('repository', {}) resources = event_data.get('resources', []) repo_full_name = repository_info.get('repo_full_name', 'неизвестный репозиторий') repo_full_name = escape_markdown(repo_full_name) # Экранирование if not resources: logging.warning("Нет ресурсов для обработки.") return 'No resources to process', 200 # Формарование сообщения для отправки messages = [] for resource in resources: tag = escape_markdown(resource.get('tag', 'неизвестный тег')) # Экранирование тега resource_url = resource.get('resource_url', 'неизвестный URL') operator = escape_markdown(data.get('operator', 'неизвестный')) # Экранирование оператора message = ( f"📦 *Новый пуш в Harbor*\n" f"*Репозиторий:* {repo_full_name}\n" f"*Тег:* {tag}\n" f"*URL:* `{resource_url}`\n" f"*Автор пуша:* {operator}" ) messages.append(message) # Отправка всех сообщений последовательно for msg in messages: logging.info(f"Отправка сообщения: {msg}") send_message_to_telegram(msg) else: logging.warning("Получены некорректные данные вебхука или отсутствует тип 'PUSH_ARTIFACT'.") return 'OK', 200
В целом остается сделать экранирование спецсимволов -
def escape_markdown(text): """ Экранирует специальные символы Markdown в тексте. """ escape_chars = ['\\', '_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '!'] for char in escape_chars: text = text.replace(char, f'\\{char}') return text
и наконец добавить чтение данных для скрипта через конфиг. На этом первая часть заканчивается, оповещения работают, но я собираюсь вынести спецсимволы в конфиг, так как у многих свои способы именования образов, а так же добавить сообщения по удалению, проверку образов и прочее. Надеюсь кому то пригодится, а если кому необходимо, вот репозиторий с проектом.
Webhook у Harbor или как я оповещения о пушах docker images нашей команды делал часть — 2
