Наш телеграм бот будет уметь следующее:
- Получать список DevOps вакансий с HH.
- Отправлять погоду по разным городам и странам.
- Показывать парочку основных курсов валют.
- Конвертировать парочку основных валют из одной в другую.
- Получать инфо из wikipedia.
- Иметь кнопки для этого.
Найти код этого бота (возможно он будет немного старым) в одном месте можно в моем github.
Заводим бота через другого бота, который botFather и сохраняем себе полученный токен.
Начнем с написания кнопок. Все это сейчас мы делаем в main.py с помощь пакета telebot.
import telebot
telegram_bot_token = "<bot_token>"
bot = telebot.TeleBot(telegram_bot_token)
@bot.message_handler(commands=["start"])
def start(m):
markup_inline = types.InlineKeyboardMarkup(row_width=2)
weather_button = types.InlineKeyboardButton("Узнать погоду", callback_data="weather")
hh_button = types.InlineKeyboardButton("Вакансии на Кипре", callback_data="cyprus")
exchange_button = types.InlineKeyboardButton("Курсы валют", callback_data="ex")
converter_button = types.InlineKeyboardButton("Конвертер валют", callback_data="convert")
help_button = types.InlineKeyboardButton("Помощь", callback_data="help")
wiki_button = types.InlineKeyboardButton("Wiki", callback_data="wiki")
markup_inline.add(weather_button, hh_button, exchange_button, help_button, wiki_button, converter_button)
if m.text == "/start":
bot.send_message(m.chat.id, "Добро пожаловать в инфо Бот!", reply_markup=markup_inline)
Собственно с помощью декоратора создаем возврат кнопок на команду /start.
Выглядеть это будет примерно вот так
Далее реализуем следующую логику:
«Узнать погоду» — нажимаем и бот просит нас ввести название города, мы вводим, бот отправляет информации о погоде в этом городе.
«Вакансии на Кипре» — тут сразу получаем выхлоп в виде вакансий DevOps инженера на Кипре.
«Курсы валют» — в ответ возвращают кнопки с тем, какую валюту хотим выбрать и только после этого курс.
«Помощь» — стандартный раздел, который вернет некий «man» о том как пользоваться ботом.
«Wiki» — тут логика как и с погодой. Бот предложит ввести какой-либо запрос в википедию и после вернет краткий ответ из нее.
«Конвертер валют» — здесь же в свою очередь логика схожа с кнопкой «Курсы валют».
Что ж давайте писать.
@bot.callback_query_handler(func=lambda call:True)
def callback(call):
currency_menu = types.InlineKeyboardMarkup(row_width=3)
converter_menu = types.InlineKeyboardMarkup(row_width=3)
convert_usd_button = types.InlineKeyboardButton("USD/RUB", callback_data="convert_usd_rub")
convert_eur_button = types.InlineKeyboardButton("EUR/RUB", callback_data="convert_eur")
convert_rub_to_usd_button = types.InlineKeyboardButton("RUB/USD", callback_data="convert_rub_to_usd")
convert_rub_to_eur_button = types.InlineKeyboardButton("RUB/EUR", callback_data="convert_rub_to_eur")
usd_rub_button = types.InlineKeyboardButton("USD/RUB", callback_data="usd_rub")
amd_rub_button = types.InlineKeyboardButton("AMD/RUB", callback_data="amd_rub")
euro_rub_button = types.InlineKeyboardButton("EUR/RUB", callback_data="eur_rub")
currency_menu.add(usd_rub_button, amd_rub_button, euro_rub_button)
converter_menu.add(convert_usd_button, convert_eur_button, convert_rub_to_usd_button, convert_rub_to_eur_button)
if call.message:
if call.data == "weather":
msg = bot.send_message(call.message.chat.id, "Введите название города: ")
bot.register_next_step_handler(msg, show_weather)
elif call.data == "cyprus":
vac = Vacancies()
vac_message = vac.get_vacancies()
bot.send_message(call.message.chat.id, vac_message)
elif call.data == "wiki":
msg = bot.send_message(call.message.chat.id, "Введите запрос в wikipedia")
bot.register_next_step_handler(msg, show_wiki_answer)
elif call.data == "ex":
bot.send_message(call.message.chat.id, "Курсы валют", reply_markup=currency_menu)
elif call.data == "usd_rub":
currency_usd_rub = Exchange("USD/RUB")
valute = float(currency_usd_rub.get_curret_rate())
message = f"USD: 1 => RUB: {round(valute, 2)}"
bot.send_message(call.message.chat.id, message)
elif call.data == "amd_rub":
currency_amd_rub = Exchange("AMD/RUB")
valute = float(currency_amd_rub.get_curret_rate())
bot.send_message(call.message.chat.id, f"AMD: 1 => RUB: {round(valute, 2)}")
elif call.data == "eur_rub":
currency_eur_rub = Exchange("EUR/RUB")
valute = float(currency_eur_rub.get_curret_rate())
bot.send_message(call.message.chat.id, f"EUR: 1 => RUB: {round(valute, 2)}")
elif call.data == "help":
bot.send_message(call.message.chat.id, "Для началы работы с ботом напишите /start")
elif call.data == "convert":
bot.send_message(call.message.chat.id, "Конвертер валют", reply_markup=converter_menu)
elif call.data == "convert_rub_to_usd":
msg = converter_send_message(call)
bot.register_next_step_handler(msg, run_converter, "RUB/USD")
elif call.data == "convert_usd_rub":
msg = converter_send_message(call)
bot.register_next_step_handler(msg, run_converter, "USD/RUB")
elif call.data == "convert_rub_to_eur":
msg = converter_send_message(call)
bot.register_next_step_handler(msg, run_converter, "RUB/EUR")
elif call.data == "convert_eur":
msg = converter_send_message(call)
bot.register_next_step_handler(msg, run_converter, "EUR/RUB")
Здесь мы отрисовываем кнопки и связываем их через «reply_markup». Это позволяет нам вместе с тектовым сообщением вернуть другое, описанное выше меню.
Далее используем специальный метод «bot.register_next_step_handler» чтобы передать выполнение в другую функцию.
Описываем финальную часть основного кода
def converter_send_message(call):
msg = bot.send_message(call.message.chat.id, "Введите конвертируемую сумму")
return msg
def show_weather(msg):
weather = Weather(msg.text)
weather_in_city = weather.get_current_weather()
bot.send_message(msg.chat.id, weather_in_city)
def show_wiki_answer(msg):
wiki = Wiki(msg.text)
wiki_response = wiki.get_wiki_info()
bot.send_message(msg.chat.id, wiki_response)
def run_converter(msg, ex):
conv = Exchange(ex)
ex_first = ex.split("/")[0]
ex_second = ex.split("/")[1]
valute = conv.get_curret_rate()
give_sum = int(msg.text)
if ex_first == "RUB":
result = float(give_sum / valute)
else:
result = float(give_sum * valute)
bot.send_message(msg.chat.id, f" {give_sum} {ex_first} => {round(result, 1)} {ex_second}")
def main():
bot.polling(none_stop=True, interval=0)
if __name__ == "__main__":
main()
С википедией, погодой и вакансиями все просто, мы создаем экземпляр нужно класса, вызываем метод, который нам отдает все, что нужно. Об этим методах позже.
Сейчас мы видим что для конвертера присутствует немного логики. Она следующая — мы разбиваем полученную строку вида «RUB/EUR» на массив через метод split, записываем переменную с первой валютой и второй.
Получаем valute , оно возвращается сразу по отношению к рублю. Забегая вперед скажу что это из-за использования сайта https://www.cbr-xml-daily.ru/.
Получаем из сообщения боту то какую сумму мы хотим конвертировать и присваиваем в переменную sum.
Далее все просто. У нас только EUR, USD, RUB. Если рубль идет первым, значит мы хотим N кол-во рублей перевести в например евро, то получаем число с плавающей запятой, равное «введеная нами сумма» / «значение полученной в valute», если рубль идет второй парой то наоборот.
Говоря проще если в меню у нас «Перевести rub/eur», то введя 1000 рублей мы эту 1000 / на курс евро к рублю, например 1000 / 100 = 10 евро. Ну и также в обратном порядке. Тут все просто.
Давайте немного отвлечемся от сложных формул начальной школы и рассмотрим другие части нашего мини бота.
Вакансии DevOps с HH
import requests
class Vacancies():
hh_url = url = "https://api.hh.ru/vacancies"
params = {
'text': 'NAME:devops Кипр',
'area': 1,
'per_page': 100
}
def send_request(self, req_url, req_params):
req = requests.get(url=req_url, params=req_params)
data = req.content.decode()
data = req.json()
req.close()
return data
def get_vacancies(self):
found_cyprus_job = []
vacancies = self.send_request(self.hh_url, self.params)
for i in vacancies['items']:
job_name = i['name']
job_url = i['alternate_url']
salary = i['salary']
if salary == None:
salary = "не указана"
else:
salary_from = salary['from']
salary_to = salary['to']
currency = salary['currency']
gross = salary['gross']
if salary_from == None:
salary_from = "0"
elif salary_to == None:
salary_to = "0"
if gross == False:
gross = "до вычета"
elif gross == True:
gross = "на руки"
salary = f"от {salary_from} до {salary_to} {currency} {gross}"
jobs = f"Вакансия: {job_name}, ЗП: {salary}, URL: {job_url}"
found_cyprus_job.append(jobs)
message = "\n\n".join(map(str, found_cyprus_job))
return message
Ну, тут все предельно просто, отправляем запрос с параметрами на api hh.ru. На Кипре немного вакансий и нам не нужно делать пагинацию. После чего парсим полученный словарь дабы вывести это все в удобном виде. Выглядеть будет как-то так:
Погода
import requests
import json
import datetime
class Weather():
openweather_key = "<api_key>"
open_weather_url = "http://api.openweathermap.org/data/2.5/weather?q="
def __init__(self, city):
self.city = city
def get_current_weather(self):
full_url = f"{self.open_weather_url}{self.city}&appid={self.openweather_key}&units=metric"
req = requests.get(url=full_url)
weather_map = req.json()
current_weather = weather_map["main"]["temp"]
pressure = weather_map["main"]["pressure"]
humidity = weather_map["main"]["humidity"]
wind = weather_map["wind"]["speed"]
sunrise = datetime.datetime.fromtimestamp(weather_map["sys"]["sunrise"])
sunset = datetime.datetime.fromtimestamp(weather_map["sys"]["sunset"])
message = f"""Погода в городе: {self.city}\nТемпература: {current_weather}C°\nВлажность: {humidity}%\nДавление: {pressure} мм.рт.ст.\nВетер: {wind} м/с\nРассвет: {sunrise}\nЗакат: {sunset}"""
return message
Погоду получаем через запрос в апи сервиса openweather. Для этого нужно зарегистрироваться на их сайте и сгенерировать api ключ.
Все аналогично с hh.ru, получаем словарь в ответ на запрос, парсим его и выводим в удобном виде. Пример ответа:
Кипр конечно не город, но так тоже прокатывает). Можно вводить конкретный город или даже населенный пункт или деревню.
Википедия
import wikipedia, re
class Wiki():
wikipedia.set_lang("ru")
def __init__(self, message_for_wiki):
self.message_for_wiki = message_for_wiki
def get_wiki_info(self):
try:
wiki_reponse = wikipedia.page(self.message_for_wiki)
wiki_text = wiki_reponse.content[:1000]
wiki_found = wiki_text.split(".")
wiki_found = wiki_found[:-1]
wiki_result = ""
for item in wiki_found:
if not('==' in item):
if(len((item.strip()))>3):
wiki_result = wiki_result + item + "."
else:
break
wiki_result=re.sub('\([^()]*\)', '', wiki_result)
wiki_result=re.sub('\([^()]*\)', '', wiki_result)
wiki_result=re.sub('\{[^\{\}]*\}', '', wiki_result)
return wiki_result
except Exception as e:
return "Не найдено информации в Wikipedia"
Для википедии используем специальную либу, установив ее перед этим. Парсим результат, выводим до 1000 слов из полученного ответа.
Курсы валют
Мы не хотим на каждый запрос дергать апи, к тому же курсы валют обновляются примерно раз в сутки, хорошо было бы и нам получать раз в сутки свежие данные а остальное брать из кеша. Но ставить еще nginx или другое решение перед ботом немного оверхед для нас. Мы сделаем проще. Используем решение с сайта валют о котором писал выше.
#!/bin/sh
set -e
mkdir -p /var/cache/cbr
cd /var/cache/cbr
for file in daily_utf8.xml daily.xml daily_eng.xml daily_eng_utf8.xml daily_json.js latest.js
do wget --timestamping --no-verbose https://www.cbr-xml-daily.ru/$file
done 2>&1 | xargs -I{} logger --tag $0 --id=$$ "{}"
Раз в сутки по крону у нас будет запускаться скрипт выше. Он записывает в директорию /var/cache/cbr файлики с курсами валют в разных форматах. Это и будет наш «кеш».
В python реализовываем следующее
import json
class Exchange():
def __init__(self, currency):
self.currency = currency
self.currency_first = currency.split("/")[0]
self.currency_second = currency.split("/")[1]
def get_cached_rate_file(self):
with open('/var/cache/cbr/daily_json.js') as data_file:
data = json.load(data_file)
return data
def get_curret_rate(self):
data = self.get_cached_rate_file()
try:
valute = data['Valute'][self.currency_first]['Value']
except KeyError:
valute = data['Valute'][self.currency_second]['Value']
return valute
В методе get_cached_rate_file() получаем этот файл и возвращаем его в виде словаря.
В другом методе парсим, возвращая либо первую валюту из пары либо вторую.
Остальное уже описано в начале статьи.
Пример работы конвертера
Последним штрихом сделаем systemd unit файл, чтобы управлять запуском скрипта через systemd.
[Unit]
Description=Telegram bot
After=multi-user.target
[Service]
Type=simple
Restart=always
ExecStart=/usr/bin/python3.6 /path_to_script/main.py
[Install]
WantedBy=multi-user.target