mirror/pyTelegramBotAPI
1
0
Fork 0
mirror of https://github.com/eternnoir/pyTelegramBotAPI.git synced 2023-08-10 21:12:57 +03:00
Code Releases Activity

Merge pull request #1612 from coder2020official/newfeatures

Browse Source 
run_webhooks() built in function to listen and process webhook requests.
This commit is contained in:
Badiboy 2022-07-11 23:33:39 +03:00 committed by GitHub
commit 1c11898ea1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 491 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
examples/asynchronous_telebot/webhooks/run_webhooks.py Normal file
View File

@ -0,0 +1,45 @@
#!/usr/bin/python
# This is a simple echo bot using the decorator mechanism.
# It echoes any incoming text messages.
# Example on built-in function to receive and process webhooks.
from telebot.async_telebot import AsyncTeleBot
import asyncio
bot = AsyncTeleBot('TOKEN')
WEBHOOK_SSL_CERT = './webhook_cert.pem' # Path to the ssl certificate
WEBHOOK_SSL_PRIV = './webhook_pkey.pem' # Path to the ssl private key
DOMAIN = '1.2.3.4' # either domain, or ip address of vps
# Quick'n'dirty SSL certificate generation:
#
# openssl genrsa -out webhook_pkey.pem 2048
# openssl req -new -x509 -days 3650 -key webhook_pkey.pem -out webhook_cert.pem
#
# When asked for "Common Name (e.g. server FQDN or YOUR name)" you should reply
# with the same value in you put in WEBHOOK_HOST
# Handle '/start' and '/help'
@bot.message_handler(commands=['help', 'start'])
async def send_welcome(message):
await bot.reply_to(message, """\
Hi there, I am EchoBot.
I am here to echo your kind words back to you. Just say anything nice and I'll say the exact same thing to you!\
""")
# Handle all other messages with content_type 'text' (content_types defaults to ['text'])
@bot.message_handler(func=lambda message: True)
async def echo_message(message):
await bot.reply_to(message, message.text)
# it uses fastapi + uvicorn
asyncio.run(bot.run_webhooks(
listen=DOMAIN,
certificate=WEBHOOK_SSL_CERT,
certificate_key=WEBHOOK_SSL_PRIV
))

45
examples/webhook_examples/run_webhooks.py Normal file
View File

@ -0,0 +1,45 @@
#!/usr/bin/python
# This is a simple echo bot using the decorator mechanism.
# It echoes any incoming text messages.
# Example on built-in function to receive and process webhooks.
import telebot
API_TOKEN = 'TOKEN'
bot = telebot.TeleBot(API_TOKEN)
WEBHOOK_SSL_CERT = './webhook_cert.pem' # Path to the ssl certificate
WEBHOOK_SSL_PRIV = './webhook_pkey.pem' # Path to the ssl private key
DOMAIN = '1.2.3.4' # either domain, or ip address of vps
# Quick'n'dirty SSL certificate generation:
#
# openssl genrsa -out webhook_pkey.pem 2048
# openssl req -new -x509 -days 3650 -key webhook_pkey.pem -out webhook_cert.pem
#
# When asked for "Common Name (e.g. server FQDN or YOUR name)" you should reply
# with the same value in you put in WEBHOOK_HOST
# Handle '/start' and '/help'
@bot.message_handler(commands=['help', 'start'])
def send_welcome(message):
bot.reply_to(message, """\
Hi there, I am EchoBot.
I am here to echo your kind words back to you. Just say anything nice and I'll say the exact same thing to you!\
""")
# Handle all other messages with content_type 'text' (content_types defaults to ['text'])
@bot.message_handler(func=lambda message: True)
def echo_message(message):
bot.reply_to(message, message.text)
bot.run_webhooks(
listen=DOMAIN,
certificate=WEBHOOK_SSL_CERT,
certificate_key=WEBHOOK_SSL_PRIV
)

3
setup.py
View File

