184 lines
5.0 KiB
Python
184 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# The server part: briding http to logic
|
|
|
|
import sys
|
|
import etcd
|
|
import json
|
|
import datetime
|
|
import inspect
|
|
|
|
from flask import Flask, abort, request, Response
|
|
from flask_restful import reqparse
|
|
|
|
from db import DB
|
|
from challenge import Challenge
|
|
import challenges
|
|
|
|
|
|
INDEX_MESSAGE = """
|
|
Welcome to the ungleich game server!
|
|
|
|
This server is still in development and is running on Nico's
|
|
notebook. In case it is off-line, ping @nico on
|
|
https://chat.ungleich.ch.
|
|
|
|
To play:
|
|
|
|
curl http://{hostname}/challenge
|
|
|
|
To see the high score:
|
|
|
|
curl http://{hostname}/points
|
|
|
|
The code for this game can be found on https://code.ungleich.ch/nico/ungleich-game
|
|
"""
|
|
|
|
POINT_MESSAGE = """
|
|
Point list (aka high score)
|
|
---------------------------
|
|
{}
|
|
|
|
"""
|
|
|
|
CHALLENGE_MESSAGE = """
|
|
The following challenges are available on this server:
|
|
|
|
{}
|
|
|
|
To play, first just curl the URL and if you want to submit solutions,
|
|
post them like this:
|
|
|
|
curl -d user=nico -d network=2a0a:e5c1:101::/64 http://.../challenge/...
|
|
"""
|
|
|
|
|
|
class Game(object):
|
|
def __init__(self, name, etcdclient, etcbase="/ungleichgame/v1"):
|
|
self.client = etcdclient
|
|
self.etcbase = etcbase
|
|
|
|
self.wrapper = DB(etcdclient, self.etcbase)
|
|
|
|
self.app = Flask(name)
|
|
|
|
self.app.add_url_rule('/', 'index', self.index)
|
|
self.app.add_url_rule('/points', 'points', self.points)
|
|
self.app.add_url_rule('/points/<username>', 'points', self.points)
|
|
self.app.add_url_rule('/register', 'register', self.register, methods=['POST'])
|
|
|
|
# etcd paths are below here
|
|
|
|
self.userbase = "{}/user".format(self.etcbase)
|
|
|
|
self.__init_challenges()
|
|
|
|
def __init_challenges(self):
|
|
# Create list of challenges
|
|
self.app.add_url_rule('/challenge', 'list_of_challenges', self.list_of_challenges)
|
|
self.app.add_url_rule('/challenge/', 'list_of_challenges', self.list_of_challenges)
|
|
|
|
self.providers = {}
|
|
|
|
self.list_challenges = []
|
|
self.challenge_instances = []
|
|
self.challenge_names = []
|
|
|
|
for name, obj in inspect.getmembers(challenges):
|
|
if inspect.isclass(obj):
|
|
c = obj(self.wrapper)
|
|
|
|
self.challenge_instances.append(c)
|
|
self.list_challenges.append(obj)
|
|
|
|
self.challenge_names.append(name)
|
|
path = "/challenge/{}".format(name)
|
|
|
|
self.app.add_url_rule(path, name, c.game, methods=['GET', 'POST'])
|
|
|
|
for provider in c.provides:
|
|
if not provider in self.providers:
|
|
self.providers[provider] = []
|
|
|
|
self.providers[provider].append(name)
|
|
|
|
# Update challenges with provider information
|
|
for challenge in self.challenge_instances:
|
|
for requirement in challenge.requires:
|
|
if not requirement in self.providers:
|
|
raise Exception("Unplayable challenge: {}".format(type(challenge).__name__))
|
|
|
|
challenge.dependencies_provided_by[requirement] = self.providers[requirement]
|
|
|
|
|
|
def list_of_challenges(self):
|
|
base = request.base_url
|
|
c = [ "{} ({})".format(name, "{}/{}".format(base, name)) for name in self.challenge_names ]
|
|
return CHALLENGE_MESSAGE.format("\n".join(c))
|
|
|
|
|
|
def get_points(self):
|
|
""" Returns a dict['username'] = points """
|
|
|
|
user_points = {}
|
|
|
|
path = "{}/".format(self.userbase)
|
|
users = self.wrapper.read_key_or_none(path)
|
|
|
|
if users:
|
|
for user in users.children:
|
|
username = user.key.replace(path,"")
|
|
user_points[username] = 0
|
|
|
|
point_path = "{}/points".format(user.key)
|
|
points = self.wrapper.read_key_or_none(point_path, recursive=True)
|
|
|
|
if not points:
|
|
continue
|
|
|
|
for challenge in points.children:
|
|
user_points[username] += int(challenge.value)
|
|
|
|
return user_points
|
|
|
|
def index(self):
|
|
return INDEX_MESSAGE.format(hostname=request.headers['Host'])
|
|
|
|
def points(self, username=None):
|
|
point_list = self.get_points()
|
|
res = []
|
|
if not point_list:
|
|
return Response("No winners yet!")
|
|
|
|
if username:
|
|
userpoints = 0
|
|
|
|
if username in point_list:
|
|
userpoints = point_list[username]
|
|
|
|
return "{} has {} points".format(username, userpoints)
|
|
|
|
for k, v in point_list.items():
|
|
res.append("{} has {} points".format(k, v))
|
|
|
|
return POINT_MESSAGE.format("\n".join(res))
|
|
|
|
def register(self):
|
|
args = require_args("user")
|
|
|
|
cur = self.wrapper.get_user_key_or_none(args['user'], "registered_at")
|
|
value = str(datetime.datetime.now())
|
|
|
|
if cur:
|
|
value = cur.value
|
|
else:
|
|
self.wrapper.set_user_key(args['user'], "registered_at", value)
|
|
|
|
return "Registered at: {}\n".format(value)
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
g = Game(__name__, etcd.Client(port=2379, host='[::1]'))
|
|
g.app.run(host="::", port='5002', debug=False)
|