feat(invite): add invite code registration system
This commit is contained in:
@@ -17,7 +17,11 @@ from app.auth import (
|
||||
)
|
||||
from app.config import SECRET_KEY, SESSION_COOKIE_NAME, UPLOAD_DIR
|
||||
from app.database import get_db
|
||||
from app.models import Postcard, Profile, User
|
||||
from datetime import datetime
|
||||
from random import choice
|
||||
from string import ascii_letters, digits
|
||||
|
||||
from app.models import InviteCode, Postcard, Profile, User
|
||||
|
||||
router = APIRouter(tags=["web"])
|
||||
templates = Jinja2Templates(directory=str(Path(__file__).resolve().parent.parent / "templates"))
|
||||
@@ -71,6 +75,189 @@ def login_submit(
|
||||
return resp
|
||||
|
||||
|
||||
# ---------- Register ----------
|
||||
|
||||
@router.get("/register", response_class=HTMLResponse)
|
||||
def register_page(request: Request, db: Session = Depends(get_db)):
|
||||
user = redirect_if_not_logged_in(request, db)
|
||||
if user:
|
||||
return _redirect("/dashboard")
|
||||
return templates.TemplateResponse("register.html", {"request": request})
|
||||
|
||||
|
||||
@router.post("/register")
|
||||
def register_submit(
|
||||
request: Request,
|
||||
invite_code: str = Form(...),
|
||||
username: str = Form(...),
|
||||
password: str = Form(...),
|
||||
confirm_password: str = Form(...),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
# Validate invite code
|
||||
code_obj = db.query(InviteCode).filter(InviteCode.code == invite_code).first()
|
||||
if not code_obj:
|
||||
return templates.TemplateResponse(
|
||||
"register.html", {"request": request, "error": "邀请码无效"}, status_code=400
|
||||
)
|
||||
if code_obj.expires_at and code_obj.expires_at < datetime.utcnow():
|
||||
return templates.TemplateResponse(
|
||||
"register.html", {"request": request, "error": "邀请码已过期"}, status_code=400
|
||||
)
|
||||
if code_obj.max_uses is not None and code_obj.use_count >= code_obj.max_uses:
|
||||
return templates.TemplateResponse(
|
||||
"register.html", {"request": request, "error": "邀请码使用次数已用尽"}, status_code=400
|
||||
)
|
||||
# Validate passwords
|
||||
if password != confirm_password:
|
||||
return templates.TemplateResponse(
|
||||
"register.html", {"request": request, "error": "两次输入的密码不一致"}, status_code=400
|
||||
)
|
||||
if len(password) < 6:
|
||||
return templates.TemplateResponse(
|
||||
"register.html", {"request": request, "error": "密码至少6个字符"}, status_code=400
|
||||
)
|
||||
# Check username
|
||||
if db.query(User).filter(User.username == username).first():
|
||||
return templates.TemplateResponse(
|
||||
"register.html", {"request": request, "error": "用户名已存在"}, status_code=400
|
||||
)
|
||||
# Create user
|
||||
new_user = User(username=username, password_hash=hash_password(password))
|
||||
db.add(new_user)
|
||||
db.flush()
|
||||
# Update invite code usage
|
||||
code_obj.use_count += 1
|
||||
db.commit()
|
||||
# Auto-login
|
||||
token = create_session(new_user.id, db)
|
||||
resp = _redirect("/dashboard")
|
||||
resp.set_cookie(SESSION_COOKIE_NAME, token, httponly=True, samesite="lax")
|
||||
return resp
|
||||
|
||||
|
||||
# ---------- Admin ----------
|
||||
|
||||
def _generate_code() -> str:
|
||||
"""Generate a random 8-character alphanumeric code."""
|
||||
return "".join(choice(ascii_letters + digits) for _ in range(8))
|
||||
|
||||
|
||||
def _require_admin(user: User) -> bool:
|
||||
return user.is_admin
|
||||
|
||||
|
||||
@router.get("/admin/invites", response_class=HTMLResponse)
|
||||
def admin_invites_page(
|
||||
request: Request,
|
||||
user: User = Depends(get_current_user_web),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
if not _require_admin(user):
|
||||
return _redirect("/dashboard")
|
||||
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
||||
return templates.TemplateResponse("admin_invites.html", {
|
||||
"request": request, "user": user, "codes": codes,
|
||||
})
|
||||
|
||||
|
||||
@router.post("/admin/invites/add")
|
||||
def admin_add_invite(
|
||||
request: Request,
|
||||
code: str = Form(""),
|
||||
limit_type: str = Form("unlimited"),
|
||||
max_uses: int = Form(None),
|
||||
expires_at: str = Form(None),
|
||||
user: User = Depends(get_current_user_web),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
if not _require_admin(user):
|
||||
return _redirect("/dashboard")
|
||||
# Generate or validate code
|
||||
if not code.strip():
|
||||
code = _generate_code()
|
||||
else:
|
||||
code = code.strip()
|
||||
if len(code) != 8:
|
||||
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
||||
return templates.TemplateResponse("admin_invites.html", {
|
||||
"request": request, "user": user, "codes": codes,
|
||||
"error": "邀请码必须为8个字符",
|
||||
})
|
||||
if db.query(InviteCode).filter(InviteCode.code == code).first():
|
||||
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
||||
return templates.TemplateResponse("admin_invites.html", {
|
||||
"request": request, "user": user, "codes": codes,
|
||||
"error": "邀请码已存在",
|
||||
})
|
||||
# Parse limits
|
||||
m_uses = None
|
||||
exp_at = None
|
||||
if limit_type == "uses" and max_uses:
|
||||
m_uses = max_uses
|
||||
elif limit_type == "expires" and expires_at:
|
||||
exp_at = datetime.fromisoformat(expires_at)
|
||||
new_code = InviteCode(
|
||||
code=code, max_uses=m_uses, expires_at=exp_at,
|
||||
use_count=0, created_by=user.id,
|
||||
)
|
||||
db.add(new_code)
|
||||
db.commit()
|
||||
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
||||
return templates.TemplateResponse("admin_invites.html", {
|
||||
"request": request, "user": user, "codes": codes,
|
||||
"success": f"邀请码 {code} 已创建",
|
||||
})
|
||||
|
||||
|
||||
@router.post("/admin/invites/{code_id}/edit")
|
||||
def admin_edit_invite(
|
||||
request: Request,
|
||||
code_id: int,
|
||||
limit_type: str = Form("unlimited"),
|
||||
max_uses: int = Form(None),
|
||||
expires_at: str = Form(None),
|
||||
user: User = Depends(get_current_user_web),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
if not _require_admin(user):
|
||||
return _redirect("/dashboard")
|
||||
code_obj = db.query(InviteCode).filter(InviteCode.id == code_id).first()
|
||||
if not code_obj:
|
||||
return _redirect("/admin/invites")
|
||||
if limit_type == "uses" and max_uses:
|
||||
code_obj.max_uses = max_uses
|
||||
code_obj.expires_at = None
|
||||
elif limit_type == "expires" and expires_at:
|
||||
code_obj.expires_at = datetime.fromisoformat(expires_at)
|
||||
code_obj.max_uses = None
|
||||
else:
|
||||
code_obj.max_uses = None
|
||||
code_obj.expires_at = None
|
||||
db.commit()
|
||||
codes = db.query(InviteCode).order_by(InviteCode.created_at.desc()).all()
|
||||
return templates.TemplateResponse("admin_invites.html", {
|
||||
"request": request, "user": user, "codes": codes,
|
||||
"success": f"邀请码 {code_obj.code} 已更新",
|
||||
})
|
||||
|
||||
|
||||
@router.post("/admin/invites/{code_id}/delete")
|
||||
def admin_delete_invite(
|
||||
request: Request,
|
||||
code_id: int,
|
||||
user: User = Depends(get_current_user_web),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
if not _require_admin(user):
|
||||
return _redirect("/dashboard")
|
||||
code_obj = db.query(InviteCode).filter(InviteCode.id == code_id).first()
|
||||
if code_obj:
|
||||
db.delete(code_obj)
|
||||
db.commit()
|
||||
return _redirect("/admin/invites")
|
||||
|
||||
|
||||
# ---------- Settings ----------
|
||||
|
||||
@router.get("/settings", response_class=HTMLResponse)
|
||||
|
||||
Reference in New Issue
Block a user