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 (
DuplicateRecordError,
account_summary,
account_to_dict,
filter_accounts,
query_records,
read_accounts,
read_classnotes,
register_class_record_lines,
register_payment_lines,
)
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"))
BASIC_AUTH_PASSWORD = os.getenv("BASIC_AUTH_PASSWORD", "")
ACCOUNTS_AUTH_PASSWORD = os.getenv("ACCOUNTS_AUTH_PASSWORD") or os.getenv("ACCOUNT_AUTH_PASSWORD", "")
RECORDS_SESSION_COOKIE = "xsk_records_session"
ACCOUNTS_SESSION_COOKIE = "xsk_accounts_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="多条原始登记文本")
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_accounts_authenticated(
request: Request,
credentials: HTTPBasicCredentials | None = None,
) -> bool:
return has_valid_session(
request,
ACCOUNTS_SESSION_COOKIE,
ACCOUNTS_AUTH_PASSWORD,
b"xsk-accounts-web-session-v1",
) or has_valid_basic_auth(credentials, ACCOUNTS_AUTH_PASSWORD)
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_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"
def verify_any_auth(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
) -> str:
if is_records_authenticated(request, credentials) or is_accounts_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 = (
'密码不正确，请重新输入。
'
if has_error
else '请输入访问密码。
'
)
return f"""
{escaped_title}
{escaped_title}
{error_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,
}
@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(
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")
@app.get("/accounts/login")
def accounts_login_page(
request: Request,
credentials: HTTPBasicCredentials | None = Depends(security),
):
next_path = safe_next_path(request.query_params.get("next") or "/accounts")
has_error = request.query_params.get("error") == "1"
if is_accounts_authenticated(request, credentials):
return RedirectResponse(url=next_path, status_code=status.HTTP_303_SEE_OTHER)
return HTMLResponse(
render_login_page(
title="课时账户查询",
action="/accounts/login",
next_path=next_path,
has_error=has_error,
)
)
@app.post("/accounts/login")
async def accounts_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):
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"),
max_age=SESSION_MAX_AGE,
httponly=True,
samesite="lax",
)
return response
error_url = f"/accounts/login?error=1&next={quote(next_path)}"
return RedirectResponse(url=error_url, status_code=status.HTTP_303_SEE_OTHER)
@app.get("/accounts/logout")
def accounts_logout():
response = RedirectResponse(url="/accounts/login", status_code=status.HTTP_303_SEE_OTHER)
response.delete_cookie(ACCOUNTS_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_accounts_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_accounts_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}")