Учим Raspberry Pi принимать Telegram'мы с помощью Bot API и Python

Давно хотел прикрутить к своей домашней Raspberry Pi удобный интерфейс «общения», который бы удовлетворял главному требованию — простота и лёгкость, с доступом из любой точки мира и с помощью любого оборудования (но в первую очередь — со смартфона).

В связи с отсутствием дома выделенного IP и наличием сурового и неподкупного NAT варианты с SSH клиентами и web-интерфейсами отпали сразу. Для небольших потребностей решение тоже должно быть простое, быстрое и, в качестве бонуса, надежное. Так что идея использования протокола одного из распространенных мессенджеров показалась мне весьма привлекательной. Под прицел попали Jabber, Telegram и WhatsApp.

Против Jabber сыграло нежелание устанавливать лишний клиент. Ну а так как Telegram — это, IMHO, тот же WhatsApp, только лучше и удобнее (и даже чуточку безопаснее), то именно на нём я и решил остановить свой выбор. К тому же появившаяся недавно в Telegram возможность создавать своих рабов ботов и взаимодействовать с ними с помощью очень простого API позволяет избавиться от необходимости регистрировать новый аккаунт, а так же дает некоторые очень полезные и удобные возможности.

На самом деле всё действительно настолько просто, что опытным человекам хватит и 30 минут, чтобы разобраться, поднять и настроить своего бота. Остальным же: Добро Пожаловать!

Результат поиска в рунете по словосочетанию «Telegram & Raspberry» оказался богат только на статью с Хабра «Raspberry и Telegram: предпосылки создания умного дома», в которой описываются базовые манипуляции с клиентом Telegram. Кстати, достаточно сырой продукт и заставить его нормально работать мне так и не удалось (на ровном месте через раз отказывается парсить одни и те же команды). Но, к счастью, мне он уже не нужен.

Итак, нам необходимо создать бота, для чего в любом клиенте Telegram'a (желательно последней версии) находим контакт с именем BotFather и просим его о /help. На что он ответит в достаточной мере подробной инструкцией и останется только следовать ей. Команды для совсем лентяев:

/newbot
<отображаемое имя нового бота>
<username нового бота>

Готово! Теперь BotFather предложит нам запомнить\сохранить token для досупа к боту через HTTP API, который нам скоро пригодится.

Так как программист я очень начинающий, то хорошо знаком только с Python, который, тем не менее, прекрасно подходит для данной задачи. Начнем.

Для существенного облегчения жизни и сокращения кода, предлагаю установить библиотеку для упрощения HTTP-запросов requests с помощью команды:

pip install requests

Теперь вся задача сводится к написанию простого скрипта, который через заданный промежуток времени будет запрашивать у сервера обновления сообщений. Если таковые имеются и сообщение отправлено определенным заранее пользователем, а так же содержит в тексте заданную команду, тогда будет выполнятся соответствующее этой команде действие. Такой скрипт я и предлагаю вашему вниманию. Пока это всего лишь шаблон, который можно адаптировать под свои нужды, но со временем планирую сделать из него что-то более приличное.

telegram.py (python2.7) - обновлено

# -*- coding: utf-8 -*-
import requests
import time
import subprocess
import os
#import mailchecker

requests.packages.urllib3.disable_warnings() # Подавление InsecureRequestWarning, с которым я пока ещё не разобрался

# Ключ авторизации Вашего бота Вы можете получить в любом клиенте Telegram у бота @BotFather
# ADMIN_ID - идентификатор пользователя (то есть Вас), которому подчиняется бот
# Чтобы определить Ваш ID, я предлагаю отправить боту сообщение от своего имени (аккаунта) через любой клиент
# А затем получить это сообщения с помощью обычного GET запроса
# Для этого вставьте в адресную строку Вашего браузера следующий адрес, заменив <token> на свой ключ:
# https://api.telegram.org/bot<token>/getUpdates
# Затем, в ответе найдите объект "from":{"id":01234567,"first_name":"Name","username":"username"}
# Внимательно проверьте имя, логин и текст сообщения
# Если всё совпадает, то цифровое значение ключа "id" - это и есть ваш идентификатор

# Переменным ADMIN_ID и TOKEN необходимо присвоить Вашим собственные значения
INTERVAL = 3 # Интервал проверки наличия новых сообщений (обновлений) на сервере в секундах
ADMIN_ID = 12345678 # ID пользователя. Комманды от других пользователей выполняться не будут
URL = 'https://api.telegram.org/bot' # Адрес HTTP Bot API
TOKEN = '123456789:???????????????????????????????????' # Ключ авторизации для Вашего бота
offset = 0 # ID последнего полученного обновления

def check_updates():
    """Проверка обновлений на сервере и инициация действий, в зависимости от команды"""
    global offset
    data = {'offset': offset + 1, 'limit': 5, 'timeout': 0} # Формируем параметры запроса

    try:
        request = requests.post(URL + TOKEN + '/getUpdates', data=data) # Отправка запроса обновлений
    except:
        log_event('Error getting updates') # Логгируем ошибку
        return False # Завершаем проверку

    if not request.status_code == 200: return False # Проверка ответа сервера
    if not request.json()['ok']: return False # Проверка успешности обращения к API
    for update in request.json()['result']: # Проверка каждого элемента списка
        offset = update['update_id'] # Извлечение ID сообщения

        # Ниже, если в обновлении отсутствует блок 'message'
        # или же в блоке 'message' отсутствует блок 'text', тогда
        if not 'message' in update or not 'text' in update['message']:
            log_event('Unknown update: %s' % update) # сохраняем в лог пришедшее обновление
            continue # и переходим к следующему обновлению
        from_id = update['message']['chat']['id'] # Извлечение ID чата (отправителя)
        name = update['message']['chat']['username'] # Извлечение username отправителя
        if from_id <> ADMIN_ID: # Если отправитель не является администратором, то
            send_text("You're not autorized to use me!", from_id) # ему отправляется соответствующее уведомление
            log_event('Unautorized: %s' % update) # обновление записывается в лог
            continue # и цикл переходит к следующему обновлению
        message = update['message']['text'] # Извлечение текста сообщения
        parameters = (offset, name, from_id, message)
        log_event('Message (id%s) from %s (id%s): "%s"' % parameters) # Вывод в лог ID и текста сообщения

        # В зависимости от сообщения, выполняем необходимое действие
        run_command(*parameters)
        
