1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import asyncio
import json
import sys
import threading
import websockets
from typing import Coroutine
from discord.intents import Intents, gen_number
from discord.user import User
class Client:
def __init__(self, intents: list[Intents]):
self.gateway = None
self.loop = asyncio.get_event_loop()
self.code = gen_number(intents)
async def connect(self, token: str, intent_code: int):
async with websockets.connect("wss://gateway.discord.gg/?v=10&encoding=json") as gateway:
hello = await gateway.recv()
self.gateway = gateway
threading.Thread(target=self.loop.run_forever).start()
heartbeat = asyncio.run_coroutine_threadsafe(
self.heartbeat(gateway, json.loads(hello)['d']['heartbeat_interval']), self.loop)
identify = {
"op": 2,
"d": {
"token": token,
"intents": intent_code,
"properties": {
"os": sys.platform,
"browser": "discobra",
"device": "discobra"
}
}
}
await gateway.send(json.dumps(identify))
ready = await gateway.recv()
if (hasattr(self, 'on_ready')):
await getattr(self, 'on_ready')()
self.user = User(json.loads(ready)['d']['user'])
async def heartbeat(self, gateway: websockets.WebSocketClientProtocol, interval: int):
while True:
await asyncio.sleep(interval / 1000)
heartbeat = {
"op": 1,
"d": None
}
await gateway.send(json.dumps(heartbeat))
ack = await gateway.recv()
def event(self, coro: Coroutine, /) -> Coroutine:
if not asyncio.iscoroutinefunction(coro):
raise TypeError('event registered must be a coroutine function')
setattr(self, coro.__name__, coro)
return coro
def run(self, token: str):
self.token = token
asyncio.run(self.connect(self.token, self.code))
|