Как правильно остановить бота telegram python?

[RuneCaster808]

Ребята, подскажите, как корректно остановить бота telegram python, чтобы он не зависал и закрывался как надо? Недавно начал ковыряться с телеги ботами, использую pyTelegramBotAPI. Всё работает норм, но при попытке завершения бота он как будто зависает, и приходится принудительно останавливать процесс. Какие есть варианты для graceful shutdown? Если кто может, киньте код для примера, буду очень признателен!

[admin]

Ждём вас в нашем чате в Телеграмм ==>> @pythoneer_chat

А ТАКЖЕ: Канал о Python, статьи и книги ==>> @pythoneer_ru

[Arwen]

Эй, привет! Ну самый простой способ - это использовать signal для перехвата сигнала завершения процесса. В python можно это сделать так:

import signal
import time
from telebot import TeleBot

bot = TeleBot('YOUR_TOKEN_HERE')

def graceful_shutdown(signum, frame):
print('Штатное завершение...')
bot.stop_polling()
exit(0)

signal.signal(signal.SIGTERM, graceful_shutdown)
signal.signal(signal.SIGINT, graceful_shutdown)

@bot.message_handler(commands=['start'])
def start_message(message):
bot.send_message(message.chat.id, 'Привет! Я бот.')

bot.polling(none_stop=True) 

Ну и вуаля, бот будет ловить сигналы и закрываться аккуратно.

[Caspian]

Эй, привет! Ну самый простой способ - это использовать signal для перехвата сигнала завершения процесса. В python можно это сделать так:

import signal
import time
from telebot import TeleBot

bot = TeleBot('YOUR_TOKEN_HERE')

def graceful_shutdown(signum, frame):
print('Штатное завершение...')
bot.stop_polling()
exit(0)

signal.signal(signal.SIGTERM, graceful_shutdown)
signal.signal(signal.SIGINT, graceful_shutdown)

@bot.message_handler(commands=['start'])
def start_message(message):
bot.send_message(message.chat.id, 'Привет! Я бот.')

bot.polling(none_stop=True) 

Ну и вуаля, бот будет ловить сигналы и закрываться аккуратно.

О, круто! Не знал, что signal можно так заюзать. Надо затестить!

[Эльфийка]

Альтернативка есть через try-except, если хочется классический метод. Вот так:

import time
from telebot import TeleBot

bot = TeleBot('YOUR_TOKEN_HERE')

def main():
try:
bot.polling(none_stop=True)
except KeyboardInterrupt:
print('Останавливаем бота...')
bot.stop_polling()

if __name__ == '__main__':
main() 

Не такой изящный как с signal, но тоже пашет.

[ДиопсидовыйВетер]

Альтернативка есть через try-except, если хочется классический метод. Вот так:

import time
from telebot import TeleBot

bot = TeleBot('YOUR_TOKEN_HERE')

def main():
try:
bot.polling(none_stop=True)
except KeyboardInterrupt:
print('Останавливаем бота...')
bot.stop_polling()

if __name__ == '__main__':
main() 

Не такой изящный как с signal, но тоже пашет.

Ну это тоже вариант, причем достаточно универсальный. Крутяк, респект!

[Валентина Павловна]

Блин, сигнал сигналом, но если ты используешь вебхуки, то можно просто сделать эндпоинт для стопа:

from telebot import TeleBot, types
from flask import Flask, request

app = Flask(__name__)
bot = TeleBot('YOUR_TOKEN_HERE')

@app.route('/stop', methods=['GET'])
def stop():
bot.remove_webhook()
return 'Бот остановлен', 200

@app.route('/YOUR_WEBHOOK_PATH', methods=['POST'])
def webhook():
bot.process_new_updates([types.Update.de_json(request.stream.read().decode("utf-8"))])
return 'Ok', 200

if __name__ == '__main__':
bot.set_webhook(url='https://YOUR_DOMAIN/YOUR_WEBHOOK_PATH')
app.run(port=5002) 

Достаточно просто закидывать запрос на /stop и всё, бот прекращает слушать апдейты.

[Эдуард]

Блин, сигнал сигналом, но если ты используешь вебхуки, то можно просто сделать эндпоинт для стопа:

from telebot import TeleBot, types
from flask import Flask, request

app = Flask(__name__)
bot = TeleBot('YOUR_TOKEN_HERE')

@app.route('/stop', methods=['GET'])
def stop():
bot.remove_webhook()
return 'Бот остановлен', 200

@app.route('/YOUR_WEBHOOK_PATH', methods=['POST'])
def webhook():
bot.process_new_updates([types.Update.de_json(request.stream.read().decode("utf-8"))])
return 'Ok', 200

if __name__ == '__main__':
bot.set_webhook(url='https://YOUR_DOMAIN/YOUR_WEBHOOK_PATH')
app.run(port=5002) 

Достаточно просто закидывать запрос на /stop и всё, бот прекращает слушать апдейты.

Хм, вебхуки вообще редко использую, но инфа реально годная! Надо попробовать.

[Arthur]

И ещё вариант с асинхронкой через aiogram, там вообще красота с asyncio:

import asyncio
from aiogram import Bot, Dispatcher, types

API_TOKEN = 'YOUR_TOKEN_HERE'

bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)

@dp.message_handler(commands=['start'])
async def start_message(msg: types.Message):
await msg.answer('Привет! Я асинхронный бот.')

async def on_shutdown(dp):
await bot.close()
await dp.storage.close()

if __name__ == '__main__':
from aiogram import executor
loop = asyncio.get_event_loop()
try:
executor.start_polling(dp, on_shutdown=on_shutdown)
except (KeyboardInterrupt, SystemExit):
loop.run_until_complete(on_shutdown(dp)) 

Асинхронные вещички для тех, кто любит помудрить с async/await!

[Юлия Игоревна]

И ещё вариант с асинхронкой через aiogram, там вообще красота с asyncio:

import asyncio
from aiogram import Bot, Dispatcher, types

API_TOKEN = 'YOUR_TOKEN_HERE'

bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)

@dp.message_handler(commands=['start'])
async def start_message(msg: types.Message):
await msg.answer('Привет! Я асинхронный бот.')

async def on_shutdown(dp):
await bot.close()
await dp.storage.close()

if __name__ == '__main__':
from aiogram import executor
loop = asyncio.get_event_loop()
try:
executor.start_polling(dp, on_shutdown=on_shutdown)
except (KeyboardInterrupt, SystemExit):
loop.run_until_complete(on_shutdown(dp)) 

Асинхронные вещички для тех, кто любит помудрить с async/await!

Оооо, async рулят! aiogram вообще тема, уважаю! Надо будет чекнуть, как там с работой после остановок.