Gambling-BOT/main.py

import asyncio
from typing import Any

import requests
import websockets
import json

from websockets import Response

from enums import OpCode, EventTitle

with open('configuration.json', 'r') as file:
    CONFIG = json.load(file)

APPLICATION_ID: int = CONFIG["APPLICATION_ID"]
GATEWAY_URL: str = "wss://gateway.discord.gg/?v=10&encoding=json"
API_URL: str = "https://discord.com/api/v10"
TOKEN: str = CONFIG["TOKEN"]


async def create_command():
    body: dict = {
        "name": "poll",
        "type": 1,
        "description": "Des votes sur lesquels gamble !",
        "options": [
            {
                "name": "intitulee",
                "description": "Intitulé du vote",
                "type": 3,
                "required": True
            },
            {
                "name": "vote1",
                "description": "Première option de vote",
                "type": 3,
                "required": True
            },
            {
                "name": "vote2",
                "description": "Deuxième option de vote",
                "type": 3,
                "required": False
            },
            {
                "name": "vote3",
                "description": "Troisième option de vote",
                "type": 3,
                "required": False
            },
            {
                "name": "vote4",
                "description": "Quatrième option de vote",
                "type": 3,
                "required": False
            },
            {
                "name": "vote5",
                "description": "Cinquième option de vote",
                "type": 3,
                "required": False
            }
        ]
    }
    res: Response = requests.post(f"{API_URL}/applications/{APPLICATION_ID}/commands", json=body,
                                  headers={"Authorization": f"Bot {TOKEN}"})
    print(res)


async def init_commands():
    res = requests.get(f"{API_URL}/applications/{APPLICATION_ID}/commands", headers={"Authorization": f"Bot {TOKEN}"})
    commands = json.loads(res.content)
    for command in commands:
        response = requests.delete(f"{API_URL}/applications/{APPLICATION_ID}/commands/{command['id']}",
                                   headers={"Authorization": f"Bot {TOKEN}"})
        if response.status_code == 204:
            print("Suppression des commandes: Supprimé avec succès !")
        else:
            print("Suppression des commandes: Une erreur est survenue")
    await create_command()


async def create_poll(channel_id: int, label: str, vote_options: list[dict]):
    body: dict = {
        "content": "salut feikrf",
        "components": [
            {
                "type": 1,
                "components": [
                    {
                        "type": 4,
                        "custom_id": "textinput",
                        "label": "Montant de la somme pariée",
                        "min_length": 1,
                        "required": True,
                        "value": "50",
                        "style": 1
                    }
                ]
            }
        ]
    }

    res: Response = requests.post(url=f"{API_URL}/channels/{channel_id}/messages", json=body,
                                  headers={"Authorization": f"Bot {TOKEN}"})
    if res.status_code == 400:
        print(res.json())
    else:
        print(res)


async def identify(websocket):
    payload: dict = {
        "op": OpCode.IDENTIFY,
        "d": {
            "token": TOKEN,
            "properties": {
                "os": "linux",
                "browser": "gambling",
                "device": "gambling"
            },
            "intents": 50364416,
        }
    }
    await websocket.send(json.dumps(payload))


async def create_interaction_response(interaction_id: int, interaction_token: str):
    body = {
        "type": 4,
        "data": {
            "embeds": [
                {
                    "title": "Les paris sont ouvert !",
                    "description": "Vous pouvez parier vos points, ils ne passeront pas en dessous de 50."
                }
            ],
            "poll": {
                "question":
                    {
                        "text": "Question du poll"
                    },
                "answers":
                    [
                        {
                            "answer_id": 1,
                            "poll_media":
                                {
                                    "text": "vote1"
                                }
                        },
                        {
                            "answer_id": 2,
                            "poll_media":
                                {
                                    "text": "vote2"
                                }
                        },
                        {
                            "answer_id": 3,
                            "poll_media":
                                {
                                    "text": "vote3"
                                }
                        }
                    ],
                "allow_multiselect": False,
                "layout_type": 1
            }
        }
    }
    res = requests.post(f"{API_URL}/interactions/{interaction_id}/{interaction_token}/callback", json=body,
                        headers={"Authorization": f"Bot {TOKEN}"})
    print(res)


async def heartbeat(websocket, interval):
    while True:
        await asyncio.sleep(interval / 1000)
        await websocket.send(json.dumps({"op": OpCode.HEARTBEAT, "d": None}))


async def get_event(response: Any):
    match response["t"]:
        case EventTitle.MESSAGE_CREATE:
            print(f'{response["d"]["author"]["username"]}: {response["d"]["content"]}')
        case EventTitle.INTERACTION_CREATE:
            print(response)
            await create_interaction_response(int(response["d"]["id"]), response["d"]["token"])
            # poll_data: list = response["d"]["data"]["options"]
            # channel_id = int(response["d"]["channel_id"])
            # label = poll_data.pop(0)
            # await create_poll(channel_id=channel_id, label=label, vote_options=poll_data)
        case _:
            print(response)


async def connect():
    async with websockets.connect(GATEWAY_URL) as websocket:
        response = json.loads(await websocket.recv())
        heartbeat_interval = response["d"]["heartbeat_interval"]

        asyncio.create_task(heartbeat(websocket, heartbeat_interval))

        await identify(websocket)

        while True:
            response = json.loads(await websocket.recv())
            match response["op"]:
                case OpCode.DISPATCH:
                    await get_event(response)
                case _:
                    print(response)


async def main():
    gateway_connect = asyncio.create_task(connect())
    # await init_commands()
    # await create_poll(840107060901052426, "f", [])
    await gateway_connect


asyncio.run(main())