From 483e4ce0213a3bb27f803aec954d7007e283ed44 Mon Sep 17 00:00:00 2001 From: khushal Date: Fri, 6 Dec 2024 18:54:16 +0530 Subject: [PATCH] (20241206) Telegram Client half complete. --- .../telegram/models/behaviour/telegram_bot.py | 39 ++- .../models/data/{message.py => update.py} | 247 ++++++++++++++++-- 2 files changed, 263 insertions(+), 23 deletions(-) rename utils_v2/telegram/models/data/{message.py => update.py} (77%) diff --git a/utils_v2/telegram/models/behaviour/telegram_bot.py b/utils_v2/telegram/models/behaviour/telegram_bot.py index cb8adf4..33ea305 100644 --- a/utils_v2/telegram/models/behaviour/telegram_bot.py +++ b/utils_v2/telegram/models/behaviour/telegram_bot.py @@ -44,6 +44,7 @@ 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 @@ -294,6 +295,14 @@ class AsyncTelegramBot: 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", @@ -315,6 +324,13 @@ class AsyncTelegramBot: 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", @@ -331,6 +347,11 @@ class AsyncTelegramBot: # Done here: return api_response + # ┳┳┓ ┓┓ ┏┓ ┓┓ ┳┓ + # ┃┃┃┏┓┏┓┓┏┏┓┃┃┓┏ ┃┃┓┏┃┃ ┃┃┏┓╋┏┓ + # ┛ ┗┗┻┛┗┗┻┗┻┗┗┗┫ ┣┛┗┻┗┗ ┻┛┗┻┗┗┻ + # ┛ + async def get_updates( self, update_id: int = None, @@ -338,6 +359,17 @@ class AsyncTelegramBot: 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", @@ -353,8 +385,9 @@ class AsyncTelegramBot: 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 + if api_json["ok"]: + api_response.success = True + api_response.data = [TelegramUpdate(**r) for r in api_json["result"]] # Done here: return api_response @@ -381,7 +414,7 @@ if __name__ == "__main__": # 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) + tg_response = await my_tg.get_updates(update_id = 872199521, limit = 5) # Show the response: print("SUMMARY:", tg_response.to_markdown(), "\n\n---\n\n") diff --git a/utils_v2/telegram/models/data/message.py b/utils_v2/telegram/models/data/update.py similarity index 77% rename from utils_v2/telegram/models/data/message.py rename to utils_v2/telegram/models/data/update.py index c01a729..25db35e 100644 --- a/utils_v2/telegram/models/data/message.py +++ b/utils_v2/telegram/models/data/update.py @@ -153,10 +153,11 @@ class TelegramChat(BaseModel): default = None ) - type: Literal["private", "group", "channel"] = Field( + type: str = Field( description = "to indicate what kind of chat this is", alias = "type", - frozen = True + frozen = True, + examples = ["private", "group", "supergroup", "channel", "admin"] ) # ┏┓ ┏• @@ -167,23 +168,6 @@ class TelegramChat(BaseModel): 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 - # --------------------------------------------------------------------------------------------------------------------- @@ -325,6 +309,120 @@ class TelegramVideo(BaseModel): # --------------------------------------------------------------------------------------------------------------------- +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( @@ -493,6 +591,27 @@ class TelegramMessage(BaseModel): 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", @@ -750,6 +869,94 @@ if __name__ == "__main__": } } } + 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_video) + parsed_update = TelegramUpdate(**raw_message_with_voice_note) print(parsed_update.model_dump_json(indent = 4))