Хост-трекер на python — ежеминутный мониторинг доступности сайтов

Disclamer. Это моя первая программа на пайтоне — предпочитаю изучать новое с постановки конкретной задачи, а в процессе углубляться в изучение. На мой взгляд, такой порядок интересней, чем провести часы за чтением туториала перед написанием «хэлло, ворлд». Но если совсем не углубляться в изучение синтаксиса языка, то продуктом, скорее всего, станет глючный говнокод. Считаю, что я соблюдаю баланс между изучением теории и практикой, но сеньор девелопер может считать иначе. В общем, я предупредил, что содержимое поста может оказаться говнокодом, написанным по модели «AS IS».

TL;DR. Код запускал на python3. Работоспособность на 2-й версии не проверял. В задачу входило использование только стандартных библиотек, чтобы код работал на дешёвом шаред-хостинге, где нет доступа к pip (позже я узнал, что на хостинге надо юзать pip с ключом --user, чтобы добавлять библиотеки пользователю с ограниченными правами).

Задачи программы:

1. Обойти (запросить по http[s]) список url-адресов.
2. Если код http-ответа или ошибки отличается от полученного в прошлый раз, уведомить об этом в Телеграм и сохранить информацию в лог.

Список адресов со статусами доступа и лог решил хранить в базе sqlite3, т.к. при всей простоте реализации это возможность хранить данные структурировано, использовать мощь языка запросов SQL и иметь возможность в дальнейшем дополнять функционал: например, добавить отчёты аптайма хостов.

Пока для поставленной задачи достаточно двух таблиц. 1-я для хранения url-адресов:

CREATE TABLE `hosts` ( `id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE, `url` TEXT, `msg` TEXT )

2-я для для ведения лога:

CREATE TABLE `events` ( `id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE, `host_id` INTEGER NOT NULL, `event` TEXT, `time` TEXT NOT NULL )

Следующий фрагмент python-кода позволяет подключиться к БД (если базы ещё нет, то автоматически создастся) и выполнить заданный SQL:

import sqlite3
conn = sqlite3.connect("tracker.db")
cursor = conn.cursor()
cursor.execute("CREATE TABLE ...")
conn.commit()

А так можно добавить хосты:

hosts = [['http://host1.ru/'],['https://host2.ru/path']]
cursor.executemany("INSERT INTO hosts (url, msg) VALUES (?, 'New')", hosts)

Также для визуальной работы с базами sqlite3 есть удобная софтина на сайте sqlitebrowser.org

Итак, таблицы готовы, список тестируемых хостов есть. Пишем основной скрипт. Для приготовления кода нам понадобятся:

import sqlite3
from urllib import request, parse
from urllib.request import urlopen
from urllib.error import HTTPError
from datetime import datetime

Мне требуется, чтобы при проверке url редирект воспринимался как ошибка. Ведь странно, если поставленный на мониторинг url вдруг стал пересылать на другой адрес. Для этой задачи поможет следующий код:

class NoRedirectHandler(request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None
noRedirect = request.build_opener(NoRedirectHandler)

Теперь чувствительные к редиректу запросы я буду делать через noRedirect.open(), а остальные — стандартно через urlopen().

Дело за малым — обойти список url, и если статус ответа изменился, отправить сообщение в Телеграм и сохранить новый статус. Для отправки в Телеграм послужит такая функция:

def tg_send(msg):
    base_url = 'https://api.telegram.org/botXXX:YYY/sendMessage?chat_id=123456&text='
    return urlopen(base_url+parse.quote_plus(msg)).status

Здесь XXX:YYY — это токен вашего телеграм бота, полученный при его создании, а 123456 — id чата вашего аккаунта и бота, в этот чат бот будет отправлять сообщения (т.е. только вам). Узнать id чата можно перейдя по ссылке https://api.telegram.org/botXXX:YYYY/getUpdates (не забыв подставить в ссылку токен). После того, как вы напишете вашему боту любое сообщение, по этой ссылке вы увидите json-данные, в которых "chat":{"id":123456, — то, что вам нужно.

Приступаем к обходу списка url, соединяемся с базой:

conn = sqlite3.connect("tracker.db")
conn.row_factory = sqlite3.Row
cursor = conn.cursor()

Здесь row_factory нам нужно, чтобы к полученным из БД полям обращаться по их имени, а не по номеру. Далее — цикл перебора хостов:

for host in cursor.execute("SELECT * FROM hosts").fetchall():
    url = host['url']
    try:
        response = urlopen(url, timeout=20)
    except (HTTPError) as e:
        msg = str(e)
    except Exception as e:
        msg = str(e.__class__.__name__)+': '+str(getattr(e, 'reason', e))
    else:
        msg = 'OK '+str(response.status)
    if host['msg'] == msg: continue
    cursor.execute("UPDATE hosts SET msg = ? WHERE id = ?", (msg, host['id']))
    cursor.execute("INSERT INTO events (host_id, event, time) VALUES (?, ?, ?)", (host['id'], msg, datetime.now()))
    conn.commit()
    tg_send(url+' '+msg)

Если не использовать .fetchall() в запросе хостов, то цикл будет работать до первого случая обнаружения изменившегося статуса хоста (msg) — т.к. при этом мы пишем в таблицы и меняем cursor. Трудноуловимый баг, особенно когда только учишься в python. timeout=20 нужен, чтобы не ждать дольше 20 секунд ответа от хоста. Я нетерпеливый. Если msg не изменился с прошлой проверки (такой же, как в БД у хоста), то с помощью continue переходим к следующему хосту, не выполняя дальнейшие инструкции. Если же msg изменился, делаем запрос на обновление записи в БД, на добавлление ивента, conn.commit() выполняет запросы, а tg_send(url+' '+msg) сообщает в телеграм.

Полученный tracker.py запускаем в консоли командой python3 tracker.py, если работает без ошибок, записываем в crontab на ежеминутное выполнение. И спим спокойные за свои сайты, пока сигнал тревоги не разбудит. Кстати, Телеграм, помимо простого API, мне нравится возможностью индивидуальной настройки уведомлений для каждого чата. В том числе, уведомления важных чатов можно сделать со звуком или вибрацией даже в беззвучном режиме телефона.