Hera Zhao
d19183923c
Extracted from oil project — business logic removed, auth/db/deploy infrastructure generalized with APP_NAME placeholders. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
from fastapi import FastAPI, HTTPException, Depends
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
import hashlib
|
|
import secrets
|
|
import os
|
|
import threading
|
|
import time as _time
|
|
|
|
from backend.database import get_db, init_db, log_audit
|
|
from backend.auth import get_current_user, require_role, require_login
|
|
|
|
app = FastAPI(title="App API")
|
|
|
|
|
|
# ── Periodic WAL checkpoint ───────────────────────────
|
|
def _wal_checkpoint_loop():
|
|
while True:
|
|
_time.sleep(300)
|
|
try:
|
|
conn = get_db()
|
|
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
|
conn.close()
|
|
except:
|
|
pass
|
|
|
|
threading.Thread(target=_wal_checkpoint_loop, daemon=True).start()
|
|
|
|
|
|
# ── Models ────────────────────────────────────────────
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
class RegisterRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
display_name: str = ""
|
|
|
|
|
|
# ── Health & Version ──────────────────────────────────
|
|
APP_VERSION = "0.1.0"
|
|
|
|
@app.get("/api/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
@app.get("/api/version")
|
|
def version():
|
|
return {"version": APP_VERSION}
|
|
|
|
|
|
# ── Auth endpoints ────────────────────────────────────
|
|
def _hash_password(pw: str) -> str:
|
|
return hashlib.sha256(pw.encode()).hexdigest()
|
|
|
|
@app.get("/api/me")
|
|
def get_me(user=Depends(get_current_user)):
|
|
return {
|
|
"id": user.get("id"),
|
|
"username": user["username"],
|
|
"role": user["role"],
|
|
"display_name": user.get("display_name", ""),
|
|
}
|
|
|
|
@app.post("/api/login")
|
|
def login(body: LoginRequest):
|
|
conn = get_db()
|
|
user = conn.execute(
|
|
"SELECT id, username, token, password, role, display_name FROM users WHERE username = ?",
|
|
(body.username,),
|
|
).fetchone()
|
|
conn.close()
|
|
if not user:
|
|
raise HTTPException(401, "用户名或密码错误")
|
|
user = dict(user)
|
|
if user.get("password") and user["password"] != _hash_password(body.password):
|
|
raise HTTPException(401, "用户名或密码错误")
|
|
return {"token": user["token"]}
|
|
|
|
@app.post("/api/register", status_code=201)
|
|
def register(body: RegisterRequest):
|
|
conn = get_db()
|
|
existing = conn.execute("SELECT id FROM users WHERE username = ?", (body.username,)).fetchone()
|
|
if existing:
|
|
conn.close()
|
|
raise HTTPException(409, "用户名已存在")
|
|
token = secrets.token_hex(24)
|
|
pw_hash = _hash_password(body.password)
|
|
conn.execute(
|
|
"INSERT INTO users (username, password, token, role, display_name) VALUES (?, ?, ?, ?, ?)",
|
|
(body.username, pw_hash, token, "viewer", body.display_name or body.username),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return {"token": token}
|
|
|
|
|
|
# ── User management (admin) ──────────────────────────
|
|
@app.get("/api/users")
|
|
def list_users(user=Depends(require_role("admin"))):
|
|
conn = get_db()
|
|
rows = conn.execute("SELECT id, username, role, display_name, token, created_at FROM users ORDER BY id").fetchall()
|
|
conn.close()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
# ── Static files (frontend) ──────────────────────────
|
|
@app.on_event("startup")
|
|
def on_startup():
|
|
init_db()
|
|
frontend_dir = os.environ.get("FRONTEND_DIR", "frontend/dist")
|
|
if os.path.isdir(frontend_dir):
|
|
app.mount("/", StaticFiles(directory=frontend_dir, html=True), name="frontend")
|