File size: 6,091 Bytes
806e511 247f100 806e511 247f100 806e511 | 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 | """The developer's status card: is each part of the system working, which version runs and since when (PRD DEV-5).
Only a developer may read it. Nothing here returns a key or a secret: a setting is reported as set or not set. The database and the
push key are really checked; sign-in, AI and pictures are reported from the settings (checking them would call outside services)."""
from __future__ import annotations
import platform
import time
from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from . import clock, version
from .auth import signed_in
router = APIRouter(prefix="/api/dev")
OK, WARN, ERROR, OFF = "ok", "warn", "error", "off"
def _check(key: str, name: str, state: str, detail: str) -> dict:
return {"key": key, "name": name, "state": state, "detail": detail}
def _database(store, supabase_on: bool) -> dict:
if not supabase_on:
return _check("database", "诪住讚 讛谞转讜谞讬诐", WARN, "讝讬讻专讜谉 讘诇讘讚: 砖讜诐 讚讘专 诇讗 谞砖诪专 讗讞专讬 讛驻注诇讛 诪讞讚砖")
raw = getattr(store, "_store", store) # past the cache, so the database itself answers
started = time.monotonic()
try:
raw.roles_of("status-check@mathly.invalid")
except Exception as exc: # any failure is the answer, not a crash of the card
return _check("database", "诪住讚 讛谞转讜谞讬诐", ERROR, f"诇讗 注谞讛 ({type(exc).__name__})")
return _check("database", "诪住讚 讛谞转讜谞讬诐", OK, f"Supabase 注谞讛 转讜讱 {round((time.monotonic() - started) * 1000)} 讗诇驻讬讜转 砖谞讬讬讛")
def _push(pusher, key: str) -> dict:
if not key:
return _check("push", "讛转专讗讜转 诇讟诇驻讜谉", OFF, "讗讬谉 诪驻转讞: 讛讛转专讗讜转 讻讘讜讬讜转")
if not getattr(pusher, "on", False): # the key is set but the server could not use it
return _check("push", "讛转专讗讜转 诇讟诇驻讜谉", ERROR, "讛诪驻转讞 诇讗 转拽讬谉: 讛讛转专讗讜转 讻讘讜讬讜转")
return _check("push", "讛转专讗讜转 诇讟诇驻讜谉", OK, "诪驻转讞 转拽讬谉")
def checks(app) -> list[dict]:
s, state = app.state.settings, app.state
bank = state.bank
return [
_database(state.store, s.supabase_on),
_check("sign_in", "讛转讞讘专讜转", OK if s.supabase_on else WARN,
"Google 讚专讱 Supabase" if s.supabase_on else "讻谞讬住转 驻讬转讜讞 注诐 讻转讜讘转 诪讬讬诇 讘诇讘讚"),
_check("bank", "诪讗讙专 讛砖讗诇讜转", OK, f"{len(bank.lessons)} 砖讬注讜专讬诐, {len(bank.skills)} 诪讬讜诪谞讜讬讜转"),
_check("ai", "注讝专转 AI", OK if s.anthropic_key else OFF, "诪驻转讞 诪讜讙讚专" if s.anthropic_key else "讗讬谉 诪驻转讞: 讛注讝专讛 讻讘讜讬讛"),
_push(state.pusher, s.vapid_private_key),
_check("pictures", "转诪讜谞讜转 转砖讜讘讛", OK if s.supabase_on else WARN,
f"讚诇讬 驻专讟讬 {s.answer_bucket}" if s.supabase_on else "讘讝讬讻专讜谉 讘诇讘讚"),
_check("cleanup", "谞讬拽讜讬 砖注转讬", OK if getattr(state, "cleanup_stop", None) else OFF,
"驻讜注诇" if getattr(state, "cleanup_stop", None) else "诇讗 驻讜注诇 讘诇讬 Supabase"),
]
def overall(found: list[dict]) -> str:
states = {c["state"] for c in found}
return ERROR if ERROR in states else WARN if WARN in states else OK
def developer(request: Request, user: dict = Depends(signed_in)) -> dict:
"""The "dev.status" permission, which only the developer role has. The developer role comes only from the settings, so it is
checked there and not through the roles in the database: the card must still open when the database is down."""
if user["email"].strip().lower() not in request.app.state.settings.developers:
raise HTTPException(403, "讗讬谉 讛专砖讗讛 诇驻注讜诇讛 讛讝讜")
return user
@router.get("/status")
def status(request: Request, user: dict = Depends(developer)) -> dict:
app = request.app
found = checks(app)
now = clock.now()
started = app.state.started_at
return {"overall": overall(found), "checks": found, "version": app.state.version,
"started_at": started.isoformat(), "uptime_seconds": max(0, int((now - started).total_seconds())), "now": now.isoformat(),
"python": platform.python_version()}
ACTIVITY_DAYS = (7, 30)
def activity(rows: list[dict], last, days: int) -> dict:
"""Per Israel day from `last - days + 1` to `last` (days without activity included, as zeros): how many people were active,
their practice answers and their started quizzes and tests; and over the whole range, how many different people."""
first = last - timedelta(days=days - 1)
per = {first + timedelta(days=i): {"users": set(), "answers": 0, "attempts": 0} for i in range(days)}
for r in rows:
if r["day"] in per:
d = per[r["day"]]
d["users"].add(r["user_id"])
d["answers"] += r["answers"]
d["attempts"] += r["attempts"]
everyone = set().union(*(d["users"] for d in per.values()))
return {"from": first.isoformat(), "to": last.isoformat(),
"days": [{"day": day.isoformat(), "users": len(d["users"]), "answers": d["answers"], "attempts": d["attempts"]}
for day, d in per.items()],
"users": len(everyone), "answers": sum(d["answers"] for d in per.values()),
"attempts": sum(d["attempts"] for d in per.values())}
@router.get("/activity")
def activity_route(request: Request, days: int = Query(7), user: dict = Depends(developer)) -> dict:
"""Who used the app on each of the last `days` days (7 or 30). Counts only, never who."""
if days not in ACTIVITY_DAYS:
raise HTTPException(422, "days must be 7 or 30")
last = clock.school_day()
store = request.app.state.store
return activity(getattr(store, "_store", store).activity_since(last - timedelta(days=days - 1)), last, days)
def setup(app) -> None:
app.state.version = version.read()
app.state.started_at = clock.now()
app.include_router(router)
|