86 lines
2.3 KiB
Python
86 lines
2.3 KiB
Python
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)
|
|
|
|
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 send_message(channel_id: int, message: str):
|
|
body: dict= {
|
|
"content": message,
|
|
"tts": False,
|
|
"embeds": [{
|
|
"title": "Hello, Embed!",
|
|
"description": "This is an embedded message."
|
|
}]
|
|
}
|
|
|
|
res: Response = requests.post(url=f"{API_URL}/channels/{channel_id}/messages", json=body, headers={"Authorization": f"Bot {TOKEN}"})
|
|
print(res)
|
|
|
|
|
|
async def identify(websocket):
|
|
payload: dict = {
|
|
"op": OpCode.IDENTIFY,
|
|
"d": {
|
|
"token": TOKEN,
|
|
"properties": {
|
|
"os": "linux",
|
|
"browser": "gambling",
|
|
"device": "gambling"
|
|
},
|
|
"intents": 65024,
|
|
}
|
|
}
|
|
await websocket.send(json.dumps(payload))
|
|
|
|
|
|
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 _:
|
|
print("Unknown event")
|
|
|
|
|
|
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())
|
|
#print("Received: ", response)
|
|
match response["op"]:
|
|
case OpCode.DISPATCH:
|
|
await get_event(response)
|
|
case _:
|
|
print("osef")
|
|
|
|
|
|
async def main():
|
|
gateway_connect = asyncio.create_task(connect())
|
|
await gateway_connect
|
|
|
|
asyncio.run(main()) |