-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathadmin.py
165 lines (142 loc) · 6.11 KB
/
admin.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
# (c) 2019–2021 Vladimír Štill <[email protected]>
import aiohttp_jinja2 # type: ignore
import config
import dateutil.parser
import datetime
import decimal
import functor
from db import DB
from dataclasses import dataclass
from aiohttp import web
from typing import Optional, Dict, Any
def parse_date(date: Optional[str], time: Optional[str],
defhour: int = 0, defminute: int = 0,
defsecond: int = 0, defmicrosecond: int = 0) \
-> Optional[datetime.datetime]:
try:
dt = dateutil.parser.parse(f"{date} {time or ''}")
if not time:
dt = dt.replace(hour=defhour, minute=defminute)
return dt.replace(second=defsecond, microsecond=defmicrosecond)
except ValueError:
return None
@dataclass
class Admin:
req: web.Request
conf: config.Config
user: str
course: Optional[str]
page: Optional[str]
async def dispatch(self, page: Optional[str]) -> web.Response:
if self.course is None:
return await self.summary()
course = self.conf.courses.get(self.course)
if course is None:
return await self._e404(f"course {self.course}")
if self.user not in course.authorized:
return self._forbidden()
if self.page is None:
return await self.log_summary(course)
if self.page == "log":
return await self.log(course)
if self.page == "log_detail":
id_ = self.req.query.get("id")
if id_ is not None and id_.isdigit():
return await self.log_detail(course, int(id_))
return await self._e404("id not specified or invalid")
return await self._e404(f"page {page}")
async def _e404(self, extra: str = "") -> web.Response:
if extra:
extra = f" ({extra})"
return web.Response(status=404, text=f"404 not found{extra}")
async def summary(self) -> web.Response:
def disp(i: int, val: Any) -> Any:
if 0 < i < 4 or 4 < i < 8:
return val
if isinstance(val, bytes):
return val.decode("utf-8")
if isinstance(val, (decimal.Decimal, float)):
return round(val, 1)
if val is None:
return "–"
return val
async with DB(self.conf).connect() as conn:
stats_ = await conn.fetch("select * from usage order by req desc")
stats = [[disp(i, r) for i, r in enumerate(row)]
for row in stats_]
since = await conn.fetchval("""
select stamp from eval_log order by stamp asc limit 1
""")
return self._render("admin/summary.html.j2",
stats=stats,
since=since.replace(microsecond=0))
async def log_summary(self, course: config.Course) -> web.Response:
assert self.course is not None
async with DB(self.conf).connect() as conn:
dates = await conn.fetch("""
select stamp :: date, count(*) from eval_log
where course = $1
group by ( stamp :: date )
order by stamp desc
""", course.name.encode('utf-8'))
return self._render("admin/log_summary.html.j2",
dates=dates,
today=datetime.datetime.now().date())
async def log(self, course: config.Course) -> web.Response:
from_ = parse_date(self.req.query.get("from"),
self.req.query.get("from_time")) \
or datetime.datetime(1, 1, 1)
to = parse_date(self.req.query.get("to"),
self.req.query.get("to_time"),
23, 59, 59, 999999) \
or datetime.datetime(9999, 12, 31)
async with DB(self.conf).connect() as conn:
rows = await conn.fetch("""
select stamp as timestamp,
author,
convert_from(question, 'UTF8') as question,
convert_from(option, 'UTF8') as option,
hint,
result,
id
from eval_log
where course = $1
and stamp >= $2
and stamp <= $3
order by stamp asc
""", course.name.encode('utf-8'), from_, to)
return self._render("admin/log.html.j2",
**{"rows": rows, "from": from_, "to": to})
async def log_detail(self, course: config.Course, eval_id: int) \
-> web.Response:
async with DB(self.conf).connect() as conn:
row = await conn.fetchrow("""
select author,
convert_from(question, 'UTF8') as question,
convert_from(option, 'UTF8') as option,
hint,
result,
convert_from(output, 'UTF8') as output,
convert_from(errors, 'UTF8') as errors,
stamp,
convert_from(answer, 'UTF8') as answer,
id
from eval_log
where course = $1 and id = $2
""", course.name.encode('utf-8'), eval_id)
return self._render("admin/log_detail.html.j2", entry=row)
def _render(self, template: str, **extra: Any) -> web.Response:
out = {"config": self.conf.to_dict(),
"user": self.user,
"course": self.course}
for k, v in extra.items():
out[k] = v
return aiohttp_jinja2.render_template( # type: ignore
template, self.req, out)
def _forbidden(self) -> web.Response:
return web.Response(status=403, text="403 Forbidden")
async def get(req: web.Request, conf: config.Config, user: str,
course: Optional[str], page: Optional[str]) -> web.Response:
admin = Admin(req, conf, user, course, page)
return await admin.dispatch(page)
# vim: colorcolumn=80 expandtab sw=4 ts=4