welcomecenterbot/bot/api.py

import aiohttp
import json
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}/"


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
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:
        params["reply_to_message_id"] = reply_to
    if reply_markup:
        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
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:
        params["reply_to_message_id"] = reply_to
    if reply_markup:
        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
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:
        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
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
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
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
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,
            }
        )

    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
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"):
    url = apiBase + "sendDocument"
    params = {"chat_id": chat_id}
    filedata = aiohttp.FormData()
    filedata.add_field("document", data, filename=filename)

    async with aiohttp.ClientSession() as session:
        async with session.post(url, params=params, data=filedata) as response:
            if response.status != 200:
                error_text = await response.text()
                print(f"Error sending document: {response.status} - {error_text}")
                return None

            try:
                return await response.json()
            except ValueError as e:
                print(f"Error decoding JSON: {e}")
                return None


# https://core.telegram.org/bots/api#getchatadministrators
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
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
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
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
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
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