""" 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))