dokku-comp
This commit is contained in:
323
bot/api.py
323
bot/api.py
@@ -1,131 +1,198 @@
|
||||
import requests
|
||||
import aiohttp
|
||||
import json
|
||||
import os
|
||||
from bot.config import BOT_TOKEN, WEBHOOK
|
||||
import logging
|
||||
|
||||
# Create a logger instance
|
||||
logger = logging.getLogger(__name__)
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
apiBase = f"https://api.telegram.org/bot{BOT_TOKEN}/"
|
||||
|
||||
|
||||
def register_webhook():
|
||||
r = requests.get(apiBase + f'setWebhook?url={WEBHOOK}')
|
||||
return r.json()
|
||||
|
||||
|
||||
def delete_message(cid: str, mid: str):
|
||||
url = apiBase + f"deleteMessage?chat_id={cid}&message_id={mid}"
|
||||
r = requests.post(url)
|
||||
return r.json()
|
||||
async def register_webhook():
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(apiBase + f"setWebhook?url={WEBHOOK}") as response:
|
||||
data = await response.json()
|
||||
logger.info("Webhook registration response: %s", data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#sendmessage
|
||||
def send_message(cid: str, body, reply_to=None, reply_markup=None):
|
||||
url = f"sendMessage?chat_id={cid}&text={body}"
|
||||
async def send_message(cid: str, body, reply_to=None, reply_markup=None):
|
||||
if not body:
|
||||
return
|
||||
url = apiBase + f"sendMessage"
|
||||
params = {"chat_id": cid, "text": body, "parse_mode": "HTML"}
|
||||
if reply_to:
|
||||
url += f'&reply_to_message_id={reply_to}'
|
||||
params["reply_to_message_id"] = reply_to
|
||||
if reply_markup:
|
||||
reply_markup = json.dumps(reply_markup)
|
||||
reply_markup = requests.utils.quote(reply_markup)
|
||||
url += f'&reply_markup={reply_markup}'
|
||||
url += f'&parse_mode=html'
|
||||
print(url)
|
||||
r = requests.post(apiBase + url)
|
||||
return r.json()
|
||||
params["reply_markup"] = json.dumps(reply_markup)
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Message sent to %s: %s", cid, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#sendchataction
|
||||
async def send_chataction(cid: str, action: str):
|
||||
url = apiBase + f"sendChatAction"
|
||||
params = {"chat_id": cid, "action": action}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Chat action sent to %s: %s", cid, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#forwardmessage
|
||||
async def forward_message(cid, mid, to_chat_id):
|
||||
url = apiBase + f"forwardMessage"
|
||||
params = {"chat_id": to_chat_id, "from_chat_id": cid, "message_id": mid}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Message forwarded from %s to %s: %s", cid, to_chat_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#deletemessage
|
||||
async def delete_message(cid: str, mid: str):
|
||||
url = apiBase + f"deleteMessage"
|
||||
params = {"chat_id": cid, "message_id": mid}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#sendphoto
|
||||
def send_photo(cid: str, file_id: str, caption="", reply_to=None, reply_markup=None):
|
||||
url = f"sendPhoto?chat_id={cid}&photo={file_id}&caption={caption}"
|
||||
async def send_photo(
|
||||
cid: str, file_id: str, caption="", reply_to=None, reply_markup=None
|
||||
):
|
||||
url = apiBase + f"sendPhoto"
|
||||
params = {
|
||||
"chat_id": cid,
|
||||
"photo": file_id,
|
||||
"caption": caption,
|
||||
"parse_mode": "HTML",
|
||||
}
|
||||
if reply_to:
|
||||
url += f'&reply_to_message_id={reply_to}'
|
||||
params["reply_to_message_id"] = reply_to
|
||||
if reply_markup:
|
||||
reply_markup = json.dumps(reply_markup)
|
||||
reply_markup = requests.utils.quote(reply_markup)
|
||||
url += f'&reply_markup={reply_markup}'
|
||||
url += f'&parse_mode=html'
|
||||
print(url)
|
||||
r = requests.post(apiBase + url)
|
||||
return r.json()
|
||||
params["reply_markup"] = json.dumps(reply_markup)
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Photo sent to %s: %s", cid, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#banchatmember
|
||||
def ban_member(chat_id, user_id, until_date=None):
|
||||
url = apiBase + f"banChatMember?chat_id={chat_id}&user_id={user_id}"
|
||||
async def ban_member(chat_id, user_id, until_date=None):
|
||||
url = apiBase + f"banChatMember"
|
||||
params = {"chat_id": chat_id, "user_id": user_id}
|
||||
if until_date:
|
||||
url += f'&until_data={until_date}'
|
||||
r = requests.post(url)
|
||||
return r.json()
|
||||
params["until_date"] = until_date
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Member banned from %s: %s", chat_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#unbanchatmember
|
||||
def unban_member(chat_id, user_id):
|
||||
url = apiBase + f"unbanChatMember?chat_id={chat_id}&user_id={user_id}&only_if_banned=1"
|
||||
r = requests.post(url)
|
||||
return r.json()
|
||||
async def unban_member(chat_id, user_id):
|
||||
url = apiBase + f"unbanChatMember"
|
||||
params = {"chat_id": chat_id, "user_id": user_id, "only_if_banned": 1}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Member unbanned from %s: %s", chat_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#addchatmember
|
||||
def add_member(chat_id, user_id):
|
||||
url = apiBase + f"addChatMember?chat_id={chat_id}&user_id={user_id}"
|
||||
r = requests.post(url)
|
||||
return r.json()
|
||||
|
||||
|
||||
def forward_message(cid, mid, to_chat_id):
|
||||
url = apiBase + f"forwardMessage?chat_id={to_chat_id}" + \
|
||||
f"&from_chat_id={cid}&message_id={mid}"
|
||||
r = requests.post(url)
|
||||
return r.json()
|
||||
async def add_member(chat_id, user_id):
|
||||
url = apiBase + f"addChatMember"
|
||||
params = {"chat_id": chat_id, "user_id": user_id}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Member added to %s: %s", chat_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#restrictchatmember
|
||||
def mute_member(chat_id, member_id):
|
||||
chat_permissions = json.dumps({ "can_send_messages": False })
|
||||
chat_permissions = requests.utils.quote(chat_permissions)
|
||||
url = apiBase + f'restrictChatMember?chat_id={chat_id}' + \
|
||||
f'&user_id={member_id}&permissions={chat_permissions}' + \
|
||||
f'&use_independent_chat_permissions=1'
|
||||
r = requests.post(url)
|
||||
return r.json()
|
||||
async def mute_member(chat_id, member_id):
|
||||
url = apiBase + f"restrictChatMember"
|
||||
params = {
|
||||
"chat_id": chat_id,
|
||||
"user_id": member_id,
|
||||
"permissions": json.dumps({"can_send_messages": False}),
|
||||
"use_independent_chat_permissions": 1,
|
||||
}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Member muted in %s: %s", chat_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#restrictchatmember
|
||||
def unmute_member(chat_id, member_id, chat_permissions=None):
|
||||
async def unmute_member(chat_id, member_id, chat_permissions=None):
|
||||
if not chat_permissions:
|
||||
chat_permissions = json.dumps({
|
||||
"can_send_messages": True,
|
||||
"can_send_photos": True,
|
||||
"can_send_other_messages": True,
|
||||
"can_send_polls": True,
|
||||
"can_add_web_page_previews": True,
|
||||
"can_send_audios": True,
|
||||
"can_invite_users": True,
|
||||
"can_send_voice_notes": True,
|
||||
"can_send_video_notes": True,
|
||||
"can_send_videos": True,
|
||||
"can_send_documents": True
|
||||
})
|
||||
chat_permissions = requests.utils.quote(chat_permissions)
|
||||
url = apiBase + f'restrictChatMember?chat_id={chat_id}' + \
|
||||
f'&user_id={member_id}&permissions={chat_permissions}' + \
|
||||
f'&use_independent_chat_permissions=1'
|
||||
r = requests.post(url)
|
||||
return r.json()
|
||||
chat_permissions = json.dumps(
|
||||
{
|
||||
"can_send_messages": True,
|
||||
"can_send_photos": True,
|
||||
"can_send_other_messages": True,
|
||||
"can_send_polls": True,
|
||||
"can_add_web_page_previews": True,
|
||||
"can_send_audios": True,
|
||||
"can_invite_users": True,
|
||||
"can_send_voice_notes": True,
|
||||
"can_send_video_notes": True,
|
||||
"can_send_videos": True,
|
||||
"can_send_documents": True,
|
||||
}
|
||||
)
|
||||
|
||||
url = apiBase + f"restrictChatMember"
|
||||
params = {
|
||||
"chat_id": chat_id,
|
||||
"user_id": member_id,
|
||||
"permissions": chat_permissions,
|
||||
"use_independent_chat_permissions": 1,
|
||||
}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Member unmuted in %s: %s", chat_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#approvechatjoinrequest
|
||||
def approve_chat_join_request(chat_id, user_id):
|
||||
url = apiBase + f"approveChatJoinRequest?chat_id={chat_id}" + \
|
||||
f'&user_id={user_id}'
|
||||
r = requests.post(url)
|
||||
return r.json()
|
||||
async def approve_chat_join_request(chat_id, user_id):
|
||||
url = apiBase + f"approveChatJoinRequest"
|
||||
params = {"chat_id": chat_id, "user_id": user_id}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Chat join request approved in %s: %s", chat_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#senddocument
|
||||
async def send_document(chat_id, data='', filename='chart.svg'):
|
||||
async def send_document(chat_id, data="", filename="chart.svg"):
|
||||
url = apiBase + "sendDocument"
|
||||
params = { "chat_id": chat_id }
|
||||
params = {"chat_id": chat_id}
|
||||
filedata = aiohttp.FormData()
|
||||
filedata.add_field('document', data, filename=filename)
|
||||
filedata.add_field("document", data, filename=filename)
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params, data=filedata) as response:
|
||||
@@ -142,46 +209,70 @@ async def send_document(chat_id, data='', filename='chart.svg'):
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#getchatadministrators
|
||||
def get_chat_administrators(chat_id):
|
||||
url = apiBase + f"getChatAdministrators?chat_id={chat_id}"
|
||||
r = requests.get(url)
|
||||
return r.json()
|
||||
async def get_chat_administrators(chat_id):
|
||||
url = apiBase + f"getChatAdministrators"
|
||||
params = {"chat_id": chat_id}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Chat administrators retrieved for %s: %s", chat_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#getchatmember
|
||||
def get_member(chat_id, member_id):
|
||||
url = apiBase + f"getChatMember?chat_id={member_id}&user_id={member_id}"
|
||||
r = requests.get(url)
|
||||
return r.json()
|
||||
async def get_member(chat_id, member_id):
|
||||
url = apiBase + f"getChatMember"
|
||||
params = {"chat_id": chat_id, "user_id": member_id}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Chat member retrieved for %s: %s", chat_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#getuserprofilephotos
|
||||
def get_userphotos(user_id):
|
||||
url = apiBase + f"getUserProfilePhotos?user_id={user_id}"
|
||||
r = requests.get(url)
|
||||
return r.json()
|
||||
async def get_userphotos(user_id):
|
||||
url = apiBase + f"getUserProfilePhotos"
|
||||
params = {"user_id": user_id}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("User profile photos retrieved for %s: %s", user_id, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#editmessagereplymarkup
|
||||
def edit_replymarkup(cid, mid, reply_markup):
|
||||
reply_markup = json.dumps(reply_markup)
|
||||
reply_markup = requests.utils.quote(reply_markup)
|
||||
url = f"editMessageReplyMarkup?chat_id={cid}&message_id={mid}&reply_markup={reply_markup}"
|
||||
r = requests.post(apiBase + url)
|
||||
return r.json()
|
||||
async def edit_replymarkup(cid, mid, reply_markup):
|
||||
url = apiBase + f"editMessageReplyMarkup"
|
||||
params = {
|
||||
"chat_id": cid,
|
||||
"message_id": mid,
|
||||
"reply_markup": json.dumps(reply_markup),
|
||||
}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Reply markup edited for message %s in %s: %s", mid, cid, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#getchat
|
||||
def get_chat(cid):
|
||||
url = apiBase + f"getChat?chat_id={cid}"
|
||||
r = requests.get(url)
|
||||
return r.json()
|
||||
async def get_chat(cid):
|
||||
url = apiBase + f"getChat"
|
||||
params = {"chat_id": cid}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Chat retrieved for %s: %s", cid, data)
|
||||
return data
|
||||
|
||||
|
||||
# https://core.telegram.org/bots/api#banchatmember
|
||||
def kick_member(chat_id, member_id):
|
||||
url = f"banChatSenderChat?chat_id={cid}&user_id={member_id}"
|
||||
r = requests.post(apiBase + url)
|
||||
print(r.json())
|
||||
url = f"unbanChatSenderChat?chat_id={cid}&user_id={member_id}&only_if_banned=1"
|
||||
r = requests.post(apiBase + url)
|
||||
return r.json()
|
||||
async def kick_member(chat_id, member_id):
|
||||
url = apiBase + f"banChatMember"
|
||||
params = {"chat_id": chat_id, "user_id": member_id}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, params=params) as response:
|
||||
data = await response.json()
|
||||
logger.info("Member kicked from %s: %s", chat_id, data)
|
||||
return data
|
||||
|
||||
Reference in New Issue
Block a user