(20241206) Sending the 'Subject' in the mails list now.
This commit is contained in:
@@ -0,0 +1,391 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Friday, 6th Dec., 2024
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a way to send and receive messages and media via Telegram asynchronously.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# System-level activities:
|
||||
import io
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import json
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
# Data models:
|
||||
from utils_v2.telegram.models.data.api_call import TelegramApiResponse
|
||||
|
||||
# To make API calls:
|
||||
import httpx
|
||||
|
||||
# To work with date and time:
|
||||
import datetime
|
||||
|
||||
# For working with datatypes:
|
||||
from typing import Literal, List
|
||||
|
||||
# For debugging:
|
||||
from icecream import IceCreamDebugger
|
||||
import inspect
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** CLASSES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class AsyncTelegramBot:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
bot_token: str,
|
||||
http_client: httpx.AsyncClient = None,
|
||||
debug = True,
|
||||
debug_prefix = "TGram | ",
|
||||
debug_only_errors = True
|
||||
):
|
||||
|
||||
"""
|
||||
To initialize the instance of this Telegram messenger.
|
||||
:param bot_token: The token granted by BotFather.
|
||||
:param http_client: An instance of 'httpx' library's AsyncClient. If not given, one will be instantiated
|
||||
internally. It is recommended that, for multi-bot use cases, you provide a common HTTP client from outside.
|
||||
:param debug: Whether, or not, you would like to show debugging messages on the terminal.
|
||||
:param debug_prefix: The prefix string to identify the debugging messages.
|
||||
:param debug_only_errors: Whether you would like to show all debugging messages or just error messages.
|
||||
"""
|
||||
|
||||
# Prepare the debugging utility:
|
||||
self._debug_prefix = debug_prefix
|
||||
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
|
||||
if not debug: self._printer.disable()
|
||||
self._debug_only_errors = debug_only_errors
|
||||
|
||||
# Accept the input configuration:
|
||||
self.__bot_token = bot_token
|
||||
|
||||
# Accept/create an HTTP client to work with:
|
||||
if http_client: self.__http_client = http_client
|
||||
else: self.__http_client = httpx.AsyncClient(
|
||||
limits = httpx.Limits(
|
||||
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
|
||||
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
|
||||
),
|
||||
timeout = httpx.Timeout(
|
||||
pool = 120.0, # .... Time to wait for a free connection from the pool.
|
||||
connect = 5.0, # ... Time to wait for establishing a connection to the server.
|
||||
write = 10.0, # .... Time to wait for sending data.
|
||||
read = 120.0 # ..... Time to wait for receiving data.
|
||||
)
|
||||
)
|
||||
|
||||
def enable_debug(self):
|
||||
self._printer.enable()
|
||||
|
||||
def disable_debug(self):
|
||||
self._printer.disable()
|
||||
|
||||
def debug_only_errors(self):
|
||||
self._debug_only_errors = True
|
||||
|
||||
def debug_everything(self):
|
||||
self._debug_only_errors = False
|
||||
|
||||
# ┏┓┏┓┳ ┏┓ ┓┓•
|
||||
# ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓
|
||||
# ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
async def __get(
|
||||
self,
|
||||
url: str,
|
||||
headers: dict = None,
|
||||
params: dict = None
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
"""
|
||||
To call an API using the GET method.
|
||||
:param url: The URL to call.
|
||||
:param headers: The headers to pass.
|
||||
:param params: The params to send in the query string itself.
|
||||
:return: A structured response that includes the raw response, the exception (if any), and so on.
|
||||
"""
|
||||
|
||||
# Prepare the structure of the response:
|
||||
api_response = TelegramApiResponse(
|
||||
action = inspect.stack()[1].function,
|
||||
url = url,
|
||||
method = "GET"
|
||||
)
|
||||
|
||||
try:
|
||||
|
||||
# Make the API call:
|
||||
response = await self.__http_client.get(
|
||||
url = url,
|
||||
headers = headers,
|
||||
params = params
|
||||
)
|
||||
|
||||
# Note down the results:
|
||||
api_response.response = response
|
||||
api_response.httpCode = response.status_code
|
||||
api_response.message = response.reason_phrase
|
||||
|
||||
# If something goes wrong:
|
||||
except Exception as exception:
|
||||
api_response.exception = exception
|
||||
api_response.message = str(exception)
|
||||
self._printer(exception, api_response.url, api_response.method, headers, params)
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
async def __post(
|
||||
self,
|
||||
url: str,
|
||||
headers: dict = None,
|
||||
json: dict = None,
|
||||
data: dict = None,
|
||||
content: str | bytes = None
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
"""
|
||||
To call an API using the POST method.
|
||||
:param url: The URL to call.
|
||||
:param headers: The headers to pass.
|
||||
:param json: The params to send in the JSON body.
|
||||
:param data: The params to send in the form-data in the body.
|
||||
:param content: The raw content to be sent in the body (typically as an octet-stream).
|
||||
:return: A structured response that includes the raw response, the exception (if any), and so on.
|
||||
"""
|
||||
|
||||
# Prepare the structure of the response:
|
||||
api_response = TelegramApiResponse(
|
||||
action = inspect.stack()[1].function,
|
||||
url = url,
|
||||
method = "POST"
|
||||
)
|
||||
|
||||
try:
|
||||
|
||||
# Make the API call:
|
||||
response = await self.__http_client.post(
|
||||
url = url,
|
||||
headers = headers,
|
||||
json = json,
|
||||
data = data,
|
||||
content = content
|
||||
)
|
||||
|
||||
# Note down the results:
|
||||
api_response.response = response
|
||||
api_response.httpCode = response.status_code
|
||||
api_response.message = response.reason_phrase
|
||||
|
||||
# If something goes wrong:
|
||||
except Exception as exception:
|
||||
api_response.exception = exception
|
||||
api_response.message = str(exception)
|
||||
self._printer(exception, api_response.url, api_response.method, headers, json, data)
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
# ┳┓ ┳ ┏
|
||||
# ┣┫┏┓╋ ┃┏┓╋┏┓
|
||||
# ┻┛┗┛┗ ┻┛┗┛┗┛
|
||||
|
||||
async def bot_info(self) -> TelegramApiResponse:
|
||||
|
||||
"""
|
||||
To get a bot's basic info.
|
||||
:return: The structured response with the JSON response from Telegram in the 'data' field.
|
||||
"""
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__get(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/getMe",
|
||||
)
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode in [200]:
|
||||
api_json = await api_response.get_json()
|
||||
if api_json["ok"]:
|
||||
api_result = api_json["result"]
|
||||
api_response.success = True
|
||||
api_response.data = {
|
||||
"id": api_result["id"],
|
||||
"displayName": api_result["first_name"],
|
||||
"username": api_result["username"],
|
||||
}
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
# ┓ ┏ ┓ ┓ ┓
|
||||
# ┃┃┃┏┓┣┓┣┓┏┓┏┓┃┏
|
||||
# ┗┻┛┗ ┗┛┛┗┗┛┗┛┛┗
|
||||
|
||||
async def set_webhook(
|
||||
self,
|
||||
webhook_url: str
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__post(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/setWebhook",
|
||||
data = {"url": webhook_url}
|
||||
)
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode in [200, 400]:
|
||||
api_json = await api_response.get_json()
|
||||
api_response.message = api_json["description"]
|
||||
api_response.data = api_json
|
||||
if api_json["ok"]:
|
||||
api_response.success = True
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
async def delete_webhook(
|
||||
self,
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__get(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/deleteWebhook",
|
||||
)
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode in [200, 400]:
|
||||
api_json = await api_response.get_json()
|
||||
api_response.message = api_json["description"]
|
||||
api_response.data = api_json
|
||||
if api_json["ok"]:
|
||||
api_response.success = True
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
async def get_updates(
|
||||
self,
|
||||
update_id: int = None,
|
||||
limit: int = 50,
|
||||
timeout: int | float = 30.0
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__get(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/getUpdates",
|
||||
params = {
|
||||
"offset": update_id,
|
||||
"limit": limit,
|
||||
"timeout": int(timeout)
|
||||
}
|
||||
)
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode in [200, 400]:
|
||||
api_json = await api_response.get_json()
|
||||
# api_response.message = api_json["description"]
|
||||
api_response.data = api_json
|
||||
# if api_json["ok"]:
|
||||
# api_response.success = True
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
import asyncio
|
||||
|
||||
async def main():
|
||||
|
||||
# Create the instance and define needed variables:
|
||||
my_tg = AsyncTelegramBot(bot_token = r"7003670393:AAH9qF6XGqa-u2_TM2JCUkP0Fp48kMw8Ka8")
|
||||
recipient_chat_id = ""
|
||||
|
||||
# Test out the services:
|
||||
# tg_response = await my_tg.bot_info()
|
||||
# tg_response = await my_tg.set_webhook(webhook_url = r"https://api.thecaoffice.com/converse/chat/webhook/123")
|
||||
# tg_response = await my_tg.delete_webhook()
|
||||
tg_response = await my_tg.get_updates(update_id = 946018731, limit = 5)
|
||||
|
||||
# Show the response:
|
||||
print("SUMMARY:", tg_response.to_markdown(), "\n\n---\n\n")
|
||||
if tg_response.success: print("DATA:", json.to_string(tg_response.data, default = str))
|
||||
else: print("JSON:", json.to_string(await tg_response.get_json(), default = str))
|
||||
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user