@ -27,7 +27,10 @@ setup(name='pyTelegramBotAPI',
'json': 'ujson',
'PIL': 'Pillow',
'redis': 'redis>=3.4.1',
'aioredis': 'aioredis',
'aiohttp': 'aiohttp',
'fastapi': 'fastapi',
'uvicorn': 'uvicorn',
},
classifiers=[
'Development Status :: 5 - Production/Stable',

87
telebot/__init__.py
View File

@ -18,6 +18,12 @@ import telebot.types
from telebot.storage import StatePickleStorage, StateMemoryStorage
# random module to generate random string
import random
import string
import ssl
logger = logging.getLogger('TeleBot')
@ -293,6 +299,87 @@ class TeleBot:
return apihelper.set_webhook(self.token, url, certificate, max_connections, allowed_updates, ip_address,
drop_pending_updates, timeout, secret_token)
def run_webhooks(self,
listen: Optional[str]="127.0.0.1",
port: Optional[int]=443,
url_path: Optional[str]=None,
certificate: Optional[str]=None,
certificate_key: Optional[str]=None,
webhook_url: Optional[str]=None,
max_connections: Optional[int]=None,
allowed_updates: Optional[List]=None,
ip_address: Optional[str]=None,
drop_pending_updates: Optional[bool] = None,
timeout: Optional[int]=None,
secret_token: Optional[str]=None,
secret_token_length: Optional[int]=20,
debug: Optional[bool]=False):
"""
This class sets webhooks and listens to a given url and port.
:param listen: IP address to listen to. Defaults to
0.0.0.0
:param port: A port which will be used to listen to webhooks.
:param url_path: Path to the webhook. Defaults to /token
:param certificate: Path to the certificate file.
:param certificate_key: Path to the certificate key file.
:param webhook_url: Webhook URL.
:param max_connections: Maximum allowed number of simultaneous HTTPS connections to the webhook for update delivery, 1-100. Defaults to 40. Use lower values to limit the load on your bot's server, and higher values to increase your bot's throughput.
:param allowed_updates: A JSON-serialized list of the update types you want your bot to receive. For example, specify [“message”, “edited_channel_post”, “callback_query”] to only receive updates of these types. See Update for a complete list of available update types. Specify an empty list to receive all updates regardless of type (default). If not specified, the previous setting will be used.
:param ip_address: The fixed IP address which will be used to send webhook requests instead of the IP address resolved through DNS
:param drop_pending_updates: Pass True to drop all pending updates
:param timeout: Integer. Request connection timeout
:param secret_token: Secret token to be used to verify the webhook request.
:return:
"""
# generate secret token if not set
if not secret_token:
secret_token = ''.join(random.choices(string.ascii_uppercase + string.digits, k=secret_token_length))
if not url_path:
url_path = self.token + '/'
if url_path[-1] != '/': url_path += '/'
protocol = "https" if certificate else "http"
if not webhook_url:
webhook_url = "{}://{}:{}/{}".format(protocol, listen, port, url_path)
if certificate and certificate_key:
ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_ctx.load_cert_chain(certificate, certificate_key)
else:
ssl_ctx = None
# open certificate if it exists
cert_file = open(certificate, 'rb') if certificate else None
self.set_webhook(
url=webhook_url,
certificate=cert_file,
max_connections=max_connections,
allowed_updates=allowed_updates,
ip_address=ip_address,
drop_pending_updates=drop_pending_updates,
timeout=timeout,
secret_token=secret_token
)
if cert_file: cert_file.close()
ssl_context = (certificate, certificate_key) if certificate else (None, None)
# webhooks module
try:
from telebot.ext.sync import SyncWebhookListener
except (NameError, ImportError):
raise ImportError("Please install uvicorn and fastapi in order to use `run_webhooks` method.")
self.webhook_listener = SyncWebhookListener(self, secret_token, listen, port, ssl_context, '/'+url_path, debug)
self.webhook_listener.run_app()
def delete_webhook(self, drop_pending_updates=None, timeout=None):
"""
Use this method to remove webhook integration if you decide to switch back to getUpdates.

84
telebot/async_telebot.py
View File

@ -30,6 +30,11 @@ REPLY_MARKUP_TYPES = Union[
types.ReplyKeyboardRemove, types.ForceReply]
import string
import random
import ssl
"""
Module : telebot
"""
@ -1438,6 +1443,85 @@ class AsyncTeleBot:
drop_pending_updates, timeout, secret_token)
async def run_webhooks(self,
listen: Optional[str]="127.0.0.1",
port: Optional[int]=443,
url_path: Optional[str]=None,
certificate: Optional[str]=None,
certificate_key: Optional[str]=None,
webhook_url: Optional[str]=None,
max_connections: Optional[int]=None,
allowed_updates: Optional[List]=None,
ip_address: Optional[str]=None,
drop_pending_updates: Optional[bool] = None,
timeout: Optional[int]=None,
secret_token: Optional[str]=None,
secret_token_length: Optional[int]=20,
debug: Optional[bool]=False):
"""
This class sets webhooks and listens to a given url and port.
:param listen: IP address to listen to. Defaults to
0.0.0.0
:param port: A port which will be used to listen to webhooks.
:param url_path: Path to the webhook. Defaults to /token
:param certificate: Path to the certificate file.
:param certificate_key: Path to the certificate key file.
:param webhook_url: Webhook URL.
:param max_connections: Maximum allowed number of simultaneous HTTPS connections to the webhook for update delivery, 1-100. Defaults to 40. Use lower values to limit the load on your bot's server, and higher values to increase your bot's throughput.
:param allowed_updates: A JSON-serialized list of the update types you want your bot to receive. For example, specify [“message”, “edited_channel_post”, “callback_query”] to only receive updates of these types. See Update for a complete list of available update types. Specify an empty list to receive all updates regardless of type (default). If not specified, the previous setting will be used.
:param ip_address: The fixed IP address which will be used to send webhook requests instead of the IP address resolved through DNS
:param drop_pending_updates: Pass True to drop all pending updates
:param timeout: Integer. Request connection timeout
:param secret_token: Secret token to be used to verify the webhook request.
:return:
"""
# generate secret token if not set
if not secret_token:
secret_token = ''.join(random.choices(string.ascii_uppercase + string.digits, k=secret_token_length))
if not url_path:
url_path = self.token + '/'
if url_path[-1] != '/': url_path += '/'
protocol = "https" if certificate else "http"
if not webhook_url:
webhook_url = "{}://{}:{}/{}".format(protocol, listen, port, url_path)
if certificate and certificate_key:
ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_ctx.load_cert_chain(certificate, certificate_key)
else:
ssl_ctx = None
# open certificate if it exists
cert_file = open(certificate, 'rb') if certificate else None
await self.set_webhook(
url=webhook_url,
certificate=cert_file,
max_connections=max_connections,
allowed_updates=allowed_updates,
ip_address=ip_address,
drop_pending_updates=drop_pending_updates,
timeout=timeout,
secret_token=secret_token
)
if cert_file: cert_file.close()
ssl_context = (certificate, certificate_key) if certificate else (None, None)
# for webhooks
try:
from telebot.ext.aio import AsyncWebhookListener
except (NameError, ImportError):
raise ImportError("Please install uvicorn and fastapi in order to use `run_webhooks` method.")
self.webhook_listener = AsyncWebhookListener(self, secret_token, listen, port, ssl_context, '/'+url_path, debug)
await self.webhook_listener.run_app()
async def delete_webhook(self, drop_pending_updates=None, timeout=None):
"""

3
telebot/ext/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""
A folder with asynchronous and synchronous extensions.
"""

10
telebot/ext/aio/__init__.py Normal file
View File

@ -0,0 +1,10 @@
"""
A folder with all the async extensions.
"""
from .webhooks import AsyncWebhookListener
__all__ = [
"AsyncWebhookListener"
]

103
telebot/ext/aio/webhooks.py Normal file
View File

@ -0,0 +1,103 @@
"""
This file is used by AsyncTeleBot.run_webhooks() function.
Fastapi and starlette(0.20.2+) libraries are required to run this script.
"""
# modules required for running this script
fastapi_installed = True
try:
import fastapi
from fastapi.responses import JSONResponse
from fastapi.requests import Request
from uvicorn import Server, Config
except ImportError:
fastapi_installed = False
import asyncio
from telebot.types import Update
from typing import Optional
class AsyncWebhookListener:
def __init__(self, bot,
secret_token: str, host: Optional[str]="127.0.0.1",
port: Optional[int]=443,
ssl_context: Optional[tuple]=None,
url_path: Optional[str]=None,
debug: Optional[bool]=False
) -> None:
"""
Aynchronous implementation of webhook listener
for asynchronous version of telebot.
:param bot: TeleBot instance
:param secret_token: Telegram secret token
:param host: Webhook host
:param port: Webhook port
:param ssl_context: SSL context
:param url_path: Webhook url path
:param debug: Debug mode
"""
self._check_dependencies()
self.app = fastapi.FastAPI()
self._secret_token = secret_token
self._bot = bot
self._port = port
self._host = host
self._ssl_context = ssl_context
self._url_path = url_path
self._debug = debug
self._prepare_endpoint_urls()
def _check_dependencies(self):
if not fastapi_installed:
raise ImportError('Fastapi or uvicorn is not installed. Please install it via pip.')
import starlette
if starlette.__version__ < '0.20.2':
raise ImportError('Starlette version is too old. Please upgrade it: `pip3 install starlette -U`')
return
def _prepare_endpoint_urls(self):
self.app.add_api_route(endpoint=self.process_update,path= self._url_path, methods=["POST"])
async def process_update(self, request: Request, update: dict):
"""
Processes updates.
"""
# header containsX-Telegram-Bot-Api-Secret-Token
if request.headers.get('X-Telegram-Bot-Api-Secret-Token') != self._secret_token:
# secret token didn't match
return JSONResponse(status_code=403, content={"error": "Forbidden"})
if request.headers.get('content-type') == 'application/json':
json_string = update
asyncio.create_task(self._bot.process_new_updates([Update.de_json(json_string)]))
return JSONResponse('', status_code=200)
return JSONResponse(status_code=403, content={"error": "Forbidden"})
async def run_app(self):
"""
Run app with the given parameters.
"""
config = Config(app=self.app,
host=self._host,
port=self._port,
debug=self._debug,
ssl_certfile=self._ssl_context[0],
ssl_keyfile=self._ssl_context[1]
)
server = Server(config)
await server.serve()
await self._bot.close_session()

10
telebot/ext/sync/__init__.py Normal file
View File

@ -0,0 +1,10 @@
"""
A folder with all the sync extensions.
"""
from .webhooks import SyncWebhookListener
__all__ = [
"SyncWebhookListener"
]

101
telebot/ext/sync/webhooks.py Normal file
View File

@ -0,0 +1,101 @@
"""
This file is used by TeleBot.run_webhooks() &
AsyncTeleBot.run_webhooks() functions.
Flask/fastapi is required to run this script.
"""
# modules required for running this script
fastapi_installed = True
try:
import fastapi
from fastapi.responses import JSONResponse
from fastapi.requests import Request
import uvicorn
except ImportError:
fastapi_installed = False
from telebot.types import Update
from typing import Optional
class SyncWebhookListener:
def __init__(self, bot,
secret_token: str, host: Optional[str]="127.0.0.1",
port: Optional[int]=443,
ssl_context: Optional[tuple]=None,
url_path: Optional[str]=None,
debug: Optional[bool]=False
) -> None:
"""
Synchronous implementation of webhook listener
for synchronous version of telebot.
:param bot: TeleBot instance
:param secret_token: Telegram secret token
:param host: Webhook host
:param port: Webhook port
:param ssl_context: SSL context
:param url_path: Webhook url path
:param debug: Debug mode
"""
self._check_dependencies()
self.app = fastapi.FastAPI()
self._secret_token = secret_token
self._bot = bot
self._port = port
self._host = host
self._ssl_context = ssl_context
self._url_path = url_path
self._debug = debug
self._prepare_endpoint_urls()
def _check_dependencies(self):
if not fastapi_installed:
raise ImportError('Fastapi or uvicorn is not installed. Please install it via pip.')
import starlette
if starlette.__version__ < '0.20.2':
raise ImportError('Starlette version is too old. Please upgrade it: `pip3 install starlette -U`')
return
def _prepare_endpoint_urls(self):
self.app.add_api_route(endpoint=self.process_update,path= self._url_path, methods=["POST"])
def process_update(self, request: Request, update: dict):
"""
Processes updates.
"""
# header containsX-Telegram-Bot-Api-Secret-Token
if request.headers.get('X-Telegram-Bot-Api-Secret-Token') != self._secret_token:
# secret token didn't match
return JSONResponse(status_code=403, content={"error": "Forbidden"})
if request.headers.get('content-type') == 'application/json':
self._bot.process_new_updates([Update.de_json(update)])
return JSONResponse('', status_code=200)
return JSONResponse(status_code=403, content={"error": "Forbidden"})
def run_app(self):
"""
Run app with the given parameters.
"""
uvicorn.run(app=self.app,
host=self._host,
port=self._port,
debug=self._debug,
ssl_certfile=self._ssl_context[0],
ssl_keyfile=self._ssl_context[1]
)