From 49f7ea13c0cf11b0dd0167969ef332f5a5190a8b Mon Sep 17 00:00:00 2001 From: khushal Date: Fri, 6 Dec 2024 17:09:35 +0530 Subject: [PATCH] (20241206) Sending the 'Subject' in the mails list now. --- models/behaviour/mail/retrieve.py | 1 + models/behaviour/mail/sync_v2.py | 4 + utils_v2/mail/mail_parser.py | 4 +- .../savvy_bulk_sms/async_savvy_bulk_sms.py | 2 + utils_v2/telegram/__init__.py | 0 utils_v2/telegram/models/__init__.py | 0 .../telegram/models/behaviour/__init__.py | 0 .../telegram/models/behaviour/telegram_bot.py | 391 +++++++++ utils_v2/telegram/models/data/__init__.py | 0 utils_v2/telegram/models/data/api_call.py | 130 +++ utils_v2/telegram/models/data/message.py | 755 ++++++++++++++++++ 11 files changed, 1285 insertions(+), 2 deletions(-) create mode 100644 utils_v2/telegram/__init__.py create mode 100644 utils_v2/telegram/models/__init__.py create mode 100644 utils_v2/telegram/models/behaviour/__init__.py create mode 100644 utils_v2/telegram/models/behaviour/telegram_bot.py create mode 100644 utils_v2/telegram/models/data/__init__.py create mode 100644 utils_v2/telegram/models/data/api_call.py create mode 100644 utils_v2/telegram/models/data/message.py diff --git a/models/behaviour/mail/retrieve.py b/models/behaviour/mail/retrieve.py index 55af60c..14b10d7 100644 --- a/models/behaviour/mail/retrieve.py +++ b/models/behaviour/mail/retrieve.py @@ -171,6 +171,7 @@ class MailRetrieveModel(BaseModel): "payload.ts": True, "payload.readTs": True, "payload.from": True, + "payload.subject": True, "payload.labels": True, "payload.snippet": True, "payload.aiSnippet": True, diff --git a/models/behaviour/mail/sync_v2.py b/models/behaviour/mail/sync_v2.py index b2f599b..de4fd81 100644 --- a/models/behaviour/mail/sync_v2.py +++ b/models/behaviour/mail/sync_v2.py @@ -281,6 +281,10 @@ class MailSyncModel(BaseModel): filter = mongo_conn.dict_to_dot_notation({ "payload": { "messageId": message_id + }, + "user_info": { + "entityId": user_info["entityId"], + "billingAccountId": user_info["billingAccountId"] } }), projection = { diff --git a/utils_v2/mail/mail_parser.py b/utils_v2/mail/mail_parser.py index 8bda8a3..f0d6abe 100644 --- a/utils_v2/mail/mail_parser.py +++ b/utils_v2/mail/mail_parser.py @@ -122,7 +122,7 @@ def parse(raw_mail: str | bytes) -> Dict[str, Any]: # in which to render the contents of the page. parts = [ { - "no": None, + "partNo": None, "offset": max( parsed_mail.message_as_string.find(t), parsed_mail.message_as_string.find(base64.b64encode(t.encode()).decode()) @@ -143,7 +143,7 @@ def parse(raw_mail: str | bytes) -> Dict[str, Any]: } for h in parsed_mail.text_html ] parts = sorted(parts, key = lambda x: x["offset"]) - for i, p in enumerate(parts): p["no"] = i + for i, p in enumerate(parts): p["partNo"] = i # Get the unformatted text from everything in the mail: unformatted_text = [] diff --git a/utils_v2/sms/savvy_bulk_sms/async_savvy_bulk_sms.py b/utils_v2/sms/savvy_bulk_sms/async_savvy_bulk_sms.py index 2cf25e2..e0769a1 100644 --- a/utils_v2/sms/savvy_bulk_sms/async_savvy_bulk_sms.py +++ b/utils_v2/sms/savvy_bulk_sms/async_savvy_bulk_sms.py @@ -91,6 +91,8 @@ class AsyncSavvyBulkSMS: :param api_key: The API key received from Savvy Bulk SMS. :param partner_id: Your id with Savvy Bulk SMS. :param short_code: Your short code with Savvy Bulk SMS. + :param http_client: An instance of 'httpx' library's AsyncClient. If not given, one will be instantiated + internally. It is recommended that, for multi-auth use cases, you provide a common HTTP client from outside. :param debug: Whether, or not, you would like to show debugging messages (can be changed on the fly). :param debug_prefix: The prefix text to show with the debug string. :param debug_only_errors: Whether you would like to show all debugging messages or just error messages. diff --git a/utils_v2/telegram/__init__.py b/utils_v2/telegram/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/telegram/models/__init__.py b/utils_v2/telegram/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/telegram/models/behaviour/__init__.py b/utils_v2/telegram/models/behaviour/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/telegram/models/behaviour/telegram_bot.py b/utils_v2/telegram/models/behaviour/telegram_bot.py new file mode 100644 index 0000000..cb8adf4 --- /dev/null +++ b/utils_v2/telegram/models/behaviour/telegram_bot.py @@ -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()) diff --git a/utils_v2/telegram/models/data/__init__.py b/utils_v2/telegram/models/data/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/telegram/models/data/api_call.py b/utils_v2/telegram/models/data/api_call.py new file mode 100644 index 0000000..6f042d7 --- /dev/null +++ b/utils_v2/telegram/models/data/api_call.py @@ -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 diff --git a/utils_v2/telegram/models/data/message.py b/utils_v2/telegram/models/data/message.py new file mode 100644 index 0000000..c01a729 --- /dev/null +++ b/utils_v2/telegram/models/data/message.py @@ -0,0 +1,755 @@ +""" + + 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 = Field( + description = "the first name of this participant", + alias = "first_name", + frozen = True + ) + + 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 = Field( + description = "the first name of this chat", + alias = "first_name", + frozen = True + ) + + 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: Literal["private", "group", "channel"] = Field( + description = "to indicate what kind of chat this is", + alias = "type", + frozen = True + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "ignore" + + # ┏┓ ┏┓ ┓ + # ┣┫┓┏╋┏┓━━┃ ┏┓┏┳┓┏┓┓┏╋┏┓┏┫ + # ┛┗┗┻┗┗┛ ┗┛┗┛┛┗┗┣┛┗┻┗┗ ┗┻ + # ┛ + + @computed_field + def isPrivateChat(self) -> bool: + return True if self.type == "private" else False + + @computed_field + def isGroupChat(self) -> bool: + return True if self.type == "group" else False + + @computed_field + def isChannel(self) -> bool: + return True if self.type == "channel" else False + + +# --------------------------------------------------------------------------------------------------------------------- + + +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 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 + ) + + 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 = True) + for index, res in enumerate(["original", "large", "medium", "small"]): value[index].resolution = res + return value + + +# --------------------------------------------------------------------------------------------------------------------- + + +class TelegramUpdate(BaseModel): + + updateId: int = Field( + description = "an update from telegram; typically a message", + alias = "update_id", + frozen = True + ) + + 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 + } + } + } + + parsed_update = TelegramUpdate(**raw_message_with_video) + print(parsed_update.model_dump_json(indent = 4))