99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
#!/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')
|