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

Squashed 'utils_v2/' content from commit af73d53e

Browse Source 
git-subtree-dir: utils_v2
git-subtree-split: af73d53e43729f79a736602775d61ef5b1b0d9cf
This commit is contained in:
khushal
2025-01-03 06:15:48 +00:00
commit 3c4cac1019
176 changed files with 138689 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram/models/__init__.py
View File
telegram/models/api_call.py
+130
View File
@@ -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
telegram/models/update.py
+967
View File
@@ -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))