[Из песочницы] Учим 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
<отображаемое имя нового бота>
Готово! Теперь BotFather предложит нам запомнить\сохранить token для досупа к боту через HTTP API, который нам скоро пригодится.
Так как программист я очень начинающий, то хорошо знаком только с Python, который, тем не менее, прекрасно подходит для данной задачи. Начнем.
Для существенного облегчения жизни и сокращения кода, предлагаю установить библиотеку для упрощения HTTP-запросов requests с помощью команды:
pip install requests
Теперь вся задача сводится к написанию простого скрипта, который через заданный промежуток времени будет запрашивать у сервера обновления сообщений. Если таковые имеются и сообщение отправлено определенным заранее пользователем, а так же содержит в тексте заданную команду, тогда будет выполнятся соответствующее этой команде действие. Такой скрипт я и предлагаю вашему вниманию. Пока это всего лишь шаблон, который можно адаптировать под свои нужды, но со временем планирую сделать из него что-то более приличное.
# -*- coding: utf-8 -*-
import requests
import time
requests.packages.urllib3.disable_warnings() # Подавление InsecureRequestWarning, с которым я пока ещё не разобрался
# Ключ авторизации Вашего бота Вы можете получить в любом клиенте Telegram у бота @BotFather
# ADMIN_ID - идентификатор пользователя (то есть Вас), которому подчиняется бот
# Чтобы определить Ваш ID, я предлагаю отправить боту сообщение от своего имени (аккаунта) через любой клиент
# А затем получить это сообщения с помощью обычного GET запроса
# Для этого вставьте в адресную строку Вашего браузера следующий адрес, заменив на свой ключ:
# https://api.telegram.org/bot/getUpdates
# Затем, в ответе найдите объект "from":{"id":01234567,"first_name":"Name","username":"username"}
# Внимательно проверьте имя, логин и текст сообщения
# Если всё совпадает, то цифровое значение ключа "id" - это и есть ваш идентификатор
# Переменным ADMIN_ID и TOKEN необходимо присвоить Вашим собственные значения
INTERVAL = 5 # Интервал проверки наличия новых сообщений (обновлений) на сервере в секундах
ADMIN_ID = 01234567 # ID пользователя. Комманды от других пользователей выполняться не будут
URL = 'https://api.telegram.org/bot' # Адрес HTTP Bot API
TOKEN = '012345678:???????????????????????' # Ключ авторизации для Вашего бота
offset = 0 # ID последнего полученного обновления
def make_url_query_string(params):
"""
Конвертирование словаря параметров в строку типа "URL query string"
Пример: '(http://site.com/home)?param1=value1¶m2=value2'
"""
return '?' + '&'.join([str(key) + '=' + str(params[key]) for key in params])
def check_updates(limit=5):
"""
Проверка обновлений на сервере и инициация действий, в зависимости от команды
ToDo: 1) повторная отправка при неудаче
2) сопоставление команд и действий
3) добавить логгирование
"""
global offset
params = make_url_query_string({'offset': offset+1, 'limit': limit, 'timeout': 0})
request = requests.get(URL + TOKEN + '/getUpdates' + params) # Отправка запроса обновлений
if not request.status_code == 200: return False # Проверка ответа сервера
if not request.json()['ok']: return False # Проверка успешности обращения к API
if not request.json()['result']: return False # Проверка наличия обновлений в возвращенном списке
for update in request.json()['result']: # Проверка каждого элемента списка
offset = update['update_id'] # Извлечение ID сообщения
from_id = update['message']['from']['id'] # Извлечение ID отправителя
if from_id <> ADMIN_ID: # Проверка ID отправителя и если контакт не является администратором, то
send_respond("You're not autorized to use me!", from_id) # ему отправляется соответствующее уведомление
continue # и цикл переходит к следующему сообщению
message = update['message']['text'] # Извлечение текста сообщения
# Следующий код выводит в консоль ID и текст сообщения
print '>> OFFSET: ', offset
print '>> MESSAGE:', message
print '-' * 10
send_respond('Принято!', ADMIN_ID)
###
# Место для кода, выполняющего определенные команды,
# в зависимости от содержания полученного сообщения
if message == 'ping':
send_respond('pong', from_id)
def send_respond(text, chat_id):
"""Отправка текстового сообщения по chat_id или user_id (чем они отличаются?)
ToDo: повторная отправка при неудаче"""
params = make_url_query_string({'chat_id': chat_id, 'text': text}) # Преобразование параметров
request = requests.get(URL + TOKEN + '/sendMessage' + params) # HTTP запрос
if not request.status_code == 200: return False # Проверка ответа сервера
if not request.json()['ok']: return False # Проверка успешности обращения к API
return True
if not __name__ == "__main__": exit()
while True:
try:
check_updates()
time.sleep(INTERVAL)
except KeyboardInterrupt:
print 'Прервано пользователем..'
break
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 ]