-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
61 lines (47 loc) · 2.16 KB
/
main.py
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 logging
from os import getenv
from urllib.parse import urlencode
from base64 import urlsafe_b64encode, urlsafe_b64decode
import uvloop
from aiohttp import ClientSession
import defectio
from defectio.ext import commands
import srsly
uvloop.install()
logging.basicConfig(level=logging.DEBUG)
baseurl = "http://twenty_api:8000/twenty/"
bot = commands.Bot(command_prefix="_")
async def fetch_endpoint(url: str, param: dict = None) -> dict:
param['agent'] = 'revolt'
async with ClientSession() as session:
async with session.get(url + urlencode(param)) as response:
return await response.json()
async def message_hook(message, ID: str, bot) -> None:
new_data = srsly.msgpack_dumps({"channel_id": message.channel.id, "message_id": message.id}, protocol=-1)
data = await fetch_endpoint(url=f"{baseurl}set?", param={"ID": ID, "prefix": "message", "data": urlsafe_b64encode(new_data)})
old_data = srsly.msgpack_loads(urlsafe_b64decode(s=data['old_data']))
await bot.http.delete_message(channel_id=old_data['channel_id'], message_id=old_data['message_id'])
async def process_move(ctx, direction: str) -> None:
await ctx.message.delete()
data = await fetch_endpoint(url=f"{baseurl}move?", param={"ID": ctx.author.id, "action": "left"})
message = await ctx.send(f"score: {data['score']}", file=defectio.File(data['image_path']))
await message_hook(message=message, ID=ctx.author.id, bot=bot)
@bot.command()
async def play(ctx):
await ctx.message.delete()
data = await fetch_endpoint(url=f"{baseurl}new_game?", param={"ID": ctx.author.id, "name": ctx.author.name})
message = await ctx.send(f"score: {data['score']}", file=defectio.File(data['image_path']))
await message_hook(message=message, ID=ctx.author.id, bot=bot)
@bot.command(aliases=["l"])
async def left(ctx):
await process_move(ctx, direction='left')
@bot.command(aliases=["r"])
async def right(ctx):
await process_move(ctx, direction='right')
@bot.command(aliases=["u"])
async def up(ctx):
await process_move(ctx, direction='up')
@bot.command(aliases=["d"])
async def down(ctx):
await process_move(ctx, direction='down')
bot.run(getenv("TOKEN"))