#!/usr/bin/python3
import asyncio
import cgitb
import glob
import html
from aiohttp import web
import io
import mimetypes
import moth
import logging
import os
import pathlib
import random
import shutil
import socketserver
import sys
import traceback
import mothballer
sys.dont_write_bytecode = True # Don't write .pyc files
def mkseed():
return bytes(random.choice(b'abcdef0123456789') for i in range(40)).decode('ascii')
class Page:
def __init__(self, title, depth=0):
self.title = title
if depth:
self.base = "/".join([".."] * depth)
else:
self.base = "."
self.body = io.StringIO()
self.scripts = []
def add_script(self, path):
self.scripts.append(path)
def write(self, s):
self.body.write(s)
def text(self):
ret = io.StringIO()
ret.write("\n")
ret.write("\n")
ret.write(" \n")
ret.write(" {}\n".format(self.title))
ret.write(" \n".format(self.base))
for s in self.scripts:
ret.write(" {}\n".format(s))
ret.write(" \n")
ret.write(" \n")
ret.write(" {}
\n".format(self.title))
ret.write(" \n")
ret.write(self.body.getvalue())
ret.write("
\n")
ret.write(" \n")
ret.write("\n")
return ret.getvalue()
def response(self, request):
return web.Response(text=self.text(), content_type="text/html")
async def handle_front(request):
p = Page("Devel Server", 0)
p.write("Yo, it's the front page!
")
p.write("")
p.write("If you use this development server to run a contest, you are a fool.
")
return p.response(request)
async def handle_puzzlelist(request):
seed = request.query.get("seed", mkseed())
p = Page("Puzzle Categories", 1)
p.write("seed = {}
".format(seed))
p.write("")
for i in sorted(glob.glob(os.path.join(request.app["puzzles_dir"], "*", ""))):
bn = os.path.basename(i.strip('/\\'))
p.write("- ")
p.write("[mb]".format(cat=bn, seed=seed))
p.write(" ")
p.write("{cat}".format(cat=bn, seed=seed))
p.write("
")
p.write("
")
return p.response(request)
async def handle_category(request):
seed = request.query.get("seed", mkseed())
category = request.match_info.get("category")
cat = moth.Category(os.path.join(request.app["puzzles_dir"], category), seed)
p = Page("Puzzles in category {}".format(category), 2)
p.write("")
return p.response(request)
async def handle_puzzle(request):
seed = request.query.get("seed", mkseed())
category = request.match_info.get("category")
points = int(request.match_info.get("points"))
cat = moth.Category(os.path.join(request.app["puzzles_dir"], category), seed)
puzzle = cat.puzzle(points)
p = Page("{} puzzle {}".format(category, points), 3)
for s in puzzle.scripts:
p.add_script(s)
p.write("Body
")
p.write("")
p.write(puzzle.html_body())
p.write("
")
p.write("Files
")
p.write("")
for name,puzzlefile in sorted(puzzle.files.items()):
if puzzlefile.visible:
visibility = ''
else:
visibility = '(unlisted)'
p.write('- {filename} {visibility}
'
.format(cat=category,
points=puzzle.points,
filename=name,
visibility=visibility))
p.write("
")
p.write("Answers
")
p.write("Input box (for scripts): ")
p.write("
")
assert puzzle.answers, 'No answers defined'
for a in puzzle.answers:
p.write("{}
")
p.write("Authors
{}
".format(', '.join(puzzle.get_authors())))
p.write("Summary
{}
".format(puzzle.summary))
if puzzle.logs:
p.write("Debug Log
")
p.write('')
for l in puzzle.logs:
p.write("- {}
".format(html.escape(l)))
p.write("
")
return p.response(request)
async def handle_puzzlefile(request):
seed = request.query.get("seed", mkseed()).encode('ascii')
category = request.match_info.get("category")
points = int(request.match_info.get("points"))
filename = request.match_info.get("filename")
cat = moth.Category(os.path.join(request.app["puzzles_dir"], category), seed)
puzzle = cat.puzzle(points)
try:
file = puzzle.files[filename]
except KeyError:
return web.Response(status=404)
resp = web.Response()
resp.content_type, _ = mimetypes.guess_type(file.name)
# This is the line where I decided Go was better than Python at multiprocessing
# You should be able to chain the puzzle file's output to the async output,
# without having to block. But if there's a way to do that, it certainly
# isn't documented anywhere.
resp.body = file.stream.read()
return resp
async def handle_mothballer(request):
seed = request.query.get("seed", mkseed())
category = request.match_info.get("category")
try:
catdir = os.path.join(request.app["puzzles_dir"], category)
mb = mothballer.package(category, catdir, seed)
except:
body = cgitb.html(sys.exc_info())
resp = web.Response(text=body, content_type="text/html")
return resp
mb_buf = mb.read()
resp = web.Response(
body=mb_buf,
headers={"Content-Disposition": "attachment; filename={}.mb".format(category)},
content_type="application/octet_stream",
)
return resp
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description="MOTH puzzle development server")
parser.add_argument(
'--puzzles', default='puzzles',
help="Directory containing your puzzles"
)
parser.add_argument(
'--bind', default="127.0.0.1:8080",
help="Bind to ip:port"
)
parser.add_argument(
'--base', default="",
help="Base URL to this server, for reverse proxy setup"
)
args = parser.parse_args()
parts = args.bind.split(":")
addr = parts[0] or "0.0.0.0"
port = int(parts[1])
logging.basicConfig(level=logging.INFO)
mydir = os.path.dirname(os.path.dirname(os.path.realpath(sys.argv[0])))
app = web.Application()
app["puzzles_dir"] = args.puzzles
app["base_url"] = args.base
app.router.add_route("GET", "/", handle_front)
app.router.add_route("GET", "/puzzles/", handle_puzzlelist)
app.router.add_route("GET", "/puzzles/{category}/", handle_category)
app.router.add_route("GET", "/puzzles/{category}/{points}/", handle_puzzle)
app.router.add_route("GET", "/puzzles/{category}/{points}/{filename}", handle_puzzlefile)
app.router.add_route("GET", "/mothballer/{category}", handle_mothballer)
app.router.add_static("/files/", mydir, show_index=True)
web.run_app(app, host=addr, port=port)