133 lines
3.4 KiB
Python
133 lines
3.4 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Coroutine, TypeVar
|
|
|
|
import aiohttp
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import (
|
|
FileResponse,
|
|
JSONResponse,
|
|
PlainTextResponse,
|
|
RedirectResponse,
|
|
Response,
|
|
)
|
|
|
|
app = FastAPI()
|
|
|
|
root = Path(__file__).parent / "static"
|
|
|
|
|
|
@app.get("/")
|
|
async def home():
|
|
return FileResponse(root / "home.html")
|
|
|
|
|
|
@app.get("/operator/")
|
|
async def operatorhome():
|
|
return FileResponse(root / "operator.html")
|
|
|
|
|
|
@app.get("/login/")
|
|
async def login():
|
|
return FileResponse(root / "login.html")
|
|
|
|
|
|
ubot = "http://v6d3music/"
|
|
ussv = "http://sessionservice/"
|
|
|
|
|
|
@app.get("/authlink/")
|
|
async def authlink():
|
|
async with aiohttp.ClientSession() as s, s.get(f"{ussv}authlink/") as response:
|
|
return PlainTextResponse(await response.text())
|
|
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
async def url_get(
|
|
url: str, params: dict[str, str], handle: Callable[[aiohttp.ClientResponse], Coroutine[Any, Any, T]]
|
|
) -> T:
|
|
async with aiohttp.ClientSession() as s, s.get(url, params=params) as response:
|
|
return await handle(response)
|
|
|
|
|
|
async def url_post(
|
|
url: str, params: dict[str, str], json: dict, handle: Callable[[aiohttp.ClientResponse], Coroutine[Any, Any, T]]
|
|
) -> T:
|
|
async with aiohttp.ClientSession() as s, s.post(url, params=params, json=json) as response:
|
|
print(await response.read())
|
|
return await handle(response)
|
|
|
|
|
|
async def repeat(repeated: Callable[[], Coroutine[Any, Any, T]]) -> T:
|
|
for _ in range(60):
|
|
try:
|
|
return await repeated()
|
|
except aiohttp.ClientConnectorError:
|
|
await asyncio.sleep(1)
|
|
raise HTTPException(504)
|
|
|
|
|
|
async def handle_auth(response: aiohttp.ClientResponse):
|
|
if response.status == 200:
|
|
return RedirectResponse("/")
|
|
else:
|
|
return Response(content=await response.read(), media_type=response.content_type, status_code=response.status)
|
|
|
|
|
|
@app.get("/auth/")
|
|
async def auth(session: str | None = None, state: str | None = None, code: str | None = None):
|
|
match session, state, code:
|
|
case str() as session, str() as state, str() as code:
|
|
params: dict[str, str] = {"session": session, "state": state, "code": code}
|
|
return await repeat(lambda: url_get(f"{ussv}auth/", params, handle_auth))
|
|
case None, str(), str():
|
|
return FileResponse(root / "auth.html")
|
|
case _:
|
|
raise HTTPException(400)
|
|
|
|
|
|
async def handle_json(response: aiohttp.ClientResponse):
|
|
return JSONResponse(await response.json())
|
|
|
|
|
|
@app.get("/state/")
|
|
async def state(session: str):
|
|
return await repeat(lambda: url_get(f"{ussv}state/", {"session": session}, handle_json))
|
|
|
|
|
|
@app.get("/status/")
|
|
async def status(session: str):
|
|
return await repeat(lambda: url_get(f"{ussv}status/", {"session": session}, handle_json))
|
|
|
|
|
|
@app.get("/main.js")
|
|
async def mainjs():
|
|
return FileResponse(root / "main.js")
|
|
|
|
|
|
@app.get("/operator.js")
|
|
async def operatorjs():
|
|
return FileResponse(root / "operator.js")
|
|
|
|
|
|
@app.get("/main.css")
|
|
async def maincss():
|
|
return FileResponse(root / "main.css")
|
|
|
|
|
|
@app.get("/operator.css")
|
|
async def operatorcss():
|
|
return FileResponse(root / "operator.css")
|
|
|
|
|
|
@app.post("/api/")
|
|
async def api(json: dict, session: str):
|
|
return await repeat(lambda: url_post(f"{ubot}api/", {"session": session}, json, handle_json))
|
|
|
|
|
|
@app.get("/win/")
|
|
async def win():
|
|
raise HTTPException(402)
|