feat: add user settings page (change username/password) and remove public registration

This commit is contained in:
2026-07-05 22:01:24 +08:00
parent 5607485c6a
commit ae2ed2d961
12 changed files with 127 additions and 348 deletions

View File

@@ -8,7 +8,7 @@ from sqlalchemy.orm import Session as DbSession
from app.config import SECRET_KEY, SESSION_COOKIE_NAME from app.config import SECRET_KEY, SESSION_COOKIE_NAME
from app.database import get_db from app.database import get_db
from app.models import ApiKey, User, UserSession from app.models import User, UserSession
# ---------- Password ---------- # ---------- Password ----------
@@ -81,22 +81,3 @@ def redirect_if_not_logged_in(request: Request, db: DbSession) -> User | None:
if uid is None: if uid is None:
return None return None
return db.query(User).filter(User.id == uid).first() return db.query(User).filter(User.id == uid).first()
# ---------- API Dependencies ----------
def get_current_user_api(
request: Request,
db: DbSession = Depends(get_db),
) -> User:
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
token = auth_header[7:]
api_key = db.query(ApiKey).filter(ApiKey.key == token).first()
if api_key is None:
raise HTTPException(status_code=401, detail="Invalid API key")
user = db.query(User).filter(User.id == api_key.user_id).first()
if user is None:
raise HTTPException(status_code=401, detail="User not found")
return user

View File

