Merge commit '5a0667beaf2d7a57f2f407d03c8ed583140a7c36' as 'utils_v2'
This commit is contained in:
@@ -0,0 +1,658 @@
|
||||
"""
|
||||
|
||||
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:
|
||||
|
||||
1. Official API Documentation: https://core.telegram.org/bots/api
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# System-level activities:
|
||||
import io
|
||||
import os
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import json
|
||||
from utils_v2.system import files
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
# Data models:
|
||||
from utils_v2.telegram.models.data.api_call import TelegramApiResponse
|
||||
from utils_v2.telegram.models.data.update import TelegramUpdate
|
||||
|
||||
# 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,
|
||||
params: dict = None,
|
||||
content: str | bytes = None,
|
||||
files: dict = 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 params: The params to send in the query string itself.
|
||||
:param content: The raw content to be sent in the body (typically as an octet-stream).
|
||||
:param files: Any file that you may want to send.
|
||||
: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,
|
||||
params = params,
|
||||
content = content,
|
||||
files = files
|
||||
)
|
||||
|
||||
# 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:
|
||||
|
||||
"""
|
||||
To set up a webhook to start receiving updates from Telegram. Use this same method to update an existing webhook
|
||||
without having to call the 'delete_webhook' method first.
|
||||
NOTE: WHEN A WEBHOOK IS SET, YOU CANNOT MANUALLY PULL UPDATES USING THE 'get_updates' METHOD.
|
||||
:param webhook_url: The URL on which Telegram must send you updates for this bot.
|
||||
:return: A standard response structure.
|
||||
"""
|
||||
|
||||
# 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:
|
||||
|
||||
"""
|
||||
To stop receiving updates for this bot via the webhook. You can call this at any time even if no webhook is
|
||||
active, and you just want to be sure.
|
||||
NOTE: WHEN A WEBHOOK IS SET, YOU CANNOT MANUALLY PULL UPDATES USING THE 'get_updates' METHOD.
|
||||
:return: A standard response structure.
|
||||
"""
|
||||
|
||||
# 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:
|
||||
|
||||
"""
|
||||
To manually pull updates from Telegram.
|
||||
NOTE: THIS DOES NOT WORK WHILE A WEBHOOK IS SET UP. YOU CAN EITHER USE THIS OR A WEBHOOK, NOT BOTH
|
||||
SIMULTANEOUSLY.
|
||||
:param update_id: The starting update id from where you want to start fetching updates. Leave this as None to
|
||||
fetch all available updates. Useful for when the webhook goes down and updates get missed.
|
||||
:param limit: The no. of updates to fetch.
|
||||
:param timeout: The max. timeout in seconds for long-polling.
|
||||
:return: A standard response structure with a list of 'TelegramUpdate's in the 'data' field.
|
||||
"""
|
||||
|
||||
# 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.data = api_json
|
||||
if api_json["ok"]:
|
||||
api_response.success = True
|
||||
api_response.data = [TelegramUpdate(**r) for r in api_json["result"]]
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
async def get_file(
|
||||
self,
|
||||
file_id: str,
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
"""
|
||||
To download a file from Telegram.
|
||||
NOTE: THIS DOES NOT WORK WHILE A WEBHOOK IS SET UP. YOU CAN EITHER USE THIS OR A WEBHOOK, NOT BOTH
|
||||
SIMULTANEOUSLY.
|
||||
:param file_id: The 'file_id' of this file. Not to be confused with 'file_unique_id'.
|
||||
:return: A standard response structure with the file's content in the 'data' field.
|
||||
"""
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__get(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/getFile",
|
||||
params = {"file_id": file_id}
|
||||
)
|
||||
|
||||
# If the call failed:
|
||||
if api_response.httpCode == 400:
|
||||
api_json = await api_response.get_json()
|
||||
api_response.message = api_json["description"]
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode == 200:
|
||||
api_json = await api_response.get_json()
|
||||
file_path = api_json["result"]["file_path"]
|
||||
api_response = await self.__get(url = f"https://api.telegram.org/file/bot{self.__bot_token}/{file_path}")
|
||||
if api_response.httpCode == 200:
|
||||
api_response.success = True
|
||||
api_response.data = await api_response.get_content()
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
# ┏┓ ┓•
|
||||
# ┗┓┏┓┏┓┏┫┓┏┓┏┓
|
||||
# ┗┛┗ ┛┗┗┻┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
@staticmethod
|
||||
def escape_special_chars(
|
||||
text: str,
|
||||
parse_mode: Literal[None, "Markdown", "MarkdownV2", "HTML"] = None
|
||||
) -> str:
|
||||
|
||||
r"""
|
||||
Escapes certain special chars for each type of parsing that is available. There's not much use of this function,
|
||||
though, TBH. This can be useful when trying to parse more complex thing like links.
|
||||
USE CASE: Suppose you want to send a message "Equation: 5 * 2 = 10" such that the actual equation part
|
||||
("5 * 2 = 10") is in bold, and you want this through, say, MarkdownV2 formatting. You could achieve this by
|
||||
first passing the part that you want to format through this function and then inserting it in the rest. Maybe
|
||||
like this:
|
||||
>> escaped_message = f"Equation: *{escape_special_chars(text = '5 * 2 = 10', parse_mode = 'MarkdownV2')}*"
|
||||
Or you could just manually achieve the same by typing:
|
||||
>> escaped_message = r"Equation: *5 \* 2 \= 10*"
|
||||
:param text: The raw string.
|
||||
:param parse_mode: The kind of parsing that is desired.
|
||||
:return: The escaped string.
|
||||
"""
|
||||
|
||||
# For MarkdownV2:
|
||||
if parse_mode == "Markdown" or parse_mode == "MarkdownV2":
|
||||
for k, v in {
|
||||
"*": "\\*",
|
||||
"`": "\\`",
|
||||
"[": "\\[",
|
||||
"]": "\\]",
|
||||
"(": "\\(",
|
||||
")": "\\)",
|
||||
"~": "\\~",
|
||||
">": "\\>",
|
||||
"#": "\\#",
|
||||
"+": "\\+",
|
||||
"-": "\\-",
|
||||
".": "\\.",
|
||||
"!": "\\!",
|
||||
"=": "\\=",
|
||||
}.items(): text = text.replace(k, v)
|
||||
|
||||
# For HTML:
|
||||
elif parse_mode == "HTML":
|
||||
for k, v in {
|
||||
"&": "&",
|
||||
"<": "<",
|
||||
">": ">",
|
||||
"\"": """,
|
||||
"'": "'"
|
||||
}.items(): text = text.replace(k, v)
|
||||
|
||||
# Done here:
|
||||
return text
|
||||
|
||||
async def send_message(
|
||||
self,
|
||||
chat_id: int | str,
|
||||
message: str,
|
||||
parse_mode: Literal[None, "Markdown", "MarkdownV2", "HTML"] = None,
|
||||
disable_notification: bool = False,
|
||||
protect_content: bool = False
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
"""
|
||||
To send one message to one recipient using a specified parse mode.
|
||||
:param chat_id: The id of the recipient user/group/channel.
|
||||
:param message: The actual message that you want to send.
|
||||
:param parse_mode: The parsing mode for adding visual formatting.
|
||||
:param disable_notification: Use this when you want to send an alert without causing the recipient's device to
|
||||
show a notification alert.
|
||||
:param protect_content: Use this to prevent recipient participants from copying/forwarding messages.
|
||||
:return: A standard response structure with an instance of 'TelegramUpdate' in the 'data' field.
|
||||
"""
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__get(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/sendMessage",
|
||||
params = {
|
||||
"chat_id": chat_id,
|
||||
"text": message,
|
||||
"parse_mode": parse_mode,
|
||||
"disable_notification": disable_notification,
|
||||
"protect_content": protect_content
|
||||
}
|
||||
)
|
||||
|
||||
# If the call failed:
|
||||
if api_response.httpCode == 400:
|
||||
api_json = await api_response.get_json()
|
||||
api_response.message = api_json["description"]
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode == 200:
|
||||
api_json = await api_response.get_json()
|
||||
if api_json["ok"]:
|
||||
api_response.success = True
|
||||
api_response.data = TelegramUpdate(message = api_json["result"])
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
async def send_photo(
|
||||
self,
|
||||
chat_id: int | str,
|
||||
file: str | io.BytesIO,
|
||||
file_name: str = None,
|
||||
caption: str = None,
|
||||
parse_mode: Literal[None, "Markdown", "MarkdownV2", "HTML"] = None,
|
||||
hide: bool = False,
|
||||
disable_notification: bool = False,
|
||||
protect_content: bool = False
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
"""
|
||||
To send one photo from the bot to one recipient. A caption with a specified parsing mode may also be added.
|
||||
Note that you must send at least one of the three file options.
|
||||
:param chat_id: The id of the recipient user/group/channel.
|
||||
:param file: Could be either a file on the disk, or a file held in a buffer in RAM or a 'file_id' of a file
|
||||
already on Telegram's server, or a URL of a file hosted on the internet.
|
||||
:param file_name: The name of the file.
|
||||
:param caption: The text to write under the file.
|
||||
:param parse_mode: The parsing mode for adding visual formatting to the caption.
|
||||
:param hide: Whether, or not, you would like to apply a spoiler on top of the file.
|
||||
:param disable_notification: Use this when you want to send an alert without causing the recipient's device to
|
||||
show a notification alert.
|
||||
:param protect_content: Use this to prevent recipient participants from copying/forwarding messages.
|
||||
:return: A standard response structure with an instance of 'TelegramUpdate' in the 'data' field.
|
||||
"""
|
||||
|
||||
# Start with placeholder params:
|
||||
request_params = {
|
||||
"chat_id": chat_id,
|
||||
"disable_notification": disable_notification,
|
||||
"protect_content": protect_content
|
||||
}
|
||||
if caption: request_params["caption"] = caption
|
||||
if parse_mode: request_params["parse_mode"] = parse_mode
|
||||
if hide: request_params["has_spoiler"] = True
|
||||
request_files = None
|
||||
|
||||
# If the file is given to you as a buffer:
|
||||
if isinstance(file, io.BytesIO):
|
||||
file.seek(0)
|
||||
request_files = {"photo": (file_name, file)}
|
||||
|
||||
# If the file is given to you as string:
|
||||
elif isinstance(file, str):
|
||||
if os.path.exists(file): request_files = {"photo": (file_name, open(file, "rb"))}
|
||||
else: request_params["photo"] = file
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__post(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/sendPhoto",
|
||||
params = request_params,
|
||||
files = request_files
|
||||
)
|
||||
|
||||
# If the call failed:
|
||||
if api_response.httpCode == 400:
|
||||
api_json = await api_response.get_json()
|
||||
api_response.message = api_json["description"]
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode == 200:
|
||||
api_json = await api_response.get_json()
|
||||
if api_json["ok"]:
|
||||
api_response.success = True
|
||||
api_response.data = TelegramUpdate(message = api_json["result"])
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
import asyncio
|
||||
from PIL import Image
|
||||
|
||||
async def main():
|
||||
|
||||
# Create the instance and define needed variables:
|
||||
my_tg = AsyncTelegramBot(bot_token = r"7003670393:AAH9qF6XGqa-u2_TM2JCUkP0Fp48kMw8Ka8")
|
||||
recipient_chat_id = 1275560043
|
||||
|
||||
# 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 = 872199521, limit = 10)
|
||||
# tg_response = await my_tg.get_file(file_id = r"BQACAgUAAxkBAAIHZWdT4NNG6WwSYLsHeT-0RccCGUVaAAKeFAAC5T-hVoKttePhObnbNgQ")
|
||||
# tg_response = await my_tg.send_message(
|
||||
# chat_id = recipient_chat_id,
|
||||
# message = f"Hello, World\! \N{WAVING HAND SIGN}\n[ASCII Art]({my_tg.escape_special_chars(text = r'https://patorjk.com/software/taag/#p=display&f=Tmplr&t=Type%20Something%20', parse_mode = 'MarkdownV2')})",
|
||||
# parse_mode = "MarkdownV2"
|
||||
# )
|
||||
tg_response = await my_tg.send_photo(
|
||||
chat_id = recipient_chat_id,
|
||||
file_name = "cat.jpg",
|
||||
file = r"https://images.unsplash.com/photo-1543852786-1cf6624b9987",
|
||||
caption = f"[Image URL]({my_tg.escape_special_chars(text = r'https://images.unsplash.com/photo-1543852786-1cf6624b9987', parse_mode = 'MarkdownV2')})",
|
||||
parse_mode = "MarkdownV2",
|
||||
hide = False
|
||||
)
|
||||
|
||||
# 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())
|
||||
@@ -0,0 +1,130 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Friday, 6th Dec., 2024.
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a data model for describing the API response from Telegram's APIs.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# For making data behaviour_models:
|
||||
from pydantic import BaseModel, Field, field_validator, model_validator
|
||||
from typing import Optional, Literal, Union, Dict, List, Any
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import json
|
||||
from utils_v2.string import regex
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class TelegramApiResponse(BaseModel):
|
||||
|
||||
action: str = Field(frozen = True, default = None)
|
||||
url: str = Field(frozen = True)
|
||||
method: str = Field(frozen = True)
|
||||
response: Any = None
|
||||
httpCode: int = None
|
||||
|
||||
success: bool = False
|
||||
message: str = None
|
||||
data: Any = None
|
||||
|
||||
exception: Any = None
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "forbid"
|
||||
|
||||
# ┏┓ ┏┓
|
||||
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
|
||||
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
|
||||
|
||||
def to_markdown(self):
|
||||
if self.exception: message = "❌ *TELEGRAM API EXCEPTION:* ❌\n\n"
|
||||
else: message = "*TELEGRAM API RESPONSE:*\n\n"
|
||||
message += f"*ACTION:*\n`{self.action}`\n\n"
|
||||
message += f"*URL:*\n`{self.url}`\n\n"
|
||||
message += f"*METHOD:*\n`{self.method}`\n\n"
|
||||
message += f"*RESPONSE:*\n`{self.response}`\n\n"
|
||||
message += f"*SUCCESS:*\n`{self.success}`\n\n"
|
||||
message += f"*MESSAGE:*\n`{self.message}`\n\n"
|
||||
message += f"*EXCEPTION:*\n`{self.exception.__class__.__name__}: {str(self.exception)}`\n\n"
|
||||
return message
|
||||
|
||||
async def get_json(self):
|
||||
try: return self.response.json()
|
||||
except: return {}
|
||||
|
||||
async def get_content(self):
|
||||
try: return self.response.content
|
||||
except: return b""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
pass
|
||||
@@ -0,0 +1,967 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Friday, 6th Dec., 2024.
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a data model for describing the message(s) from Telegram's APIs.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
N/A
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# For making data behaviour_models:
|
||||
from pydantic import BaseModel, Field, field_validator, model_validator, AwareDatetime, computed_field
|
||||
from typing import Optional, Literal, Union, Dict, List, Any
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import json
|
||||
from utils_v2.string import regex
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class TelegramParticipant(BaseModel):
|
||||
|
||||
id: int | str = Field(
|
||||
description = "the id of the human/bot",
|
||||
alias = "id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
isBot: bool = Field(
|
||||
description = "whether, or not, this participant is a bot",
|
||||
alias = "is_bot",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
firstName: str | None = Field(
|
||||
description = "the first name of this participant",
|
||||
alias = "first_name",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
lastName: str | None = Field(
|
||||
description = "the last name of this participant",
|
||||
alias = "last_name",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
username: str | None = Field(
|
||||
description = "the username of this participant",
|
||||
alias = "username",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
languageCode: str | None = Field(
|
||||
description = "to indicate what language the participant is using",
|
||||
alias = "language_code",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramChat(BaseModel):
|
||||
|
||||
id: int | str = Field(
|
||||
description = "the id of the chat",
|
||||
alias = "id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
firstName: str | None = Field(
|
||||
description = "the first name of this chat",
|
||||
alias = "first_name",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
lastName: str | None = Field(
|
||||
description = "the last name of this chat",
|
||||
alias = "last_name",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
username: str | None = Field(
|
||||
description = "the username of this chat",
|
||||
alias = "username",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
type: str = Field(
|
||||
description = "to indicate what kind of chat this is",
|
||||
alias = "type",
|
||||
frozen = True,
|
||||
examples = ["private", "group", "supergroup", "channel", "admin"]
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramPhoto(BaseModel):
|
||||
|
||||
fileId: str = Field(
|
||||
description = "the id of this photo; useful when fetching the photo",
|
||||
alias = "file_id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
fileUniqueId: str = Field(
|
||||
description = "the id of this photo; useful when fetching the photo",
|
||||
alias = "file_unique_id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
url: str | None = Field(
|
||||
description = "the url from where the photo can be accessed",
|
||||
alias = "url",
|
||||
frozen = False,
|
||||
default = None
|
||||
)
|
||||
|
||||
fileSize: int = Field(
|
||||
description = "the size of this photo in bytes",
|
||||
alias = "file_size",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
width: int | None = Field(
|
||||
description = "the width of the photo",
|
||||
alias = "width",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
height: int | None = Field(
|
||||
description = "the width of the photo",
|
||||
alias = "height",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
resolution: str | None = Field(
|
||||
description = "the indicator of the frame size",
|
||||
alias = "resolution",
|
||||
frozen = False,
|
||||
default = None,
|
||||
examples = ["small", "medium", "large", "original"]
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramVideo(BaseModel):
|
||||
|
||||
fileId: str = Field(
|
||||
description = "the id of this file; useful when fetching the file",
|
||||
alias = "file_id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
fileUniqueId: str = Field(
|
||||
description = "the id of this file; useful when fetching the file",
|
||||
alias = "file_unique_id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
url: str | None = Field(
|
||||
description = "the url from where the file can be accessed",
|
||||
alias = "url",
|
||||
frozen = False,
|
||||
default = None
|
||||
)
|
||||
|
||||
fileSize: int = Field(
|
||||
description = "the size of this file in bytes",
|
||||
alias = "file_size",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
width: int = Field(
|
||||
description = "the width of the frame of the video",
|
||||
alias = "width",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
height: int = Field(
|
||||
description = "the width of the frame of the video",
|
||||
alias = "height",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
duration: int = Field(
|
||||
description = "the runtime of the video in seconds",
|
||||
alias = "duration",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
mimeType: str = Field(
|
||||
description = "to indicate the format of the file",
|
||||
alias = "mime_type",
|
||||
frozen = True,
|
||||
default = None,
|
||||
examples = ["video/mp4"]
|
||||
)
|
||||
|
||||
thumbnail: TelegramPhoto = Field(
|
||||
description = "the thumbnail file of the video",
|
||||
alias = "thumbnail",
|
||||
frozen = True,
|
||||
default = None,
|
||||
examples = ["video/mp4"]
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramAudio(BaseModel):
|
||||
|
||||
fileId: str = Field(
|
||||
description = "the id of this file; useful when fetching the file",
|
||||
alias = "file_id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
fileUniqueId: str = Field(
|
||||
description = "the id of this file; useful when fetching the file",
|
||||
alias = "file_unique_id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
fileName: str | None = Field(
|
||||
description = "the filename of this file",
|
||||
alias = "file_name",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
url: str | None = Field(
|
||||
description = "the url from where the file can be accessed",
|
||||
alias = "url",
|
||||
frozen = False,
|
||||
default = None
|
||||
)
|
||||
|
||||
fileSize: int = Field(
|
||||
description = "the size of this file in bytes",
|
||||
alias = "file_size",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
duration: int = Field(
|
||||
description = "the runtime of the audio in seconds",
|
||||
alias = "duration",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
mimeType: str | None = Field(
|
||||
description = "to indicate the format of the file",
|
||||
alias = "mime_type",
|
||||
frozen = True,
|
||||
default = None,
|
||||
examples = ["audio/mpeg", "audio/ogg"]
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramDocument(BaseModel):
|
||||
|
||||
fileId: str = Field(
|
||||
description = "the id of this document; useful when fetching the document",
|
||||
alias = "file_id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
fileUniqueId: str = Field(
|
||||
description = "the id of this document; useful when fetching the document",
|
||||
alias = "file_unique_id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
fileName: str = Field(
|
||||
description = "the filename of this document",
|
||||
alias = "file_name",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
url: str | None = Field(
|
||||
description = "the url from where the document can be accessed",
|
||||
alias = "url",
|
||||
frozen = False,
|
||||
default = None
|
||||
)
|
||||
|
||||
fileSize: int = Field(
|
||||
description = "the size of this document in bytes",
|
||||
alias = "file_size",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
mimeType: str | None = Field(
|
||||
description = "to indicate the format of the file",
|
||||
alias = "mime_type",
|
||||
frozen = True,
|
||||
default = None,
|
||||
examples = ["application/pdf", "application/json"]
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramLocation(BaseModel):
|
||||
|
||||
latitude: float | int = Field(
|
||||
description = "the latitudinal coordinate of the location",
|
||||
alias = "latitude",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
longitude: float | int = Field(
|
||||
description = "the longitudinal coordinate of the location",
|
||||
alias = "longitude",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramContact(BaseModel):
|
||||
|
||||
phoneNo: str = Field(
|
||||
description = "the contact no. from the contact card",
|
||||
alias = "phone_number",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
firstName: str = Field(
|
||||
description = "the first name from the contact card",
|
||||
alias = "first_name",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
lastName: str | None = Field(
|
||||
description = "the last name from the contact card",
|
||||
alias = "last_name",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
vCard: str | None = Field(
|
||||
description = "the full vcard text",
|
||||
alias = "vcard",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
userId: int | str | None = Field(
|
||||
description = "the participant's id if this person is already registered on telegram",
|
||||
alias = "user_id",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramEntity(BaseModel):
|
||||
|
||||
type: str = Field(
|
||||
description = "the type of entity",
|
||||
alias = "type",
|
||||
frozen = True,
|
||||
examples = ["bot_command"]
|
||||
)
|
||||
|
||||
offset: int = Field(
|
||||
description = "where in the message text this entity starts",
|
||||
alias = "offset",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
length: int = Field(
|
||||
description = "the no. of chars in this entity",
|
||||
alias = "length",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramMessage(BaseModel):
|
||||
|
||||
messageId: int = Field(
|
||||
description = "the index of the message in this chat",
|
||||
alias = "message_id",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
sender: TelegramParticipant = Field(
|
||||
description = "the details of the sender",
|
||||
alias = "from",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
chat: TelegramChat = Field(
|
||||
description = "the details of the chat where this message arrived",
|
||||
alias = "chat",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
ts: AwareDatetime = Field(
|
||||
description = "the time at which this message arrived",
|
||||
alias = "date",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
text: str | None = Field(
|
||||
description = "the actual message text that was sent",
|
||||
alias = "text",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
entities: List[TelegramEntity] | None = Field(
|
||||
description = "the 'entities' sent in this update/message; you'll find the commands here",
|
||||
alias = "entities",
|
||||
frozen = True,
|
||||
default = None,
|
||||
exclude = True
|
||||
)
|
||||
|
||||
mediaGroupId: int | str | None = Field(
|
||||
description = "the group id of the media files if they are sent together in a batch",
|
||||
alias = "media_group_id",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
photo: List[TelegramPhoto] | None = Field(
|
||||
description = "the photo sent in this update/message",
|
||||
alias = "photo",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
video: TelegramVideo | None = Field(
|
||||
description = "the video sent in this update/message",
|
||||
alias = "video",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
audio: TelegramAudio | None = Field(
|
||||
description = "the audio file sent in this update/message",
|
||||
alias = "audio",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
voice: TelegramAudio | None = Field(
|
||||
description = "the voice-note sent in this update/message",
|
||||
alias = "voice",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
document: TelegramDocument | None = Field(
|
||||
description = "the document sent in this update/message",
|
||||
alias = "document",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
caption: str | None = Field(
|
||||
description = "the caption sent under the file",
|
||||
alias = "caption",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
location: TelegramLocation | None = Field(
|
||||
description = "the location sent in this update/message",
|
||||
alias = "location",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
contact: TelegramContact | None = Field(
|
||||
description = "the vcard sent in this update/message",
|
||||
alias = "contact",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
# ┏┓ ┏┓ ┓
|
||||
# ┣┫┓┏╋┏┓━━┃ ┏┓┏┳┓┏┓┓┏╋┏┓┏┫
|
||||
# ┛┗┗┻┗┗┛ ┗┛┗┛┛┗┗┣┛┗┻┗┗ ┗┻
|
||||
# ┛
|
||||
|
||||
@computed_field
|
||||
def commands(self) -> List[str]:
|
||||
if not self.entities: return []
|
||||
else: return [self.text[e.offset:e.offset+e.length] for e in self.entities if e.type == "bot_command"]
|
||||
|
||||
# ┓┏ ┓• ┓ •
|
||||
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
|
||||
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
|
||||
|
||||
@field_validator("ts", mode = "before")
|
||||
def parse_date_time(cls, value):
|
||||
return date_time.parse_date_time(input_value = value, timezone = date_time.TIMEZONE_UTC)
|
||||
|
||||
@field_validator("photo")
|
||||
def sort_photo_sizes(cls, value):
|
||||
value = sorted(value, key = lambda x: x.width, reverse = False)
|
||||
# res_list = ["original", "large", "medium", "small"]
|
||||
res_list = ["small", "medium", "large", "max"]
|
||||
for index, res in enumerate(res_list[:len(value)]): value[index].resolution = res
|
||||
return value
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TelegramUpdate(BaseModel):
|
||||
|
||||
updateId: int | None = Field(
|
||||
description = "an update from telegram; typically a message",
|
||||
alias = "update_id",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
message: TelegramMessage | None = Field(
|
||||
description = "the actual message from telegram",
|
||||
alias = "message",
|
||||
frozen = True,
|
||||
default = None
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
raw_message_with_commands = {
|
||||
"update_id": 872199505,
|
||||
"message": {
|
||||
"message_id": 3,
|
||||
"from": {
|
||||
"id": 1275560043,
|
||||
"is_bot": False,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"language_code": "en"
|
||||
},
|
||||
"chat": {
|
||||
"id": 1275560043,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"type": "private"
|
||||
},
|
||||
"date": 1733462436,
|
||||
"text": "/hello bot /stock_reel okay? /command",
|
||||
"entities": [
|
||||
{
|
||||
"offset": 0,
|
||||
"length": 6,
|
||||
"type": "bot_command"
|
||||
},
|
||||
{
|
||||
"offset": 11,
|
||||
"length": 11,
|
||||
"type": "bot_command"
|
||||
},
|
||||
{
|
||||
"offset": 29,
|
||||
"length": 8,
|
||||
"type": "bot_command"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
raw_message_with_photo = {
|
||||
"update_id": 872199516,
|
||||
"message": {
|
||||
"message_id": 14,
|
||||
"from": {
|
||||
"id": 1275560043,
|
||||
"is_bot": False,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"language_code": "en"
|
||||
},
|
||||
"chat": {
|
||||
"id": 1275560043,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"type": "private"
|
||||
},
|
||||
"date": 1733480584,
|
||||
"photo": [
|
||||
{
|
||||
"file_id": "AgACAgUAAxkBAAMOZ1LQiA912GNT-3mF8v0PXr74froAAuG_MRvYz5lWe8YAAU_PBPhzAQADAgADcwADNgQ",
|
||||
"file_unique_id": "AQAD4b8xG9jPmVZ4",
|
||||
"file_size": 1700,
|
||||
"width": 67,
|
||||
"height": 90
|
||||
},
|
||||
{
|
||||
"file_id": "AgACAgUAAxkBAAMOZ1LQiA912GNT-3mF8v0PXr74froAAuG_MRvYz5lWe8YAAU_PBPhzAQADAgADbQADNgQ",
|
||||
"file_unique_id": "AQAD4b8xG9jPmVZy",
|
||||
"file_size": 31914,
|
||||
"width": 240,
|
||||
"height": 320
|
||||
},
|
||||
{
|
||||
"file_id": "AgACAgUAAxkBAAMOZ1LQiA912GNT-3mF8v0PXr74froAAuG_MRvYz5lWe8YAAU_PBPhzAQADAgADeAADNgQ",
|
||||
"file_unique_id": "AQAD4b8xG9jPmVZ9",
|
||||
"file_size": 166406,
|
||||
"width": 600,
|
||||
"height": 800
|
||||
},
|
||||
{
|
||||
"file_id": "AgACAgUAAxkBAAMOZ1LQiA912GNT-3mF8v0PXr74froAAuG_MRvYz5lWe8YAAU_PBPhzAQADAgADeQADNgQ",
|
||||
"file_unique_id": "AQAD4b8xG9jPmVZ-",
|
||||
"file_size": 306284,
|
||||
"width": 960,
|
||||
"height": 1280
|
||||
}
|
||||
],
|
||||
"caption": "Image with caption..."
|
||||
}
|
||||
}
|
||||
raw_message_with_vcard = {
|
||||
"update_id": 872199508,
|
||||
"message": {
|
||||
"message_id": 6,
|
||||
"from": {
|
||||
"id": 1275560043,
|
||||
"is_bot": False,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"language_code": "en"
|
||||
},
|
||||
"chat": {
|
||||
"id": 1275560043,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"type": "private"
|
||||
},
|
||||
"date": 1733462585,
|
||||
"contact": {
|
||||
"phone_number": "+919821005929",
|
||||
"first_name": "Bhushan",
|
||||
"last_name": "Thakkar",
|
||||
"vcard": "BEGIN:VCARD\nVERSION:2.1\nN:Thakkar;Bhushan;;Mr. ;\nFN:Mr. Bhushan Thakkar\nTEL;CELL;PREF:+919821005929\nEND:VCARD",
|
||||
"user_id": 1587276493
|
||||
}
|
||||
}
|
||||
}
|
||||
raw_message_with_video = {
|
||||
"update_id": 872199507,
|
||||
"message": {
|
||||
"message_id": 5,
|
||||
"from": {
|
||||
"id": 1275560043,
|
||||
"is_bot": False,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"language_code": "en"
|
||||
},
|
||||
"chat": {
|
||||
"id": 1275560043,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"type": "private"
|
||||
},
|
||||
"date": 1733462529,
|
||||
"video": {
|
||||
"duration": 36,
|
||||
"width": 480,
|
||||
"height": 848,
|
||||
"mime_type": "video/mp4",
|
||||
"thumbnail": {
|
||||
"file_id": "AAMCBQADGQEAAwVnUooB5F2ecOozdBki4-lbRfDpuAACXhUAAtjPkVYW4ElGmB3cxAEAB20AAzYE",
|
||||
"file_unique_id": "AQADXhUAAtjPkVZy",
|
||||
"file_size": 644,
|
||||
"width": 181,
|
||||
"height": 320
|
||||
},
|
||||
"thumb": {
|
||||
"file_id": "AAMCBQADGQEAAwVnUooB5F2ecOozdBki4-lbRfDpuAACXhUAAtjPkVYW4ElGmB3cxAEAB20AAzYE",
|
||||
"file_unique_id": "AQADXhUAAtjPkVZy",
|
||||
"file_size": 644,
|
||||
"width": 181,
|
||||
"height": 320
|
||||
},
|
||||
"file_id": "BAACAgUAAxkBAAMFZ1KKAeRdnnDqM3QZIuPpW0Xw6bgAAl4VAALYz5FWFuBJRpgd3MQ2BA",
|
||||
"file_unique_id": "AgADXhUAAtjPkVY",
|
||||
"file_size": 6672554
|
||||
}
|
||||
}
|
||||
}
|
||||
raw_message_with_document = {
|
||||
"update_id": 872199520,
|
||||
"message": {
|
||||
"message_id": 18,
|
||||
"from": {
|
||||
"id": 1275560043,
|
||||
"is_bot": False,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"language_code": "en"
|
||||
},
|
||||
"chat": {
|
||||
"id": 1275560043,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"type": "private"
|
||||
},
|
||||
"date": 1733485003,
|
||||
"document": {
|
||||
"file_name": "1 (1).pdf",
|
||||
"mime_type": "application/pdf",
|
||||
"file_id": "BQACAgUAAxkBAAMSZ1LhyhLOznQAActduf0PpbO9kZerAAK8EgAC2M-ZVjWCHqYAAWYPwDYE",
|
||||
"file_unique_id": "AgADvBIAAtjPmVY",
|
||||
"file_size": 459547
|
||||
}
|
||||
}
|
||||
}
|
||||
raw_message_with_audio = {
|
||||
"update_id": 946018733,
|
||||
"message": {
|
||||
"message_id": 1832,
|
||||
"from": {
|
||||
"id": 1275560043,
|
||||
"is_bot": False,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"language_code": "en"
|
||||
},
|
||||
"chat": {
|
||||
"id": 1275560043,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"type": "private"
|
||||
},
|
||||
"date": 1733487195,
|
||||
"audio": {
|
||||
"duration": 250,
|
||||
"file_name": "Cheez_Badi_Song_Neha_Kakkar,_Udit_Narayan_Tanishk_B,_Viju_Sh_Anand.mp3",
|
||||
"mime_type": "audio/mpeg",
|
||||
"file_id": "CQACAgUAAxkBAAIHKGdS6lrkRyk155HTkHLIvwd1x4nSAAJFFgACnmCYVjm1SEwQ19LbNgQ",
|
||||
"file_unique_id": "AgADRRYAAp5gmFY",
|
||||
"file_size": 10007124
|
||||
}
|
||||
}
|
||||
}
|
||||
raw_message_with_voice_note = {
|
||||
"update_id": 872199521,
|
||||
"message": {
|
||||
"message_id": 19,
|
||||
"from": {
|
||||
"id": 1275560043,
|
||||
"is_bot": False,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"language_code": "en"
|
||||
},
|
||||
"chat": {
|
||||
"id": 1275560043,
|
||||
"first_name": "K",
|
||||
"last_name": "S",
|
||||
"username": "kpspsps",
|
||||
"type": "private"
|
||||
},
|
||||
"date": 1733486161,
|
||||
"voice": {
|
||||
"duration": 8,
|
||||
"mime_type": "audio/ogg",
|
||||
"file_id": "AwACAgUAAxkBAAMTZ1LmUVh9RhuT_yaJlMSbvhy-hdAAAssSAALYz5lWaUcdWedPUBc2BA",
|
||||
"file_unique_id": "AgADyxIAAtjPmVY",
|
||||
"file_size": 30004
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
parsed_update = TelegramUpdate(**raw_message_with_voice_note)
|
||||
print(parsed_update.model_dump_json(indent = 4))
|
||||
Reference in New Issue
Block a user