def run_command(offset, name, from_id, cmd):
    if cmd == '/ping': # Ответ на ping
        send_text(from_id, 'pong') # Отправка ответа

    elif cmd == '/help': # Ответ на help
        send_text(from_id, 'No help today. Sorry.') # Ответ

    elif cmd == '/photo': # Запрос фотографии с подключенной Web-камеры
        # Для оператора If ниже. Если первая попытка успешна - выполняется условие, если нет, то вторая попытка и условие
        # Если и вторая не успешна, тогда отчитываемся об ошибке
        # Всё потому, что на моей конфигурации крайне изредка камера бывает недоступна с первого раза
        if make_photo(offset) or make_photo(offset):
            # Ниже, отправка пользователю уведомления об активности бота
            requests.post(URL + TOKEN + '/sendChatAction', data={'chat_id': from_id, 'action': 'upload_photo'})
            send_photo(from_id, offset) # Вызов процедуры отправки фото
        else:
            send_text(from_id, 'Error occured') # Ответ, сообщающий об ошибке

    elif cmd == '/mail':
        check_mail() # Вызов процедуры проверки почты
    else:
        send_text(from_id, 'Got it.') # Отправка ответа

def log_event(text):
    """
    Процедура логгирования
    ToDo: 1) Запись лога в файл
    """
    event = '%s >> %s' % (time.ctime(), text)
    print event

def send_text(chat_id, text):
    """Отправка текстового сообщения по chat_id
    ToDo: повторная отправка при неудаче"""
    log_event('Sending to %s: %s' % (chat_id, text)) # Запись события в лог
    data = {'chat_id': chat_id, 'text': text} # Формирование запроса
    request = requests.post(URL + TOKEN + '/sendMessage', data=data) # HTTP запрос
    if not request.status_code == 200: # Проверка ответа сервера
        return False # Возврат с неудачей
    return request.json()['ok'] # Проверка успешности обращения к API

def make_photo(photo_id):
    """Обращение к приложению fswebcam для получения снимка с Web-камеры"""
    photo_name = 'photo/%s.jpg' % photo_id # Формирование имени файла фотографии
    subprocess.call('fswebcam -q -r 1280x720 %s' % photo_name, shell=True) # Вызов shell-команды
    return os.path.exists(photo_name) # Проверка, появился ли файл с таким названием

def send_photo(chat_id, photo_id):
    """Отправка фото по его идентификатору выбранному контакту"""
    data = {'chat_id': chat_id} # Формирование параметров запроса
    photo_name = 'photo/%s.jpg' % photo_id # Формирования имени файла фотографии
    if not os.path.exists(photo_name): return False # Проверка существования фотографии
    files = {'photo': open(photo_name, 'rb')} # Открытие фото и присвоение
    request = requests.post(URL + TOKEN + '/sendPhoto', data=data, files=files) # Отправка фото
    return request.json()['ok'] # Возврат True или False, полученного из ответа сервера, в зависимости от результата

def check_mail():
    """Проверка почтовых ящиков с помощью самодельного модуля"""
    print "Подключите и настройте модуль проверки почты"
    return False
    try:
        log_event('Checking mail...') # Запись в лог
        respond = mailchecker.check_all() # Получаем ответ от модуля проверки
    except:
        log_event('Mail check failed.') # Запись в лог
        return False # И возврат с неудачей
    if not respond: respond = 'No new mail.' # Если ответ пустой, тогда заменяем его на соответствующее сообщение
    send_text(ADMIN_ID, respond) # Отправляем это сообщение администратору
    return True

if __name__ == "__main__":
    while True:
        try:
            check_updates()
            time.sleep(INTERVAL)
        except KeyboardInterrupt:
            print 'Прервано пользователем..'
            break

Советы

Чтобы запустить данный скрипт в фоновом режиме на Raspberry Pi, можно воспользоваться двумя способами:
1) С помощью screen. Инструкция по использованию тут.
2) Командами:

python telegram.py
CTRL+Z
bg

Если хотите поставить этот скрипт в автозапуск, необходимо в файл /etc/rc.local, перед строкой 'exit 0', добавить:

python <путь к файлу>/telegram.py

Например так:

nano /etc/rc.local
    ...
    python /home/pi/telegram.py

    exit 0

И естественно, на вашей Raspberry должен быть установлен python2.7.

Это, конечно же, только начало, наброски. Чуть позже прикручу несколько интересных функций. Например, получение снимка с камеры по команде и некоторые посложнее, такие как как управление гирляндой на WS2801 и другие.

Буду очень рад любым замечаниям, советам и предложениям.

Также, как вы уже заметили, скрипт проверяет сообщения с определенным промежутком времени. Реализовать прием WebHook без посредника не представляется возможным. Игрался со значениями «timeout» в методе «getUpdates», — безрезультатно. Буду благодарен за любые идеи и на этот счет.

[ Telegram Bot API ]

UPD (от 03.07). Код обновлен.