from datetime import datetime import json import traceback import aiofiles import fastapi from fastapi import Cookie, File, Form, Request, UploadFile, WebSocket, WebSocketDisconnect from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, PlainTextResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.middleware.cors import CORSMiddleware import reporthook import sbenv import searchlib MAX_SEARCH_DAYS = 180 MAX_SHOW_DAYS = 20 ################################ # Core configuration ################################ app = fastapi.FastAPI(docs_url=None) origins = [ 'https://prograde.gg', 'http://localhost', 'http://localhost:5000', 'http://localhost:8080', 'http://localhost:8000' ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=['*'], allow_headers=['*'] ) app.mount('/static', StaticFiles(directory='static'), name='static') tmplts = Jinja2Templates(directory='templates') # TODO Get the path correctly. ################################ # User-facing endpoints ################################ @app.exception_handler(Exception) async def handle_exception(req: Request, exc: Exception): tb = traceback.format_exc() await reporthook.send_report(tb) return PlainTextResponse('error', status_code=500) @app.get('/') async def render_main(req: Request): raw_articles = await load_days_from_file('testresults.json') converted = convert_days_from_articles(raw_articles) num_days = calc_num_days(converted) p = { 'sb': { 'num_days': num_days, 'days': converted }, 'request': req, } return tmplts.TemplateResponse('main.htm', p) ################################ # API endpoints ################################ @app.post('/api/addarticle') async def handle_addarticle(req: Request): if not check_admin_token(req): return JSONResponse(status_code=403, content={'error': 'forbidden'}) body = await req.json() add_article(body) return {'status': 'OK'} def add_article(article): # TODO pass ################################ # Utilities ################################ def check_admin_token(req: Request): ak = sbenv.get_admin_key() if ak is None: raise RuntimeError('checked api endpoint without key loaded') if ak == 'UNSAFE_TESTING': return True if 'Authorization' in req.headers: auth = req.headers['Authorization'] if not auth.startswith('Bearer '): return False tok = auth[len('Bearer '):] return tok == sbenv.get_admin_key() else: return False async def load_days_from_file(path): async with aiofiles.open(path, mode='r') as f: contents = await f.read() return json.loads(contents) def convert_days_from_articles(rarts): processed = searchlib.process_results(rarts) output = [] for dstr, arts in processed.items(): day = { 'date': dstr, 'links': [convert_article(a) for a in arts['pass']], 'maybe_links': [convert_article(a) for a in arts['maybe']] } if len(day['links']) > 0: output.append(day) if len(output) > MAX_SHOW_DAYS: break return output def convert_article(a): return { 'url': a['url'], 'title': a['gtitle'], 'slug': a['slug'], } DATE_FORMAT = "%Y-%m-%d" def calc_num_days(dayslist): today = datetime.now() lowest = -1 for d in dayslist: pd = datetime.strptime(d['date'], DATE_FORMAT) diff = today - pd ndays = diff.days if ndays < lowest or lowest == -1: lowest = ndays return lowest