@@ -5,7 +5,7 @@ from fastapi.staticfiles import StaticFiles
from app.config import UPLOAD_DIR from app.config import UPLOAD_DIR
from app.database import init_db from app.database import init_db
from app.routers import api, web from app.routers import web
app = FastAPI(title="Postcard Manager") app = FastAPI(title="Postcard Manager")
@@ -14,7 +14,6 @@ app.mount("/static", StaticFiles(directory=str(Path(__file__).resolve().parent /
app.mount("/uploads", StaticFiles(directory=str(UPLOAD_DIR)), name="uploads") app.mount("/uploads", StaticFiles(directory=str(UPLOAD_DIR)), name="uploads")
app.include_router(web.router) app.include_router(web.router)
app.include_router(api.router)
init_db() init_db()

View File

@@ -1,7 +1,6 @@
import uuid
from datetime import datetime, timezone from datetime import datetime, timezone
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, event from sqlalchemy import Column, DateTime, ForeignKey, Integer, String
from sqlalchemy.orm import relationship, validates from sqlalchemy.orm import relationship, validates
from app.database import Base from app.database import Base
@@ -19,7 +18,6 @@ class User(Base):
password_hash = Column(String(128), nullable=False) password_hash = Column(String(128), nullable=False)
created_at = Column(DateTime, default=_utcnow) created_at = Column(DateTime, default=_utcnow)
api_keys = relationship("ApiKey", back_populates="user", cascade="all, delete-orphan")
profiles = relationship("Profile", back_populates="user", cascade="all, delete-orphan") profiles = relationship("Profile", back_populates="user", cascade="all, delete-orphan")
sessions = relationship("UserSession", back_populates="user", cascade="all, delete-orphan") sessions = relationship("UserSession", back_populates="user", cascade="all, delete-orphan")
@@ -35,21 +33,6 @@ class UserSession(Base):
user = relationship("User", back_populates="sessions") user = relationship("User", back_populates="sessions")
class ApiKey(Base):
__tablename__ = "api_keys"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
key = Column(String(64), unique=True, nullable=False, index=True)
created_at = Column(DateTime, default=_utcnow)
user = relationship("User", back_populates="api_keys")
@staticmethod
def generate_key() -> str:
return uuid.uuid4().hex
class Profile(Base): class Profile(Base):
__tablename__ = "profiles" __tablename__ = "profiles"
@@ -68,7 +51,7 @@ class Postcard(Base):
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
profile_id = Column(Integer, ForeignKey("profiles.id", ondelete="CASCADE"), nullable=False) profile_id = Column(Integer, ForeignKey("profiles.id", ondelete="CASCADE"), nullable=False)
card_number = Column(String(64), nullable=False) card_number = Column(String(64), nullable=False)
status = Column(String(16), nullable=False) # sent / delivered / received status = Column(String(16), nullable=False)
# sent & delivered # sent & delivered
recipient_name = Column(String(64)) recipient_name = Column(String(64))
country = Column(String(64)) country = Column(String(64))

View File

@@ -1,209 +0,0 @@
from pathlib import Path
from uuid import uuid4
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
from sqlalchemy.orm import Session
from app.auth import get_current_user_api, hash_password, verify_password
from app.config import UPLOAD_DIR
from app.database import get_db
from app.models import ApiKey, Postcard, Profile, User
from app.schemas import (
ApiKeyOut,
LoginRequest,
PostcardCreate,
PostcardOut,
PostcardUpdate,
ProfileCreate,
ProfileOut,
RegisterRequest,
)
router = APIRouter(prefix="/api", tags=["api"])
# ---------- Auth ----------
@router.post("/register", response_model=dict)
def api_register(body: RegisterRequest, db: Session = Depends(get_db)):
if db.query(User).filter(User.username == body.username).first():
raise HTTPException(400, "Username already exists")
user = User(username=body.username, password_hash=hash_password(body.password))
db.add(user)
db.commit()
return {"message": "ok"}
@router.post("/login", response_model=dict)
def api_login(body: LoginRequest, db: Session = Depends(get_db)):
user = db.query(User).filter(User.username == body.username).first()
if not user or not verify_password(body.password, user.password_hash):
raise HTTPException(401, "Invalid credentials")
api_key = ApiKey(user_id=user.id, key=ApiKey.generate_key())
db.add(api_key)
db.commit()
return {"api_key": api_key.key}
# ---------- Profile ----------
def _get_profile(profile_id: int, user: User, db: Session) -> Profile:
p = db.query(Profile).filter(Profile.id == profile_id, Profile.user_id == user.id).first()
if not p:
raise HTTPException(404, "Profile not found")
return p
@router.get("/profiles", response_model=list[ProfileOut])
def api_list_profiles(user: User = Depends(get_current_user_api), db: Session = Depends(get_db)):
return db.query(Profile).filter(Profile.user_id == user.id).all()
@router.post("/profiles", response_model=ProfileOut, status_code=201)
def api_create_profile(body: ProfileCreate, user: User = Depends(get_current_user_api), db: Session = Depends(get_db)):
p = Profile(user_id=user.id, nickname=body.nickname)
db.add(p)
db.commit()
db.refresh(p)
return p
@router.put("/profiles/{profile_id}", response_model=ProfileOut)
def api_update_profile(profile_id: int, body: ProfileCreate, user: User = Depends(get_current_user_api), db: Session = Depends(get_db)):
p = _get_profile(profile_id, user, db)
p.nickname = body.nickname
db.commit()
db.refresh(p)
return p
@router.delete("/profiles/{profile_id}")
def api_delete_profile(profile_id: int, user: User = Depends(get_current_user_api), db: Session = Depends(get_db)):
p = _get_profile(profile_id, user, db)
db.delete(p)
db.commit()
return {"message": "ok"}
# ---------- Postcard ----------
def _get_postcard(postcard_id: int, user: User, db: Session) -> Postcard:
pc = (
db.query(Postcard)
.join(Profile)
.filter(Postcard.id == postcard_id, Profile.user_id == user.id)
.first()
)
if not pc:
raise HTTPException(404, "Postcard not found")
return pc
@router.get("/profiles/{profile_id}/postcards", response_model=list[PostcardOut])
def api_list_postcards(
profile_id: int,
status: str | None = None,
country: str | None = None,
user: User = Depends(get_current_user_api),
db: Session = Depends(get_db),
):
_get_profile(profile_id, user, db)
q = db.query(Postcard).filter(Postcard.profile_id == profile_id)
if status:
q = q.filter(Postcard.status == status)
if country:
q = q.filter(Postcard.country.ilike(f"%{country}%"))
return q.order_by(Postcard.created_at.desc()).all()
@router.post("/profiles/{profile_id}/postcards", response_model=PostcardOut, status_code=201)
def api_create_postcard(
profile_id: int,
body: PostcardCreate,
user: User = Depends(get_current_user_api),
db: Session = Depends(get_db),
):
_get_profile(profile_id, user, db)
pc = Postcard(profile_id=profile_id, **body.model_dump())
db.add(pc)
db.commit()
db.refresh(pc)
return pc
@router.get("/postcards/{postcard_id}", response_model=PostcardOut)
def api_get_postcard(postcard_id: int, user: User = Depends(get_current_user_api), db: Session = Depends(get_db)):
return _get_postcard(postcard_id, user, db)
@router.put("/postcards/{postcard_id}", response_model=PostcardOut)
def api_update_postcard(
postcard_id: int,
body: PostcardUpdate,
user: User = Depends(get_current_user_api),
db: Session = Depends(get_db),
):
pc = _get_postcard(postcard_id, user, db)
for k, v in body.model_dump(exclude_unset=True).items():
setattr(pc, k, v)
db.commit()
db.refresh(pc)
return pc
@router.delete("/postcards/{postcard_id}")
def api_delete_postcard(postcard_id: int, user: User = Depends(get_current_user_api), db: Session = Depends(get_db)):
pc = _get_postcard(postcard_id, user, db)
db.delete(pc)
db.commit()
return {"message": "ok"}
# ---------- Image Upload ----------
@router.post("/postcards/{postcard_id}/image/{side}", response_model=PostcardOut)
async def api_upload_image(
postcard_id: int,
side: str,
file: UploadFile = File(...),
user: User = Depends(get_current_user_api),
db: Session = Depends(get_db),
):
if side not in ("front", "back"):
raise HTTPException(400, "side must be 'front' or 'back'")
pc = _get_postcard(postcard_id, user, db)
ext = Path(file.filename or "upload.jpg").suffix or ".jpg"
filename = f"{uuid4().hex}{ext}"
dest = UPLOAD_DIR / filename
content = await file.read()
dest.write_bytes(content)
setattr(pc, f"image_{side}", f"/uploads/{filename}")
db.commit()
db.refresh(pc)
return pc
# ---------- API Key ----------
@router.get("/api-keys", response_model=list[ApiKeyOut])
def api_list_keys(user: User = Depends(get_current_user_api), db: Session = Depends(get_db)):
return db.query(ApiKey).filter(ApiKey.user_id == user.id).all()
@router.post("/api-keys", response_model=ApiKeyOut, status_code=201)
def api_create_key(user: User = Depends(get_current_user_api), db: Session = Depends(get_db)):
ak = ApiKey(user_id=user.id, key=ApiKey.generate_key())
db.add(ak)
db.commit()
db.refresh(ak)
return ak
@router.delete("/api-keys/{key_id}")
def api_delete_key(key_id: int, user: User = Depends(get_current_user_api), db: Session = Depends(get_db)):
ak = db.query(ApiKey).filter(ApiKey.id == key_id, ApiKey.user_id == user.id).first()
if not ak:
raise HTTPException(404, "API key not found")
db.delete(ak)
db.commit()
return {"message": "ok"}

View File

@@ -17,7 +17,7 @@ from app.auth import (
) )
from app.config import SECRET_KEY, SESSION_COOKIE_NAME, UPLOAD_DIR from app.config import SECRET_KEY, SESSION_COOKIE_NAME, UPLOAD_DIR
from app.database import get_db from app.database import get_db
from app.models import ApiKey, Postcard, Profile, User from app.models import Postcard, Profile, User
router = APIRouter(tags=["web"]) router = APIRouter(tags=["web"])
templates = Jinja2Templates(directory=str(Path(__file__).resolve().parent.parent / "templates")) templates = Jinja2Templates(directory=str(Path(__file__).resolve().parent.parent / "templates"))
@@ -71,35 +71,69 @@ def login_submit(
return resp return resp
@router.get("/register", response_class=HTMLResponse) # ---------- Settings ----------
def register_page(request: Request):
return templates.TemplateResponse("register.html", {"request": request})
@router.get("/settings", response_class=HTMLResponse)
@router.post("/register") def settings_page(
def register_submit(
request: Request, request: Request,
username: str = Form(...), user: User = Depends(get_current_user_web),
):
return templates.TemplateResponse("settings.html", {"request": request, "user": user})
@router.post("/settings/change-username")
def change_username(
request: Request,
new_username: str = Form(...),
password: str = Form(...), password: str = Form(...),
password2: str = Form(...), user: User = Depends(get_current_user_web),
db: Session = Depends(get_db), db: Session = Depends(get_db),
): ):
if password != password2: if not verify_password(password, user.password_hash):
return templates.TemplateResponse( return templates.TemplateResponse(
"register.html", {"request": request, "error": "两次密码不一致"}, status_code=400 "settings.html", {"request": request, "user": user, "error": "密码错误"}, status_code=400
) )
if db.query(User).filter(User.username == username).first(): if db.query(User).filter(User.username == new_username, User.id != user.id).first():
return templates.TemplateResponse( return templates.TemplateResponse(
"register.html", {"request": request, "error": "用户名已存在"}, status_code=400 "settings.html", {"request": request, "user": user, "error": "用户名已存在"}, status_code=400
) )
user = User(username=username, password_hash=hash_password(password)) user.username = new_username
db.add(user)
db.commit() db.commit()
resp = _redirect("/dashboard") return templates.TemplateResponse(
resp.set_cookie(SESSION_COOKIE_NAME, create_session(user.id), httponly=True) "settings.html", {"request": request, "user": user, "success": "用户名修改成功"}
return resp )
@router.post("/settings/change-password")
def change_password(
request: Request,
old_password: str = Form(...),
new_password: str = Form(...),
confirm_password: str = Form(...),
user: User = Depends(get_current_user_web),
db: Session = Depends(get_db),
):
if not verify_password(old_password, user.password_hash):
return templates.TemplateResponse(
"settings.html", {"request": request, "user": user, "error": "当前密码错误"}, status_code=400
)
if new_password != confirm_password:
return templates.TemplateResponse(
"settings.html", {"request": request, "user": user, "error": "两次输入的新密码不一致"}, status_code=400
)
if len(new_password) < 6:
return templates.TemplateResponse(
"settings.html", {"request": request, "user": user, "error": "新密码长度至少6位"}, status_code=400
)
user.password_hash = hash_password(new_password)
db.commit()
return templates.TemplateResponse(
"settings.html", {"request": request, "user": user, "success": "密码修改成功"}
)
# ---------- Auth ----------
@router.get("/logout") @router.get("/logout")
def logout(request: Request, db: Session = Depends(get_db)): def logout(request: Request, db: Session = Depends(get_db)):
token = request.cookies.get(SESSION_COOKIE_NAME) token = request.cookies.get(SESSION_COOKIE_NAME)

View File

@@ -3,18 +3,6 @@ from datetime import datetime
from pydantic import BaseModel from pydantic import BaseModel
# ---------- Auth ----------
class RegisterRequest(BaseModel):
username: str
password: str
class LoginRequest(BaseModel):
username: str
password: str
# ---------- Profile ---------- # ---------- Profile ----------
class ProfileCreate(BaseModel): class ProfileCreate(BaseModel):
@@ -33,7 +21,7 @@ class ProfileOut(BaseModel):
class PostcardCreate(BaseModel): class PostcardCreate(BaseModel):
card_number: str card_number: str
status: str # sent / delivered / received status: str
recipient_name: str | None = None recipient_name: str | None = None
country: str | None = None country: str | None = None
send_time: datetime | None = None send_time: datetime | None = None
@@ -50,7 +38,7 @@ class PostcardUpdate(BaseModel):
send_time: datetime | None = None send_time: datetime | None = None
arrival_time: datetime | None = None arrival_time: datetime | None = None
sender_name: str | None = None sender_name: str | None = None
receive_time: datetime | None = None receive_time: str | None = None
class PostcardOut(BaseModel): class PostcardOut(BaseModel):
@@ -70,13 +58,3 @@ class PostcardOut(BaseModel):
updated_at: datetime updated_at: datetime
model_config = {"from_attributes": True} model_config = {"from_attributes": True}
# ---------- ApiKey ----------
class ApiKeyOut(BaseModel):
id: int
key: str
created_at: datetime
model_config = {"from_attributes": True}

View File

@@ -1,34 +0,0 @@
{% extends "base.html" %}
{% block title %}API Keys - Postcard Manager{% endblock %}
{% block content %}
<div class="section-header">
<h1>API Keys</h1>
<form method="post" action="/api-keys">
<button type="submit" class="btn btn-primary">+ 创建新 Key</button>
</form>
</div>
{% if keys %}
<table>
<thead><tr><th>Key</th><th>创建时间</th><th>操作</th></tr></thead>
<tbody>
{% for k in keys %}
<tr>
<td><code>{{ k.key }}</code></td>
<td>{{ k.created_at.strftime('%Y-%m-%d %H:%M') }}</td>
<td>
<form method="post" action="/api-keys/{{ k.id }}/delete" class="inline">
<button type="submit" class="btn btn-danger btn-sm" onclick="return confirm('确定删除该 Key')">删除</button>
</form>
</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<p class="empty">还没有 API Key点击上方按钮创建一个。</p>
{% endif %}
<section class="section" style="margin-top:2rem">
<h2>使用说明</h2>
<pre>curl -H "Authorization: Bearer YOUR_API_KEY" http://localhost:8000/api/profiles</pre>
</section>
{% endblock %}

View File

@@ -12,7 +12,7 @@
<div class="nav-links"> <div class="nav-links">
{% if user is defined and user %} {% if user is defined and user %}
<a href="/profiles">Profiles</a> <a href="/profiles">Profiles</a>
<a href="/api-keys">API Keys</a> <a href="/settings">Settings</a>
<a href="/logout" class="nav-logout">Logout</a> <a href="/logout" class="nav-logout">Logout</a>
{% endif %} {% endif %}
</div> </div>

View File

@@ -16,6 +16,5 @@
</label> </label>
<button type="submit" class="btn btn-primary">登录</button> <button type="submit" class="btn btn-primary">登录</button>
</form> </form>
<p class="auth-link">没有账号?<a href="/register">注册</a></p>
</div> </div>
{% endblock %} {% endblock %}

View File

@@ -1,20 +0,0 @@
{% extends "base.html" %}
{% block title %}注册 - Postcard Manager{% endblock %}
{% block content %}
<div class="auth-form">
<h2>注册</h2>
{% if error %}
<div class="alert alert-error">{{ error }}</div>
{% endif %}
<form method="post" action="/register">
<label>用户名</label>
<input type="text" name="username" required autofocus>
<label>密码</label>
<input type="password" name="password" required>
<label>确认密码</label>
<input type="password" name="password2" required>
<button type="submit" class="btn btn-primary">注册</button>
</form>
<p class="auth-link">已有账号?<a href="/login">登录</a></p>
</div>
{% endblock %}

View File

@@ -0,0 +1,44 @@
{% extends "base.html" %}
{% block title %}用户设置 - Postcard Manager{% endblock %}
{% block content %}
<div class="section-header">
<h1>用户设置</h1>
</div>
{% if success %}
<div class="alert alert-success">{{ success }}</div>
{% endif %}
{% if error %}
<div class="alert alert-error">{{ error }}</div>
{% endif %}
<div class="form-card" style="max-width:400px">
<h3 style="margin-bottom:.75rem">修改用户名</h3>
<form method="post" action="/settings/change-username">
<label>当前用户名</label>
<input type="text" value="{{ user.username }}" disabled>
<label>新用户名</label>
<input type="text" name="new_username" required maxlength="64" placeholder="输入新用户名">
<label>确认密码</label>
<input type="password" name="password" required placeholder="输入当前密码确认身份">
<div class="form-actions">
<button type="submit" class="btn btn-primary">修改用户名</button>
</div>
</form>
</div>
<div class="form-card" style="max-width:400px;margin-top:1.5rem">
<h3 style="margin-bottom:.75rem">修改密码</h3>
<form method="post" action="/settings/change-password">
<label>当前密码</label>
<input type="password" name="old_password" required placeholder="输入当前密码">
<label>新密码</label>
<input type="password" name="new_password" required placeholder="输入新密码">
<label>确认新密码</label>
<input type="password" name="confirm_password" required placeholder="再次输入新密码">
<div class="form-actions">
<button type="submit" class="btn btn-primary">修改密码</button>
</div>
</form>
</div>
{% endblock %}

24
test_clean.py Normal file
View File

@@ -0,0 +1,24 @@
from fastapi.testclient import TestClient
from app.main import app
c = TestClient(app)
# login
r = c.post("/login", data={"username": "admin", "password": "admin123"}, follow_redirects=False)
print("login:", r.status_code)
session = r.cookies.get("pc_session")
if session:
c.cookies.set("pc_session", session)
r = c.get("/dashboard")
print("dashboard:", r.status_code)
assert r.status_code == 200
r = c.get("/profiles")
print("profiles:", r.status_code)
assert r.status_code == 200
print("All checks passed")
else:
print("login failed, no session")
r2 = c.post("/login", data={"username": "admin", "password": "admin123"}, follow_redirects=True)
print("follow:", r2.status_code, "pc_session:", r2.cookies.get("pc_session"))