aci/backend/app.py

#!/usr/bin/env python3

from typing import Union
import hashlib
import re
import sqlite3

from fastapi import Depends, FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware

import uvicorn

from aci.db import get_db
from aci.migrate import run_migrations
from aci.models import CodeSubmission
from aci import utils
from aci import proxy

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origin_regex=r"moz-extension://.*",
    allow_methods=["GET", "POST"],
    allow_headers="*",
)

@app.middleware('http')
async def badbots_block(req: Request, call_next):
    ua = req.headers.get('User-Agent', '')
    for bot in utils.BAD_BOTS:
        if re.match(bot, ua):
            return Response('forbidden', status_code=403)
    return await call_next(req)

@app.get('/')
def index():
    return {
        '🏳️‍⚧️': 'trans rights!',
    }

@app.post('/api/codes')
def submit_code(date: str, time: str, activity: str, space: str, submission: CodeSubmission, db: sqlite3.Cursor=Depends(get_db)):
    # TODO: validate parameters meet expected form.
    db.execute('INSERT INTO code_results (date, time, space, activity, code, result) VALUES (?,?,?,?,?,?)', (date, time, space, activity, submission.code, submission.success))
    return {'ok': True}

@app.get('/api/codes')
def current_codes(db: sqlite3.Cursor=Depends(get_db)):
    result = db.execute('SELECT date, time, space, activity, code, result FROM code_results WHERE date = ?', (utils.today(),))
    codes = result.fetchall()
    code_scores = {}
    code_info = {}
    for date, time, space, activity, code, result in codes:
        k = str((date, time, space, activity))
        code_info[k] = (date, time, space, activity)
        if k not in code_scores:
            code_scores[k] = {}
        if code not in code_scores[k]:
            code_scores[k][code] = 0
        code_scores[k][code] += 1 if result else -1
    ret = []
    for k, v in code_scores.items():
        date, time, space, activity = code_info[k]
        codes = list(map(lambda c: {'code': c[0], 'score': c[1]}, v.items()))
        codes.sort(key=lambda c: -c['score'])
        ret.append({
            'date': date,
            'time': time,
            'space': space,
            'activity': activity,
            'codes': codes,
        })
    return {
        'activities': ret,
    }

@app.get('/api/codes/query')
def query_codes(date: str, time: str, activity: str, space: str, db: sqlite3.Cursor=Depends(get_db)):
    result = db.execute('SELECT code, result FROM code_results WHERE date = ? AND time = ? AND activity = ? AND space = ?', (date, time, activity, space))
    codes = result.fetchall()
    code_scores = {}
    for code, success in codes:
        if code not in code_scores:
            code_scores[code] = 0
        code_scores[code] += 1 if success else -1
    return {
        'date': date,
        'time': time,
        'space': space,
        'activity': activity,
        'codes': list(map(lambda c: {'code': c[0], 'score': c[1]}, code_scores.items()))
    }

app.include_router(proxy.router)

if __name__ == '__main__':
    run_migrations()
    uvicorn.run(app, host='0.0.0.0')