wolfydw/xsk-education-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add admin console and review workflow

Browse Source
This commit is contained in:
Codex
2026-06-14 21:13:23 +08:00
parent 2591549734
commit 6417d2baa9
13 changed files with 1184 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/main.py
+191 -31
View File
@@ -15,15 +15,24 @@ from fastapi.security import HTTPBasic, HTTPBasicCredentials
from pydantic import BaseModel, Field, ValidationError
from .data import (
ACCOUNT_STATUSES,
Account,
DuplicateRecordError,
Payment,
approve_correction_task,
account_summary,
account_to_dict,
create_account,
filter_accounts,
list_admin_tasks,
query_records,
read_accounts,
read_classnotes,
register_class_record_lines,
register_payment_lines,
reject_admin_task,
submit_correction_tasks,
update_account,
)
@@ -31,10 +40,13 @@ APP_DIR = Path(__file__).resolve().parent
STATIC_DIR = APP_DIR / "static"
CLASSNOTES_PATH = Path(os.getenv("CLASSNOTES_PATH", "/data/classnotes.txt"))
ACCOUNTS_PATH = Path(os.getenv("ACCOUNTS_PATH", "/data/学生课时账户.md"))
ADMIN_TASKS_PATH = Path(os.getenv("ADMIN_TASKS_PATH", "/data/admin_tasks.json"))
BASIC_AUTH_PASSWORD = os.getenv("BASIC_AUTH_PASSWORD", "")
ACCOUNTS_AUTH_PASSWORD = os.getenv("ACCOUNTS_AUTH_PASSWORD") or os.getenv("ACCOUNT_AUTH_PASSWORD", "")
ADMIN_AUTH_PASSWORD = os.getenv("ADMIN_AUTH_PASSWORD") or ACCOUNTS_AUTH_PASSWORD
RECORDS_SESSION_COOKIE = "xsk_records_session"
ACCOUNTS_SESSION_COOKIE = "xsk_accounts_session"
ADMIN_SESSION_COOKIE = "xsk_admin_session"
SESSION_MAX_AGE = 60 * 60 * 24 * 30
app = FastAPI(title="新时空课程记录查询", version="1.0.0")
@@ -47,6 +59,29 @@ class RegisterLinesPayload(BaseModel):
lines: list[str] | None = Field(default=None, description="多条原始登记文本")
class PaymentPayload(BaseModel):
date: str
hours: float
class AccountPayload(BaseModel):
student_id: str = ""
student: str
payments: list[PaymentPayload] = Field(default_factory=list)
remaining: float = 0
account_status: str = "正常"
note: str = ""
class CorrectionItemPayload(BaseModel):
original_line: str
corrected_line: str
class CorrectionSubmitPayload(BaseModel):
items: list[CorrectionItemPayload]
async def read_register_payload(request: Request) -> RegisterLinesPayload:
body = await request.body()
if not body.strip():
@@ -113,16 +148,28 @@ def is_records_authenticated(
) or has_valid_basic_auth(credentials, BASIC_AUTH_PASSWORD)
def is_accounts_authenticated(
def is_admin_authenticated(
request: Request,
credentials: HTTPBasicCredentials | None = None,
) -> bool:
return has_valid_session(
request,
ADMIN_SESSION_COOKIE,
ADMIN_AUTH_PASSWORD,
b"xsk-admin-web-session-v1",
) or has_valid_session(
request,
ACCOUNTS_SESSION_COOKIE,
ACCOUNTS_AUTH_PASSWORD,
ADMIN_AUTH_PASSWORD,
b"xsk-accounts-web-session-v1",
) or has_valid_basic_auth(credentials, ACCOUNTS_AUTH_PASSWORD)
) or has_valid_basic_auth(credentials, ADMIN_AUTH_PASSWORD)
def is_accounts_authenticated(
request: Request,
credentials: HTTPBasicCredentials | None = None,
) -> bool:
return is_admin_authenticated(request, credentials)
def verify_records_auth(
@@ -138,24 +185,31 @@ def verify_records_auth(
return "records"
def verify_admin_auth(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
) -> str:
configured_password("管理后台", ADMIN_AUTH_PASSWORD)
if not is_admin_authenticated(request, credentials):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="请先登录管理后台",
)
return "admin"
def verify_accounts_auth(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
) -> str:
configured_password("课时账户", ACCOUNTS_AUTH_PASSWORD)
if not is_accounts_authenticated(request, credentials):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="请先登录课时账户",
)
return "accounts"
return verify_admin_auth(request, credentials)
def verify_any_auth(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
) -> str:
if is_records_authenticated(request, credentials) or is_accounts_authenticated(request, credentials):
if is_records_authenticated(request, credentials) or is_admin_authenticated(request, credentials):
return "authenticated"
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
@@ -376,6 +430,17 @@ def file_meta(path: Path) -> dict:
}
def payload_to_account(payload: AccountPayload, student_id: str | None = None) -> Account:
return Account(
student_id=student_id if student_id is not None else payload.student_id,
student=payload.student,
payments=[Payment(date=item.date, hours=item.hours) for item in payload.payments],
remaining=payload.remaining,
account_status=payload.account_status,
note=payload.note,
)
@app.exception_handler(ValueError)
async def value_error_handler(_request: Request, exc: ValueError):
return JSONResponse(status_code=500, content={"detail": str(exc)})
@@ -450,58 +515,84 @@ def logout():
@app.get("/accounts")
def accounts_index(
):
return RedirectResponse(url="/admin", status_code=status.HTTP_303_SEE_OTHER)
@app.get("/admin")
def admin_index(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
):
if not is_accounts_authenticated(request, credentials):
return RedirectResponse(url="/accounts/login?next=/accounts", status_code=status.HTTP_303_SEE_OTHER)
return FileResponse(STATIC_DIR / "accounts.html")
if not is_admin_authenticated(request, credentials):
return RedirectResponse(url="/admin/login?next=/admin", status_code=status.HTTP_303_SEE_OTHER)
return FileResponse(STATIC_DIR / "admin.html")
@app.get("/accounts/login")
def accounts_login_page(
@app.get("/admin/login")
def admin_login_page(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
):
next_path = safe_next_path(request.query_params.get("next") or "/accounts")
next_path = safe_next_path(request.query_params.get("next") or "/admin")
has_error = request.query_params.get("error") == "1"
if is_accounts_authenticated(request, credentials):
if is_admin_authenticated(request, credentials):
return RedirectResponse(url=next_path, status_code=status.HTTP_303_SEE_OTHER)
return HTMLResponse(
render_login_page(
title="课时账户查询",
action="/accounts/login",
title="管理后台",
action="/admin/login",
next_path=next_path,
has_error=has_error,
)
)
@app.post("/accounts/login")
async def accounts_login_submit(request: Request):
@app.post("/admin/login")
async def admin_login_submit(request: Request):
body = (await request.body()).decode("utf-8")
form = parse_qs(body, keep_blank_values=True)
password = form.get("password", [""])[0]
next_path = safe_next_path(form.get("next", ["/accounts"])[0])
accounts_password = configured_password("课时账户", ACCOUNTS_AUTH_PASSWORD)
if hmac.compare_digest(password, accounts_password):
next_path = safe_next_path(form.get("next", ["/admin"])[0])
admin_password = configured_password("管理后台", ADMIN_AUTH_PASSWORD)
if hmac.compare_digest(password, admin_password):
response = RedirectResponse(url=next_path, status_code=status.HTTP_303_SEE_OTHER)
response.set_cookie(
ACCOUNTS_SESSION_COOKIE,
session_token(accounts_password, b"xsk-accounts-web-session-v1"),
ADMIN_SESSION_COOKIE,
session_token(admin_password, b"xsk-admin-web-session-v1"),
max_age=SESSION_MAX_AGE,
httponly=True,
samesite="lax",
)
return response
error_url = f"/accounts/login?error=1&next={quote(next_path)}"
error_url = f"/admin/login?error=1&next={quote(next_path)}"
return RedirectResponse(url=error_url, status_code=status.HTTP_303_SEE_OTHER)
@app.get("/admin/logout")
def admin_logout():
response = RedirectResponse(url="/admin/login", status_code=status.HTTP_303_SEE_OTHER)
response.delete_cookie(ADMIN_SESSION_COOKIE)
response.delete_cookie(ACCOUNTS_SESSION_COOKIE)
return response
@app.get("/accounts/login")
def accounts_login_page(
):
return RedirectResponse(url="/admin/login?next=/admin", status_code=status.HTTP_303_SEE_OTHER)
@app.post("/accounts/login")
async def accounts_login_submit(request: Request):
return await admin_login_submit(request)
@app.get("/accounts/logout")
def accounts_logout():
response = RedirectResponse(url="/accounts/login", status_code=status.HTTP_303_SEE_OTHER)
response = RedirectResponse(url="/admin/login", status_code=status.HTTP_303_SEE_OTHER)
response.delete_cookie(ACCOUNTS_SESSION_COOKIE)
response.delete_cookie(ADMIN_SESSION_COOKIE)
return response
@@ -515,7 +606,7 @@ def static_asset(asset_path: str, _user: str = Depends(verify_any_auth)):
@app.post("/api/register/class-records")
async def register_class_records(request: Request, _user: str = Depends(verify_accounts_auth)):
async def register_class_records(request: Request, _user: str = Depends(verify_admin_auth)):
try:
payload = await read_register_payload(request)
with write_lock:
@@ -533,7 +624,7 @@ async def register_class_records(request: Request, _user: str = Depends(verify_a
@app.post("/api/register/payments")
async def register_payments(request: Request, _user: str = Depends(verify_accounts_auth)):
async def register_payments(request: Request, _user: str = Depends(verify_admin_auth)):
try:
payload = await read_register_payload(request)
with write_lock:
@@ -606,3 +697,72 @@ def account_detail(student: str, _user: str = Depends(verify_accounts_auth)):
if account.student == student or account.student_id == student:
return account_to_dict(account)
raise HTTPException(status_code=404, detail=f"未找到学生账户: {student}")
@app.get("/api/admin/statuses")
def admin_statuses(_user: str = Depends(verify_admin_auth)):
return {"account_statuses": sorted(ACCOUNT_STATUSES)}
@app.post("/api/admin/accounts")
def admin_create_account(payload: AccountPayload, _user: str = Depends(verify_admin_auth)):
try:
with write_lock:
result = create_account(ACCOUNTS_PATH, payload_to_account(payload))
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"ok": True, **result}
@app.put("/api/admin/accounts/{student_id}")
def admin_update_account(student_id: str, payload: AccountPayload, _user: str = Depends(verify_admin_auth)):
try:
with write_lock:
result = update_account(ACCOUNTS_PATH, student_id, payload_to_account(payload))
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"ok": True, **result}
@app.post("/api/corrections")
def submit_corrections(payload: CorrectionSubmitPayload, _user: str = Depends(verify_records_auth)):
try:
result = submit_correction_tasks(
ADMIN_TASKS_PATH,
[item.dict() for item in payload.items],
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"ok": True, **result}
@app.get("/api/admin/tasks")
def admin_tasks(
status_filter: str = Query("", alias="status"),
task_type: str = Query("", alias="type"),
_user: str = Depends(verify_admin_auth),
):
try:
return list_admin_tasks(ADMIN_TASKS_PATH, status_filter=status_filter, task_type=task_type)
except ValueError as exc:
raise HTTPException(status_code=500, detail=str(exc)) from exc
@app.post("/api/admin/tasks/{task_id}/approve")
def admin_approve_task(task_id: int, _user: str = Depends(verify_admin_auth)):
try:
with write_lock:
result = approve_correction_task(ADMIN_TASKS_PATH, CLASSNOTES_PATH, task_id)
except ValueError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
return {"ok": True, **result}
@app.post("/api/admin/tasks/{task_id}/reject")
def admin_reject_task(task_id: int, _user: str = Depends(verify_admin_auth)):
try:
with write_lock:
task = reject_admin_task(ADMIN_TASKS_PATH, task_id)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"ok": True, "task": task}