Codex
769 lines
24 KiB
Python
769 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import html
|
|
import hmac
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import threading
|
|
from urllib.parse import parse_qs, quote
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException, Query, Request, status
|
|
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, Response
|
|
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,
|
|
)
|
|
|
|
|
|
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")
|
|
security = HTTPBasic(auto_error=False)
|
|
write_lock = threading.Lock()
|
|
|
|
|
|
class RegisterLinesPayload(BaseModel):
|
|
line: str | None = Field(default=None, description="单条原始登记文本")
|
|
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():
|
|
return RegisterLinesPayload()
|
|
|
|
content_type = request.headers.get("content-type", "").lower()
|
|
if "application/json" in content_type:
|
|
try:
|
|
data = json.loads(body)
|
|
if isinstance(data, str):
|
|
return RegisterLinesPayload(line=data)
|
|
if isinstance(data, list):
|
|
return RegisterLinesPayload(lines=data)
|
|
if isinstance(data, dict):
|
|
return RegisterLinesPayload(**data)
|
|
except (json.JSONDecodeError, ValidationError) as exc:
|
|
raise ValueError("登记 JSON 格式错误") from exc
|
|
raise ValueError("登记 JSON 必须是字符串、字符串数组，或包含 line/lines 的对象")
|
|
|
|
try:
|
|
text = body.decode("utf-8")
|
|
except UnicodeDecodeError as exc:
|
|
raise ValueError("登记内容必须使用 UTF-8 编码") from exc
|
|
return RegisterLinesPayload(lines=text.splitlines())
|
|
|
|
|
|
def configured_password(label: str, password: str) -> str:
|
|
if not password:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=f"服务未配置{label}访问密码",
|
|
)
|
|
return password
|
|
|
|
|
|
def session_token(password: str, purpose: bytes) -> str:
|
|
return hmac.new(
|
|
password.encode("utf-8"),
|
|
purpose,
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
|
|
def has_valid_session(request: Request, cookie_name: str, password: str, purpose: bytes) -> bool:
|
|
token = request.cookies.get(cookie_name, "")
|
|
return bool(password and token) and hmac.compare_digest(token, session_token(password, purpose))
|
|
|
|
|
|
def has_valid_basic_auth(credentials: HTTPBasicCredentials | None, password: str) -> bool:
|
|
if credentials is None:
|
|
return False
|
|
return bool(password) and hmac.compare_digest(credentials.password, password)
|
|
|
|
|
|
def is_records_authenticated(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = None,
|
|
) -> bool:
|
|
return has_valid_session(
|
|
request,
|
|
RECORDS_SESSION_COOKIE,
|
|
BASIC_AUTH_PASSWORD,
|
|
b"xsk-records-web-session-v1",
|
|
) or has_valid_basic_auth(credentials, BASIC_AUTH_PASSWORD)
|
|
|
|
|
|
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,
|
|
ADMIN_AUTH_PASSWORD,
|
|
b"xsk-accounts-web-session-v1",
|
|
) 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(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = Depends(security),
|
|
) -> str:
|
|
configured_password("课程记录", BASIC_AUTH_PASSWORD)
|
|
if not is_records_authenticated(request, credentials):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="请先登录",
|
|
)
|
|
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:
|
|
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_admin_authenticated(request, credentials):
|
|
return "authenticated"
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="请先登录",
|
|
)
|
|
|
|
|
|
def safe_next_path(value: str | None) -> str:
|
|
if not value or not value.startswith("/") or value.startswith("//"):
|
|
return "/"
|
|
return value
|
|
|
|
|
|
def render_login_page(
|
|
title: str,
|
|
action: str,
|
|
next_path: str = "/",
|
|
has_error: bool = False,
|
|
) -> str:
|
|
escaped_title = html.escape(title)
|
|
escaped_action = html.escape(action, quote=True)
|
|
escaped_next = html.escape(next_path, quote=True)
|
|
error_html = (
|
|
'<p class="error">密码不正确，请重新输入。</p>'
|
|
if has_error
|
|
else '<p class="hint">请输入访问密码。</p>'
|
|
)
|
|
return f"""<!doctype html>
|
|
<html lang="zh-CN">
|
|
<head>
|
|
<meta charset="utf-8" />
|
|
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
|
<title>{escaped_title}</title>
|
|
<style>
|
|
:root {{
|
|
color-scheme: light;
|
|
--bg: #f6f7f9;
|
|
--panel: #ffffff;
|
|
--text: #18202a;
|
|
--muted: #687487;
|
|
--line: #d9dee7;
|
|
--accent: #0f766e;
|
|
--accent-strong: #0b5d57;
|
|
--danger: #b42318;
|
|
font-family: Arial, "Songti SC", SimSun, sans-serif;
|
|
}}
|
|
* {{ box-sizing: border-box; }}
|
|
body {{
|
|
margin: 0;
|
|
min-height: 100vh;
|
|
display: grid;
|
|
place-items: center;
|
|
padding: 20px;
|
|
background: var(--bg);
|
|
color: var(--text);
|
|
}}
|
|
main {{
|
|
width: min(100%, 380px);
|
|
padding: 28px;
|
|
border: 1px solid var(--line);
|
|
border-radius: 8px;
|
|
background: var(--panel);
|
|
box-shadow: 0 12px 28px rgba(20, 31, 46, 0.08);
|
|
}}
|
|
h1 {{
|
|
margin: 0 0 10px;
|
|
font-size: 22px;
|
|
line-height: 1.25;
|
|
letter-spacing: 0;
|
|
}}
|
|
p {{
|
|
margin: 0 0 18px;
|
|
color: var(--muted);
|
|
font-size: 14px;
|
|
}}
|
|
.error {{ color: var(--danger); }}
|
|
label {{
|
|
display: block;
|
|
margin-bottom: 8px;
|
|
color: #344054;
|
|
font-size: 14px;
|
|
font-weight: 700;
|
|
}}
|
|
input {{
|
|
width: 100%;
|
|
height: 44px;
|
|
padding: 0 12px;
|
|
border: 1px solid var(--line);
|
|
border-radius: 6px;
|
|
color: var(--text);
|
|
font: inherit;
|
|
outline: none;
|
|
}}
|
|
.password-field {{
|
|
position: relative;
|
|
}}
|
|
.password-field input {{
|
|
padding-right: 48px;
|
|
}}
|
|
input:focus {{
|
|
border-color: var(--accent);
|
|
box-shadow: 0 0 0 3px rgba(15, 118, 110, 0.15);
|
|
}}
|
|
button {{
|
|
width: 100%;
|
|
min-height: 44px;
|
|
margin-top: 14px;
|
|
border: 0;
|
|
border-radius: 6px;
|
|
background: var(--accent);
|
|
color: #fff;
|
|
cursor: pointer;
|
|
font: inherit;
|
|
font-weight: 700;
|
|
}}
|
|
button:hover {{ background: var(--accent-strong); }}
|
|
.password-toggle {{
|
|
position: absolute;
|
|
top: 1px;
|
|
right: 1px;
|
|
width: 42px;
|
|
min-height: 42px;
|
|
margin-top: 0;
|
|
border: 0;
|
|
border-radius: 0 6px 6px 0;
|
|
background: transparent;
|
|
color: var(--muted);
|
|
}}
|
|
.password-toggle:hover {{
|
|
background: #eef2f6;
|
|
color: var(--accent-strong);
|
|
}}
|
|
.password-toggle svg {{
|
|
display: block;
|
|
width: 20px;
|
|
height: 20px;
|
|
margin: 0 auto;
|
|
fill: none;
|
|
stroke: currentColor;
|
|
stroke-width: 2;
|
|
stroke-linecap: round;
|
|
stroke-linejoin: round;
|
|
}}
|
|
.password-toggle .eye-open {{
|
|
display: none;
|
|
}}
|
|
.password-toggle.is-visible .eye-open {{
|
|
display: block;
|
|
}}
|
|
.password-toggle.is-visible .eye-closed {{
|
|
display: none;
|
|
}}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<main>
|
|
<h1>{escaped_title}</h1>
|
|
{error_html}
|
|
<form method="post" action="{escaped_action}" autocomplete="off">
|
|
<input type="hidden" name="next" value="{escaped_next}" />
|
|
<label for="password">访问密码</label>
|
|
<div class="password-field">
|
|
<input id="password" name="password" type="password" autocomplete="current-password" autofocus required />
|
|
<button id="togglePassword" class="password-toggle" type="button" aria-label="显示密码" title="显示密码">
|
|
<svg class="eye-open" viewBox="0 0 24 24" aria-hidden="true">
|
|
<path d="M2 12s3.5-7 10-7 10 7 10 7-3.5 7-10 7-10-7-10-7Z" />
|
|
<circle cx="12" cy="12" r="3" />
|
|
</svg>
|
|
<svg class="eye-closed" viewBox="0 0 24 24" aria-hidden="true">
|
|
<path d="M17.94 17.94A10.8 10.8 0 0 1 12 19C5.5 19 2 12 2 12a18.4 18.4 0 0 1 5.06-5.94" />
|
|
<path d="M9.9 4.24A10.9 10.9 0 0 1 12 4.05C18.5 4.05 22 12 22 12a18.5 18.5 0 0 1-2.16 3.19" />
|
|
<path d="M14.12 14.12A3 3 0 0 1 9.88 9.88" />
|
|
<path d="M3 3l18 18" />
|
|
</svg>
|
|
</button>
|
|
</div>
|
|
<button type="submit">进入</button>
|
|
</form>
|
|
</main>
|
|
<script>
|
|
const passwordInput = document.querySelector("#password");
|
|
const togglePassword = document.querySelector("#togglePassword");
|
|
togglePassword.addEventListener("click", () => {{
|
|
const isVisible = passwordInput.type === "text";
|
|
passwordInput.type = isVisible ? "password" : "text";
|
|
togglePassword.classList.toggle("is-visible", !isVisible);
|
|
const label = isVisible ? "显示密码" : "隐藏密码";
|
|
togglePassword.setAttribute("aria-label", label);
|
|
togglePassword.setAttribute("title", label);
|
|
passwordInput.focus();
|
|
}});
|
|
</script>
|
|
</body>
|
|
</html>"""
|
|
|
|
|
|
def load_records():
|
|
if not CLASSNOTES_PATH.exists():
|
|
raise HTTPException(status_code=503, detail=f"课程记录文件不存在: {CLASSNOTES_PATH}")
|
|
return read_classnotes(CLASSNOTES_PATH)
|
|
|
|
|
|
def load_accounts():
|
|
if not ACCOUNTS_PATH.exists():
|
|
raise HTTPException(status_code=503, detail=f"课时账户文件不存在: {ACCOUNTS_PATH}")
|
|
return read_accounts(ACCOUNTS_PATH)
|
|
|
|
|
|
def file_meta(path: Path) -> dict:
|
|
if not path.exists():
|
|
return {"exists": False, "path": str(path)}
|
|
stat = path.stat()
|
|
return {
|
|
"exists": True,
|
|
"path": str(path),
|
|
"size": stat.st_size,
|
|
"mtime": stat.st_mtime,
|
|
}
|
|
|
|
|
|
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)})
|
|
|
|
|
|
@app.get("/")
|
|
def index(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = Depends(security),
|
|
):
|
|
if not is_records_authenticated(request, credentials):
|
|
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
|
return FileResponse(STATIC_DIR / "index.html")
|
|
|
|
|
|
@app.head("/")
|
|
def index_head(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = Depends(security),
|
|
):
|
|
if not is_records_authenticated(request, credentials):
|
|
return RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
|
return Response(status_code=status.HTTP_200_OK)
|
|
|
|
|
|
@app.get("/login")
|
|
def login_page(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = Depends(security),
|
|
):
|
|
next_path = safe_next_path(request.query_params.get("next"))
|
|
has_error = request.query_params.get("error") == "1"
|
|
if is_records_authenticated(request, credentials):
|
|
return RedirectResponse(url=next_path, status_code=status.HTTP_303_SEE_OTHER)
|
|
return HTMLResponse(
|
|
render_login_page(
|
|
title="新时空教务管家",
|
|
action="/login",
|
|
next_path=next_path,
|
|
has_error=has_error,
|
|
)
|
|
)
|
|
|
|
|
|
@app.post("/login")
|
|
async def 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", ["/"])[0])
|
|
records_password = configured_password("课程记录", BASIC_AUTH_PASSWORD)
|
|
if hmac.compare_digest(password, records_password):
|
|
response = RedirectResponse(url=next_path, status_code=status.HTTP_303_SEE_OTHER)
|
|
response.set_cookie(
|
|
RECORDS_SESSION_COOKIE,
|
|
session_token(records_password, b"xsk-records-web-session-v1"),
|
|
max_age=SESSION_MAX_AGE,
|
|
httponly=True,
|
|
samesite="lax",
|
|
)
|
|
return response
|
|
error_url = f"/login?error=1&next={quote(next_path)}"
|
|
return RedirectResponse(url=error_url, status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
@app.get("/logout")
|
|
def logout():
|
|
response = RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
|
response.delete_cookie(RECORDS_SESSION_COOKIE)
|
|
return response
|
|
|
|
|
|
@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_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("/admin/login")
|
|
def admin_login_page(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = Depends(security),
|
|
):
|
|
next_path = safe_next_path(request.query_params.get("next") or "/admin")
|
|
has_error = request.query_params.get("error") == "1"
|
|
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="/admin/login",
|
|
next_path=next_path,
|
|
has_error=has_error,
|
|
)
|
|
)
|
|
|
|
|
|
@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", ["/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(
|
|
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"/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="/admin/login", status_code=status.HTTP_303_SEE_OTHER)
|
|
response.delete_cookie(ACCOUNTS_SESSION_COOKIE)
|
|
response.delete_cookie(ADMIN_SESSION_COOKIE)
|
|
return response
|
|
|
|
|
|
@app.get("/static/{asset_path:path}")
|
|
def static_asset(asset_path: str, _user: str = Depends(verify_any_auth)):
|
|
target = (STATIC_DIR / asset_path).resolve()
|
|
static_root = STATIC_DIR.resolve()
|
|
if not target.is_file() or static_root not in target.parents:
|
|
raise HTTPException(status_code=404, detail="静态资源不存在")
|
|
return FileResponse(target)
|
|
|
|
|
|
@app.post("/api/register/class-records")
|
|
async def register_class_records(request: Request, _user: str = Depends(verify_admin_auth)):
|
|
try:
|
|
payload = await read_register_payload(request)
|
|
with write_lock:
|
|
result = register_class_record_lines(
|
|
CLASSNOTES_PATH,
|
|
ACCOUNTS_PATH,
|
|
lines=payload.lines,
|
|
line=payload.line,
|
|
)
|
|
except DuplicateRecordError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
return {"ok": True, **result}
|
|
|
|
|
|
@app.post("/api/register/payments")
|
|
async def register_payments(request: Request, _user: str = Depends(verify_admin_auth)):
|
|
try:
|
|
payload = await read_register_payload(request)
|
|
with write_lock:
|
|
result = register_payment_lines(ACCOUNTS_PATH, lines=payload.lines, line=payload.line)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
return {"ok": True, **result}
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health(_user: str = Depends(verify_records_auth)):
|
|
records = load_records()
|
|
accounts = load_accounts()
|
|
return {
|
|
"ok": True,
|
|
"classnotes": file_meta(CLASSNOTES_PATH),
|
|
"accounts": file_meta(ACCOUNTS_PATH),
|
|
"records_count": len(records),
|
|
"accounts_count": len(accounts),
|
|
"account_summary": account_summary(accounts),
|
|
}
|
|
|
|
|
|
@app.get("/api/records")
|
|
def records(
|
|
q: str = Query(..., min_length=1, description="自然语言查询，例如：王鑫鹏5月数学课"),
|
|
limit: int = Query(200, ge=1, le=1000),
|
|
_user: str = Depends(verify_records_auth),
|
|
):
|
|
return query_records(load_records(), q, limit=limit)
|
|
|
|
|
|
@app.get("/api/student-account/{student}")
|
|
def record_student_account(student: str, _user: str = Depends(verify_records_auth)):
|
|
for account in load_accounts():
|
|
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/account-health")
|
|
def account_health(_user: str = Depends(verify_accounts_auth)):
|
|
accounts = load_accounts()
|
|
return {
|
|
"ok": True,
|
|
"accounts": file_meta(ACCOUNTS_PATH),
|
|
"accounts_count": len(accounts),
|
|
"account_summary": account_summary(accounts),
|
|
}
|
|
|
|
|
|
@app.get("/api/accounts")
|
|
def accounts(
|
|
q: str = Query("", description="学生姓名或学生ID"),
|
|
status_filter: str = Query("", alias="status", description="账户状态"),
|
|
_user: str = Depends(verify_accounts_auth),
|
|
):
|
|
all_accounts = load_accounts()
|
|
rows = filter_accounts(all_accounts, keyword=q, status=status_filter)
|
|
return {
|
|
"summary": account_summary(all_accounts),
|
|
"count": len(rows),
|
|
"accounts": [account_to_dict(account) for account in rows],
|
|
}
|
|
|
|
|
|
@app.get("/api/accounts/{student}")
|
|
def account_detail(student: str, _user: str = Depends(verify_accounts_auth)):
|
|
for account in load_accounts():
|
|
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}
|