khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241206) Telegram Client half complete.

Browse Source
This commit is contained in:
khushal
2024-12-06 18:54:16 +05:30
parent 49f7ea13c0
commit 483e4ce021
2 changed files with 263 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/telegram/models/behaviour/telegram_bot.py
+36 -3
View File
@@ -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")
utils_v2/telegram/models/data/message.py → utils_v2/telegram/models/data/update.py
+227 -20
View File
@@ -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))