237 lines
7.7 KiB
Python
237 lines
7.7 KiB
Python
import datetime
|
|
import os
|
|
from typing import Any, IO, Dict, Iterator, List, Tuple, Union
|
|
|
|
import jinja2.exceptions
|
|
from flask import request, redirect, url_for, render_template, Response, session, abort, send_from_directory
|
|
from pony.orm import count, commit, db_session
|
|
from werkzeug.utils import secure_filename
|
|
|
|
from worlds.AutoWorld import AutoWorldRegister
|
|
from . import app, cache
|
|
from .models import Seed, Room, Command, UUID, uuid4
|
|
|
|
|
|
def get_world_theme(game_name: str):
|
|
if game_name in AutoWorldRegister.world_types:
|
|
return AutoWorldRegister.world_types[game_name].web.theme
|
|
return 'grass'
|
|
|
|
|
|
@app.errorhandler(404)
|
|
@app.errorhandler(jinja2.exceptions.TemplateNotFound)
|
|
def page_not_found(err):
|
|
return render_template('404.html'), 404
|
|
|
|
|
|
# Start Playing Page
|
|
@app.route('/start-playing')
|
|
@cache.cached()
|
|
def start_playing():
|
|
return render_template(f"startPlaying.html")
|
|
|
|
|
|
# Game Info Pages
|
|
@app.route('/games/<string:game>/info/<string:lang>')
|
|
@cache.cached()
|
|
def game_info(game, lang):
|
|
return render_template('gameInfo.html', game=game, lang=lang, theme=get_world_theme(game))
|
|
|
|
|
|
# List of supported games
|
|
@app.route('/games')
|
|
@cache.cached()
|
|
def games():
|
|
worlds = {}
|
|
for game, world in AutoWorldRegister.world_types.items():
|
|
if not world.hidden:
|
|
worlds[game] = world
|
|
return render_template("supportedGames.html", worlds=worlds)
|
|
|
|
|
|
@app.route('/tutorial/<string:game>/<string:file>/<string:lang>')
|
|
@cache.cached()
|
|
def tutorial(game, file, lang):
|
|
return render_template("tutorial.html", game=game, file=file, lang=lang, theme=get_world_theme(game))
|
|
|
|
|
|
@app.route('/tutorial/')
|
|
@cache.cached()
|
|
def tutorial_landing():
|
|
return render_template("tutorialLanding.html")
|
|
|
|
|
|
@app.route('/faq/<string:lang>/')
|
|
@cache.cached()
|
|
def faq(lang: str):
|
|
import markdown
|
|
with open(os.path.join(app.static_folder, "assets", "faq", secure_filename(lang)+".md")) as f:
|
|
document = f.read()
|
|
return render_template(
|
|
"markdown_document.html",
|
|
title="Frequently Asked Questions",
|
|
html_from_markdown=markdown.markdown(
|
|
document,
|
|
extensions=["toc", "mdx_breakless_lists"],
|
|
extension_configs={
|
|
"toc": {"anchorlink": True}
|
|
}
|
|
),
|
|
)
|
|
|
|
|
|
@app.route('/glossary/<string:lang>/')
|
|
@cache.cached()
|
|
def glossary(lang: str):
|
|
import markdown
|
|
with open(os.path.join(app.static_folder, "assets", "glossary", secure_filename(lang)+".md")) as f:
|
|
document = f.read()
|
|
return render_template(
|
|
"markdown_document.html",
|
|
title="Glossary",
|
|
html_from_markdown=markdown.markdown(
|
|
document,
|
|
extensions=["toc", "mdx_breakless_lists"],
|
|
extension_configs={
|
|
"toc": {"anchorlink": True}
|
|
}
|
|
),
|
|
)
|
|
|
|
|
|
@app.route('/seed/<suuid:seed>')
|
|
def view_seed(seed: UUID):
|
|
seed = Seed.get(id=seed)
|
|
if not seed:
|
|
abort(404)
|
|
return render_template("viewSeed.html", seed=seed, slot_count=count(seed.slots))
|
|
|
|
|
|
@app.route('/new_room/<suuid:seed>')
|
|
def new_room(seed: UUID):
|
|
seed = Seed.get(id=seed)
|
|
if not seed:
|
|
abort(404)
|
|
room = Room(seed=seed, owner=session["_id"], tracker=uuid4())
|
|
commit()
|
|
return redirect(url_for("host_room", room=room.id))
|
|
|
|
|
|
def _read_log(log: IO[Any], offset: int = 0) -> Iterator[bytes]:
|
|
marker = log.read(3) # skip optional BOM
|
|
if marker != b'\xEF\xBB\xBF':
|
|
log.seek(0, os.SEEK_SET)
|
|
log.seek(offset, os.SEEK_CUR)
|
|
yield from log
|
|
log.close() # free file handle as soon as possible
|
|
|
|
|
|
@app.route('/log/<suuid:room>')
|
|
def display_log(room: UUID) -> Union[str, Response, Tuple[str, int]]:
|
|
room = Room.get(id=room)
|
|
if room is None:
|
|
return abort(404)
|
|
if room.owner == session["_id"]:
|
|
file_path = os.path.join("logs", str(room.id) + ".txt")
|
|
try:
|
|
log = open(file_path, "rb")
|
|
range_header = request.headers.get("Range")
|
|
if range_header:
|
|
range_type, range_values = range_header.split('=')
|
|
start, end = map(str.strip, range_values.split('-', 1))
|
|
if range_type != "bytes" or end != "":
|
|
return "Unsupported range", 500
|
|
# NOTE: we skip Content-Range in the response here, which isn't great but works for our JS
|
|
return Response(_read_log(log, int(start)), mimetype="text/plain", status=206)
|
|
return Response(_read_log(log), mimetype="text/plain")
|
|
except FileNotFoundError:
|
|
return Response(f"Logfile {file_path} does not exist. "
|
|
f"Likely a crash during spinup of multiworld instance or it is still spinning up.",
|
|
mimetype="text/plain")
|
|
|
|
return "Access Denied", 403
|
|
|
|
|
|
@app.post("/room/<suuid:room>")
|
|
def host_room_command(room: UUID):
|
|
room: Room = Room.get(id=room)
|
|
if room is None:
|
|
return abort(404)
|
|
|
|
if room.owner == session["_id"]:
|
|
cmd = request.form["cmd"]
|
|
if cmd:
|
|
Command(room=room, commandtext=cmd)
|
|
commit()
|
|
return redirect(url_for("host_room", room=room.id))
|
|
|
|
|
|
@app.get("/room/<suuid:room>")
|
|
def host_room(room: UUID):
|
|
room: Room = Room.get(id=room)
|
|
if room is None:
|
|
return abort(404)
|
|
|
|
now = datetime.datetime.utcnow()
|
|
# indicate that the page should reload to get the assigned port
|
|
should_refresh = ((not room.last_port and now - room.creation_time < datetime.timedelta(seconds=3))
|
|
or room.last_activity < now - datetime.timedelta(seconds=room.timeout))
|
|
with db_session:
|
|
room.last_activity = now # will trigger a spinup, if it's not already running
|
|
|
|
browser_tokens = "Mozilla", "Chrome", "Safari"
|
|
automated = ("update" in request.args
|
|
or "Discordbot" in request.user_agent.string
|
|
or not any(browser_token in request.user_agent.string for browser_token in browser_tokens))
|
|
|
|
def get_log(max_size: int = 0 if automated else 1024000) -> str:
|
|
if max_size == 0:
|
|
return "…"
|
|
try:
|
|
with open(os.path.join("logs", str(room.id) + ".txt"), "rb") as log:
|
|
raw_size = 0
|
|
fragments: List[str] = []
|
|
for block in _read_log(log):
|
|
if raw_size + len(block) > max_size:
|
|
fragments.append("…")
|
|
break
|
|
raw_size += len(block)
|
|
fragments.append(block.decode("utf-8"))
|
|
return "".join(fragments)
|
|
except FileNotFoundError:
|
|
return ""
|
|
|
|
return render_template("hostRoom.html", room=room, should_refresh=should_refresh, get_log=get_log)
|
|
|
|
|
|
@app.route('/favicon.ico')
|
|
def favicon():
|
|
return send_from_directory(os.path.join(app.root_path, "static", "static"),
|
|
'favicon.ico', mimetype='image/vnd.microsoft.icon')
|
|
|
|
|
|
@app.route('/discord')
|
|
def discord():
|
|
return redirect("https://discord.gg/8Z65BR2")
|
|
|
|
|
|
@app.route('/datapackage')
|
|
@cache.cached()
|
|
def get_datapackage():
|
|
"""A pretty print version of /api/datapackage"""
|
|
from worlds import network_data_package
|
|
import json
|
|
return Response(json.dumps(network_data_package, indent=4), mimetype="text/plain")
|
|
|
|
|
|
@app.route('/index')
|
|
@app.route('/sitemap')
|
|
@cache.cached()
|
|
def get_sitemap():
|
|
available_games: List[Dict[str, Union[str, bool]]] = []
|
|
for game, world in AutoWorldRegister.world_types.items():
|
|
if not world.hidden:
|
|
has_settings: bool = isinstance(world.web.options_page, bool) and world.web.options_page
|
|
available_games.append({ 'title': game, 'has_settings': has_settings })
|
|
return render_template("siteMap.html", games=available_games)
|