97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
import requests
|
|
import json
|
|
import os
|
|
|
|
TOKEN = os.environ.get('BOT_TOKEN')
|
|
apiBase = f"https://api.telegram.org/bot{TOKEN}/"
|
|
|
|
|
|
def register_webhook(url):
|
|
r = requests.get(
|
|
apiBase + f'setWebhook?url={url}'
|
|
)
|
|
return r
|
|
|
|
|
|
def delete_message(cid: str, mid: str):
|
|
url = apiBase + f"deleteMessage?chat_id={cid}&message_id={mid}"
|
|
r = requests.post(url)
|
|
return r
|
|
|
|
|
|
def send_message(cid: str, body, reply_to=None, reply_markup=None):
|
|
url = apiBase + f"sendMessage?chat_id={cid}&text={body}"
|
|
if reply_to:
|
|
url += f'&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}'
|
|
r = requests.post(url)
|
|
print(f'{url}')
|
|
return r
|
|
|
|
|
|
# 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}"
|
|
if until_date:
|
|
url += f'&until_data={until_date}'
|
|
r = requests.post(url)
|
|
return r
|
|
|
|
|
|
# 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
|
|
|
|
# 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
|
|
|
|
|
|
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
|
|
|
|
|
|
# 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={user_id}&permissions={chat_permissions}'
|
|
r = requests.post(url)
|
|
return r
|
|
|
|
|
|
# https://core.telegram.org/bots/api#restrictchatmember
|
|
def unmute_member(chat_id, member_id):
|
|
chat_permissions = json.dumps({ "can_send_messages": True })
|
|
chat_permissions = requests.utils.quote(chat_permissions)
|
|
url = apiBase + f'restrictChatMember?chat_id={chat_id}' + \
|
|
f'&user_id={user_id}&permissions={chat_permissions}'
|
|
r = requests.post(url)
|
|
return r
|
|
|
|
# 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
|
|
|
|
|
|
def send_graph(png_data, chat_id):
|
|
url = apiBase + f"sendPhoto"
|
|
headers = {"Content-Type": "multipart/form-data"}
|
|
files = {"photo": ("chart.png", png_data)}
|
|
params = {"chat_id": chat_id}
|
|
response = requests.post(url, headers=headers, files=files, params=params)
|
|
return response.json() |