Squashed 'utils_v2/' content from commit 82830bcf
git-subtree-dir: utils_v2 git-subtree-split: 82830bcfa67cba321624ed37529d7a80a70196cc
This commit is contained in:
@@ -0,0 +1,659 @@
|
||||
"""
|
||||
|
||||
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:
|
||||
|
||||
1. Official API Documentation: https://core.telegram.org/bots/api
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# System-level activities:
|
||||
import io
|
||||
import os
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import json
|
||||
from utils_v2.system import files
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
# Data models:
|
||||
from utils_v2.telegram.models.api_call import TelegramApiResponse
|
||||
from utils_v2.telegram.models.update import TelegramUpdate
|
||||
|
||||
# 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,
|
||||
params: dict = None,
|
||||
content: str | bytes = None,
|
||||
files: dict = 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 params: The params to send in the query string itself.
|
||||
:param content: The raw content to be sent in the body (typically as an octet-stream).
|
||||
:param files: Any file that you may want to send.
|
||||
: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,
|
||||
params = params,
|
||||
content = content,
|
||||
files = files
|
||||
)
|
||||
|
||||
# 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:
|
||||
|
||||
"""
|
||||
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",
|
||||
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:
|
||||
|
||||
"""
|
||||
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",
|
||||
)
|
||||
|
||||
# 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:
|
||||
|
||||
"""
|
||||
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",
|
||||
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.data = api_json
|
||||
if api_json["ok"]:
|
||||
api_response.success = True
|
||||
api_response.data = [TelegramUpdate(**r) for r in api_json["result"]]
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
async def get_file(
|
||||
self,
|
||||
file_id: str,
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
"""
|
||||
To download a file 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 file_id: The 'file_id' of this file. Not to be confused with 'file_unique_id'.
|
||||
:return: A standard response structure with the file's content in the 'data' field.
|
||||
"""
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__get(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/getFile",
|
||||
params = {"file_id": file_id}
|
||||
)
|
||||
|
||||
# If the call failed:
|
||||
if api_response.httpCode == 400:
|
||||
api_json = await api_response.get_json()
|
||||
api_response.message = api_json["description"]
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode == 200:
|
||||
api_json = await api_response.get_json()
|
||||
file_path = api_json["result"]["file_path"]
|
||||
api_response = await self.__get(url = f"https://api.telegram.org/file/bot{self.__bot_token}/{file_path}")
|
||||
if api_response.httpCode == 200:
|
||||
api_response.success = True
|
||||
api_response.data = await api_response.get_content()
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
# ┏┓ ┓•
|
||||
# ┗┓┏┓┏┓┏┫┓┏┓┏┓
|
||||
# ┗┛┗ ┛┗┗┻┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
@staticmethod
|
||||
def escape_special_chars(
|
||||
text: str,
|
||||
parse_mode: Literal[None, "Markdown", "MarkdownV2", "HTML"] = None
|
||||
) -> str:
|
||||
|
||||
r"""
|
||||
Escapes certain special chars for each type of parsing that is available. There's not much use of this function,
|
||||
though, TBH. This can be useful when trying to parse more complex thing like links.
|
||||
USE CASE: Suppose you want to send a message "Equation: 5 * 2 = 10" such that the actual equation part
|
||||
("5 * 2 = 10") is in bold, and you want this through, say, MarkdownV2 formatting. You could achieve this by
|
||||
first passing the part that you want to format through this function and then inserting it in the rest. Maybe
|
||||
like this:
|
||||
>> escaped_message = f"Equation: *{escape_special_chars(text = '5 * 2 = 10', parse_mode = 'MarkdownV2')}*"
|
||||
Or you could just manually achieve the same by typing:
|
||||
>> escaped_message = r"Equation: *5 \* 2 \= 10*"
|
||||
:param text: The raw string.
|
||||
:param parse_mode: The kind of parsing that is desired.
|
||||
:return: The escaped string.
|
||||
"""
|
||||
|
||||
# For MarkdownV2:
|
||||
if parse_mode == "Markdown" or parse_mode == "MarkdownV2":
|
||||
for k, v in {
|
||||
"*": "\\*",
|
||||
"`": "\\`",
|
||||
"[": "\\[",
|
||||
"]": "\\]",
|
||||
"(": "\\(",
|
||||
")": "\\)",
|
||||
"~": "\\~",
|
||||
">": "\\>",
|
||||
"#": "\\#",
|
||||
"+": "\\+",
|
||||
"-": "\\-",
|
||||
".": "\\.",
|
||||
"!": "\\!",
|
||||
"=": "\\=",
|
||||
}.items(): text = text.replace(k, v)
|
||||
|
||||
# For HTML:
|
||||
elif parse_mode == "HTML":
|
||||
for k, v in {
|
||||
"&": "&",
|
||||
"<": "<",
|
||||
">": ">",
|
||||
"\"": """,
|
||||
"'": "'"
|
||||
}.items(): text = text.replace(k, v)
|
||||
|
||||
# Done here:
|
||||
return text
|
||||
|
||||
async def send_message(
|
||||
self,
|
||||
chat_id: int | str,
|
||||
message: str,
|
||||
parse_mode: Literal[None, "Markdown", "MarkdownV2", "HTML"] = None,
|
||||
disable_notification: bool = False,
|
||||
protect_content: bool = False
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
"""
|
||||
To send one message to one recipient using a specified parse mode.
|
||||
:param chat_id: The id of the recipient user/group/channel.
|
||||
:param message: The actual message that you want to send.
|
||||
:param parse_mode: The parsing mode for adding visual formatting.
|
||||
:param disable_notification: Use this when you want to send an alert without causing the recipient's device to
|
||||
show a notification alert.
|
||||
:param protect_content: Use this to prevent recipient participants from copying/forwarding messages.
|
||||
:return: A standard response structure with an instance of 'TelegramUpdate' in the 'data' field.
|
||||
"""
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__get(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/sendMessage",
|
||||
params = {
|
||||
"chat_id": chat_id,
|
||||
"text": message,
|
||||
"parse_mode": parse_mode,
|
||||
"disable_notification": disable_notification,
|
||||
"protect_content": protect_content
|
||||
}
|
||||
)
|
||||
|
||||
# If the call failed:
|
||||
if api_response.httpCode == 400:
|
||||
api_json = await api_response.get_json()
|
||||
api_response.message = api_json["description"]
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode == 200:
|
||||
api_json = await api_response.get_json()
|
||||
if api_json["ok"]:
|
||||
api_response.success = True
|
||||
api_response.data = TelegramUpdate(message = api_json["result"])
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
async def send_photo(
|
||||
self,
|
||||
chat_id: int | str,
|
||||
file: str | io.BytesIO,
|
||||
file_name: str = None,
|
||||
caption: str = None,
|
||||
parse_mode: Literal[None, "Markdown", "MarkdownV2", "HTML"] = None,
|
||||
hide: bool = False,
|
||||
disable_notification: bool = False,
|
||||
protect_content: bool = False
|
||||
) -> TelegramApiResponse:
|
||||
|
||||
"""
|
||||
To send one photo from the bot to one recipient. A caption with a specified parsing mode may also be added.
|
||||
Note that you must send at least one of the three file options.
|
||||
:param chat_id: The id of the recipient user/group/channel.
|
||||
:param file: Could be either a file on the disk, or a file held in a buffer in RAM or a 'file_id' of a file
|
||||
already on Telegram's server, or a URL of a file hosted on the internet.
|
||||
:param file_name: The name of the file.
|
||||
:param caption: The text to write under the file.
|
||||
:param parse_mode: The parsing mode for adding visual formatting to the caption.
|
||||
:param hide: Whether, or not, you would like to apply a spoiler on top of the file.
|
||||
:param disable_notification: Use this when you want to send an alert without causing the recipient's device to
|
||||
show a notification alert.
|
||||
:param protect_content: Use this to prevent recipient participants from copying/forwarding messages.
|
||||
:return: A standard response structure with an instance of 'TelegramUpdate' in the 'data' field.
|
||||
"""
|
||||
|
||||
# Start with placeholder params:
|
||||
request_params = {
|
||||
"chat_id": chat_id,
|
||||
"disable_notification": disable_notification,
|
||||
"protect_content": protect_content
|
||||
}
|
||||
if caption: request_params["caption"] = caption
|
||||
if parse_mode: request_params["parse_mode"] = parse_mode
|
||||
if hide: request_params["has_spoiler"] = True
|
||||
request_files = None
|
||||
|
||||
# If the file is given to you as a buffer:
|
||||
if isinstance(file, io.BytesIO):
|
||||
file.seek(0)
|
||||
request_files = {"photo": (file_name, file)}
|
||||
|
||||
# If the file is given to you as string:
|
||||
elif isinstance(file, str):
|
||||
if os.path.exists(file): request_files = {"photo": (file_name, open(file, "rb"))}
|
||||
else: request_params["photo"] = file
|
||||
|
||||
# Make the API call:
|
||||
api_response = await self.__post(
|
||||
url = f"https://api.telegram.org/bot{self.__bot_token}/sendPhoto",
|
||||
params = request_params,
|
||||
files = request_files
|
||||
)
|
||||
|
||||
# If the call failed:
|
||||
if api_response.httpCode == 400:
|
||||
api_json = await api_response.get_json()
|
||||
api_response.message = api_json["description"]
|
||||
|
||||
# If the call succeeds:
|
||||
if api_response.httpCode == 200:
|
||||
api_json = await api_response.get_json()
|
||||
if api_json["ok"]:
|
||||
api_response.success = True
|
||||
api_response.data = TelegramUpdate(message = api_json["result"])
|
||||
|
||||
# Done here:
|
||||
return api_response
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
import asyncio
|
||||
from PIL import Image
|
||||
|
||||
async def main():
|
||||
|
||||
# Create the instance and define needed variables:
|
||||
my_tg = AsyncTelegramBot(bot_token = r"7003670393:AAH9qF6XGqa-u2_TM2JCUkP0Fp48kMw8Ka8")
|
||||
# recipient_chat_id = 1275560043
|
||||
recipient_chat_id = 7501974519
|
||||
|
||||
# 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 = 872199521, limit = 10)
|
||||
# tg_response = await my_tg.get_file(file_id = r"BQACAgUAAxkBAAIHZWdT4NNG6WwSYLsHeT-0RccCGUVaAAKeFAAC5T-hVoKttePhObnbNgQ")
|
||||
# tg_response = await my_tg.send_message(
|
||||
# chat_id = recipient_chat_id,
|
||||
# message = f"Hello, World\! \N{WAVING HAND SIGN}\n[ASCII Art]({my_tg.escape_special_chars(text = r'https://patorjk.com/software/taag/#p=display&f=Tmplr&t=Type%20Something%20', parse_mode = 'MarkdownV2')})",
|
||||
# parse_mode = "MarkdownV2"
|
||||
# )
|
||||
tg_response = await my_tg.send_photo(
|
||||
chat_id = recipient_chat_id,
|
||||
file_name = "cat.jpg",
|
||||
file = r"https://images.unsplash.com/photo-1543852786-1cf6624b9987",
|
||||
caption = f"[Image URL]({my_tg.escape_special_chars(text = r'https://images.unsplash.com/photo-1543852786-1cf6624b9987', parse_mode = 'MarkdownV2')})",
|
||||
parse_mode = "MarkdownV2",
|
||||
hide = True
|
||||
)
|
||||
|
||||
# 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())
|
||||
Reference in New Issue
Block a user