forked from AvaCity/avacity-2.0
-
Notifications
You must be signed in to change notification settings - Fork 0
/
web.py
142 lines (119 loc) · 4.25 KB
/
web.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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
import asyncio
import base64
import configparser
import redis
from aiohttp import web
import aiohttp_session
import aiohttp_jinja2
import jinja2
from cryptography import fernet
from aiohttp_session.cookie_storage import EncryptedCookieStorage
import utils.bot_common
routes = web.RouteTableDef()
routes.static("/files", "files")
xml = """<?xml version="1.0" ?>
<cross-domain-policy>
<allow-access-from domain="*" />
</cross-domain-policy>"""
config = configparser.ConfigParser()
config.read("web.ini")
if config["webserver"]["allow_reg"].lower() == "true":
registation = True
else:
registation = False
def get_level(exp):
expSum = 0
i = 0
while expSum <= exp:
i += 1
expSum += i * 50
return i
@routes.get("/")
async def index(request):
session = await aiohttp_session.get_session(request)
context = {}
if "token" not in session:
context["logged_in"] = False
else:
context["logged_in"] = True
context["token"] = session["token"]
return aiohttp_jinja2.render_template("index.html", request,
context=context)
@routes.post("/login")
async def login(request):
session = await aiohttp_session.new_session(request)
data = await request.post()
password = data["password"]
uid = app["redis"].get(f"auth:{password}")
if uid == data["login"]:
session["uid"] = uid
session["token"] = password
raise web.HTTPFound("/")
@routes.get("/logout")
async def logout(request):
session = await aiohttp_session.get_session(request)
if "token" in session:
del session["token"]
del session["uid"]
raise web.HTTPFound("/")
@routes.get("/register")
async def register(request):
if not registation:
return web.Response(text="Регистрация отключена")
uid, password = utils.bot_common.new_account(app["redis"])
return web.Response(text=f"Аккаунт создан, ваш логин - {uid}, "
f"пароль - {password}")
@routes.get("/prelogin")
async def prelogin(request):
if "sid" not in request.query:
raise web.HTTPClientError()
try:
uid = int(request.query["sid"])
except ValueError:
raise web.HTTPClientError()
exp = int(app["redis"].get(f"uid:{uid}:exp"))
return web.json_response({"user": {"bannerNetworkId": None, "reg": 0,
"paymentGroup": "",
"preloginModuleIds": "", "id": uid,
"avatariaLevel": get_level(exp)}})
@routes.post("/social")
async def social(request):
data = await request.post()
if not data["method"]:
raise web.HTTPClientError()
if data["method"] == "getTestUsers":
return web.json_response({"uids": [str(data["user_id"])]})
elif data["method"] in ["getFriends", "getAppFriends"]:
return web.json_response([])
elif data["method"] == "getProfiles":
return web.json_response([{"id": data["uids"]}])
@routes.get("/appconfig.xml")
async def appconfig(request):
session = await aiohttp_session.get_session(request)
if "token" not in session:
raise web.HTTPUnauthorized()
context = {"token": session["token"], "uid": session["uid"],
"ip": config["webserver"]["server_ip"],
"address": config["webserver"]["web_address"]}
return aiohttp_jinja2.render_template("appconfig.xml", request,
context=context)
@routes.get("/crossdomain.xml")
async def crossdomain(requst):
return web.Response(text=xml)
async def main():
global app
app = web.Application()
app.add_routes(routes)
app["redis"] = redis.Redis(decode_responses=True)
fernet_key = fernet.Fernet.generate_key()
secret_key = base64.urlsafe_b64decode(fernet_key)
aiohttp_session.setup(app, EncryptedCookieStorage(secret_key))
aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader("templates"))
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "0.0.0.0", int(config["webserver"]["web_port"]))
await site.start()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()