import logging from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from models.database import get_db from models.models import User from .account_limits import ( UNLIMITED_ACCOUNTS, count_user_account_breakdown, normalize_max_accounts, ) from .account_quota import default_stop_worker, sync_user_account_quota from .dependencies import get_current_user, require_user_manager from .email_service import ( build_password_reset_link, build_verification_link, send_password_reset_email, send_verification_email, ) from .email_verification import create_verification_token, mask_email, verify_email_token from .password_reset import create_password_reset_token, verify_password_reset_token from .jwt_utils import create_access_token from .passwords import hash_password, verify_password from .roles import ALL_ROLES, ROLE_LABELS, ROLE_OPERATOR, ensure_role, is_admin from .schemas import ( LoginRequest, MessageResponse, ForgotPasswordRequest, ForgotPasswordResponse, RegisterRequest, RegisterResponse, ResendVerificationRequest, ResetPasswordRequest, RolesResponse, RoleInfo, TokenResponse, UserCreate, UserResponse, UserUpdate, VerifyEmailRequest, ) from .system_settings import SystemSettingsData, load_settings logger = logging.getLogger("auth.router") router = APIRouter(prefix="/api/auth", tags=["auth"]) async def _build_user_response(db: AsyncSession, user: User, with_count: bool = False) -> UserResponse: payload = UserResponse.model_validate(user) if with_count: breakdown = await count_user_account_breakdown(db, user.id) payload.account_count = breakdown["total"] payload.active_account_count = breakdown["active"] payload.disabled_account_count = breakdown["disabled"] return payload async def _ensure_email_available( db: AsyncSession, email: str | None, exclude_user_id: int | None = None, ) -> str | None: value = (str(email).strip().lower() if email else "") or None if not value: return None stmt = select(User).where(User.email == value) if exclude_user_id: stmt = stmt.where(User.id != exclude_user_id) exists = await db.execute(stmt) if exists.scalar_one_or_none(): raise HTTPException(status_code=400, detail="邮箱已被其他用户使用") return value def _apply_email_verified(user: User, verified: bool | None) -> None: if verified is None: return user.email_verified = verified if verified: if not user.email_verified_at: user.email_verified_at = datetime.utcnow() else: user.email_verified_at = None def _email_not_verified_detail(user: User) -> dict: return { "code": "email_not_verified", "message": "邮箱尚未验证,请先完成邮箱验证后再登录", "email": mask_email(user.email or ""), } def _email_not_bound_detail() -> dict: return { "code": "email_not_bound", "message": "该账号未绑定邮箱,请联系管理员绑定邮箱后再登录", } def _require_email_access(user: User, settings: SystemSettingsData) -> None: if is_admin(user.role): return if settings.email_binding_required and not user.email: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=_email_not_bound_detail(), ) if not settings.email_verification_required: return if user.email_verified: return if user.email: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=_email_not_verified_detail(user), ) async def _issue_token(user: User) -> TokenResponse: token = create_access_token(str(user.id), {"role": user.role, "username": user.username}) return TokenResponse(access_token=token) async def _send_user_verification( db: AsyncSession, user: User, settings: SystemSettingsData, ) -> tuple[bool, str | None]: if not user.email: raise HTTPException(status_code=400, detail="该账号未绑定邮箱") token = await create_verification_token( db, user, expire_hours=settings.email_verify_token_hours ) await db.commit() link = build_verification_link(token, settings) if not settings.smtp_configured(): logger.warning( "SMTP 未配置,验证链接: user=%s email=%s link=%s", user.username, user.email, link, ) dev_url = link if settings.debug_show_verify_link else None return False, dev_url try: await send_verification_email(user.email, user.username, token, settings) return True, None except Exception as exc: logger.exception("Send verification email failed for user=%s", user.username) dev_url = link if settings.debug_show_verify_link else None if dev_url: return False, dev_url raise HTTPException(status_code=400, detail=f"邮件发送失败:{exc}") from exc @router.post("/login", response_model=TokenResponse) async def login(body: LoginRequest, db: AsyncSession = Depends(get_db)): settings = await load_settings(db) result = await db.execute(select(User).where(User.username == body.username.strip())) user = result.scalar_one_or_none() if not user or not verify_password(body.password, user.password_hash): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误") if not user.is_active: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="账号已禁用") _require_email_access(user, settings) return await _issue_token(user) @router.post("/register", response_model=RegisterResponse) async def register(body: RegisterRequest, db: AsyncSession = Depends(get_db)): settings = await load_settings(db) if not settings.registration_enabled: raise HTTPException(status_code=403, detail="当前未开放用户注册") username = body.username.strip() email = str(body.email).strip().lower() exists = await db.execute(select(User).where(User.username == username)) if exists.scalar_one_or_none(): raise HTTPException(status_code=400, detail="用户名已存在") exists = await db.execute(select(User).where(User.email == email)) if exists.scalar_one_or_none(): raise HTTPException(status_code=400, detail="邮箱已被注册") verification_required = settings.email_verification_required user = User( username=username, email=email, password_hash=hash_password(body.password), display_name=body.display_name or username, role=ROLE_OPERATOR, is_active=True, email_verified=not verification_required, email_verified_at=datetime.utcnow() if not verification_required else None, max_accounts=max(0, int(settings.default_register_max_accounts or 3)), ) db.add(user) await db.flush() if not verification_required: await db.commit() await db.refresh(user) return RegisterResponse( message="注册成功,可直接登录", email=mask_email(user.email), verification_sent=False, verification_required=False, ) verification_sent, dev_url = await _send_user_verification(db, user, settings) await db.refresh(user) message = "注册成功,验证邮件已发送,请查收并完成验证后再登录" if not verification_sent: message = ( "注册成功。验证邮件未能发出,请使用下方链接或联系管理员检查 SMTP 配置" if dev_url else "注册成功。邮件服务未配置或发送失败,请联系管理员" ) return RegisterResponse( message=message, email=mask_email(user.email), verification_sent=verification_sent, verification_required=True, dev_verify_url=dev_url, ) @router.post("/verify-email", response_model=MessageResponse) async def verify_email(body: VerifyEmailRequest, db: AsyncSession = Depends(get_db)): user = await verify_email_token(db, body.token.strip()) if not user: raise HTTPException(status_code=400, detail="验证链接无效或已过期") await db.commit() return MessageResponse(message="邮箱验证成功,现在可以登录了") @router.get("/verify-email", response_model=MessageResponse) async def verify_email_get(token: str = Query(..., min_length=8), db: AsyncSession = Depends(get_db)): user = await verify_email_token(db, token.strip()) if not user: raise HTTPException(status_code=400, detail="验证链接无效或已过期") await db.commit() return MessageResponse(message="邮箱验证成功,现在可以登录了") @router.post("/forgot-password", response_model=ForgotPasswordResponse) async def forgot_password(body: ForgotPasswordRequest, db: AsyncSession = Depends(get_db)): settings = await load_settings(db) if not settings.smtp_configured(): raise HTTPException( status_code=400, detail="邮件服务未配置,请联系管理员重置密码", ) email = str(body.email).strip().lower() if body.email else None username = body.username.strip() if body.username else None if not email and not username: raise HTTPException(status_code=400, detail="请提供邮箱或用户名") user = None if email: result = await db.execute(select(User).where(User.email == email)) user = result.scalar_one_or_none() if not user and username: result = await db.execute(select(User).where(User.username == username)) user = result.scalar_one_or_none() generic = ForgotPasswordResponse( message="若账号存在且已绑定邮箱,重置邮件将发送到注册邮箱", email=mask_email(user.email if user and user.email else (email or "")), reset_sent=True, ) if not user or not user.email or not user.is_active: return generic token = await create_password_reset_token( db, user, expire_hours=settings.email_verify_token_hours ) await db.commit() link = build_password_reset_link(token, settings) try: await send_password_reset_email(user.email, user.username, token, settings) except Exception as exc: logger.exception("Send password reset email failed for user=%s", user.username) dev_url = link if settings.debug_show_verify_link else None if dev_url: return ForgotPasswordResponse( message="邮件发送失败,请使用下方开发重置链接", email=mask_email(user.email), reset_sent=False, dev_reset_url=dev_url, ) raise HTTPException(status_code=400, detail=f"邮件发送失败:{exc}") from exc return ForgotPasswordResponse( message="重置邮件已发送,请查收并按邮件说明设置新密码", email=mask_email(user.email), reset_sent=True, dev_reset_url=None, ) @router.post("/reset-password", response_model=MessageResponse) async def reset_password(body: ResetPasswordRequest, db: AsyncSession = Depends(get_db)): user = await verify_password_reset_token(db, body.token.strip()) if not user: raise HTTPException(status_code=400, detail="重置链接无效或已过期") if not user.is_active: raise HTTPException(status_code=400, detail="账号已禁用,请联系管理员") user.password_hash = hash_password(body.password) await db.commit() return MessageResponse(message="密码已重置,请使用新密码登录") @router.post("/resend-verification", response_model=RegisterResponse) async def resend_verification(body: ResendVerificationRequest, db: AsyncSession = Depends(get_db)): settings = await load_settings(db) if not settings.email_verification_required: raise HTTPException(status_code=400, detail="当前系统未开启邮箱验证") email = str(body.email).strip().lower() if body.email else None username = body.username.strip() if body.username else None if not email and not username: raise HTTPException(status_code=400, detail="请提供邮箱或用户名") user = None if email: result = await db.execute(select(User).where(User.email == email)) user = result.scalar_one_or_none() if not user and username: result = await db.execute(select(User).where(User.username == username)) user = result.scalar_one_or_none() if not user: return RegisterResponse( message="若账号存在且未验证,验证邮件将发送到注册邮箱", email=mask_email(email or ""), verification_sent=True, verification_required=True, ) if user.email_verified: raise HTTPException(status_code=400, detail="该账号邮箱已验证,可直接登录") verification_sent, dev_url = await _send_user_verification(db, user, settings) message = ( "验证邮件已重新发送,请查收" if verification_sent else "邮件未能发出,请使用下方验证链接或联系管理员检查 SMTP 配置" ) return RegisterResponse( message=message, email=mask_email(user.email or ""), verification_sent=verification_sent, verification_required=True, dev_verify_url=dev_url, ) @router.get("/me", response_model=UserResponse) async def get_me(user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): return await _build_user_response(db, user, with_count=True) @router.get("/roles", response_model=RolesResponse) async def list_roles(_: User = Depends(get_current_user)): return RolesResponse( roles=[RoleInfo(value=r, label=ROLE_LABELS.get(r, r)) for r in ALL_ROLES] ) users_router = APIRouter(prefix="/api/users", tags=["users"]) @users_router.get("", response_model=list[UserResponse]) async def list_users( db: AsyncSession = Depends(get_db), _: User = Depends(require_user_manager), ): result = await db.execute(select(User).order_by(User.id.asc())) users = result.scalars().all() responses = [] for user in users: responses.append(await _build_user_response(db, user, with_count=True)) return responses @users_router.post("", response_model=UserResponse) async def create_user( body: UserCreate, db: AsyncSession = Depends(get_db), _: User = Depends(require_user_manager), ): settings = await load_settings(db) exists = await db.execute(select(User).where(User.username == body.username)) if exists.scalar_one_or_none(): raise HTTPException(status_code=400, detail="用户名已存在") try: role = ensure_role(body.role) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) email = await _ensure_email_available(db, str(body.email) if body.email else None) if settings.email_binding_required and not is_admin(role) and not email: raise HTTPException(status_code=400, detail="系统已开启「登录必须绑定邮箱」,请填写邮箱") if email: email_verified = body.email_verified else: email_verified = not settings.email_binding_required user = User( username=body.username.strip(), password_hash=hash_password(body.password), display_name=body.display_name or body.username, role=role, is_active=True, email=email, email_verified=email_verified, email_verified_at=datetime.utcnow() if email and email_verified else None, max_accounts=normalize_max_accounts(body.max_accounts, role), ) db.add(user) await db.commit() await db.refresh(user) return await _build_user_response(db, user, with_count=True) @users_router.put("/{user_id}", response_model=UserResponse) async def update_user( user_id: int, body: UserUpdate, db: AsyncSession = Depends(get_db), current: User = Depends(require_user_manager), ): result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="用户不存在") settings = await load_settings(db) if user.id == current.id and body.is_active is False: raise HTTPException(status_code=400, detail="不能禁用当前登录账号") if body.display_name is not None: user.display_name = body.display_name if body.role is not None: prev_role = user.role try: user.role = ensure_role(body.role) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) if is_admin(user.role): user.max_accounts = UNLIMITED_ACCOUNTS await sync_user_account_quota(db, user, stop_worker=default_stop_worker) elif is_admin(prev_role) and not is_admin(user.role): settings = await load_settings(db) user.max_accounts = max(0, int(settings.default_register_max_accounts or 3)) await sync_user_account_quota(db, user, stop_worker=default_stop_worker) if body.is_active is not None: user.is_active = body.is_active if body.password: user.password_hash = hash_password(body.password) updates = body.model_dump(exclude_unset=True) if "max_accounts" in updates and not is_admin(user.role): user.max_accounts = normalize_max_accounts(updates["max_accounts"], user.role) await sync_user_account_quota(db, user, stop_worker=default_stop_worker) if "email" in updates: raw_email = updates.get("email") user.email = await _ensure_email_available( db, str(raw_email) if raw_email else None, exclude_user_id=user.id, ) if not user.email: user.email_verified = not ( settings.email_binding_required and not is_admin(user.role) ) user.email_verified_at = None if "email_verified" in updates: if not user.email: raise HTTPException(status_code=400, detail="未绑定邮箱时无法设置验证状态") _apply_email_verified(user, updates["email_verified"]) if settings.email_binding_required and not is_admin(user.role) and not user.email: raise HTTPException(status_code=400, detail="系统已开启「登录必须绑定邮箱」,该用户需绑定邮箱") await db.commit() await db.refresh(user) return await _build_user_response(db, user, with_count=True) @users_router.delete("/{user_id}") async def delete_user( user_id: int, db: AsyncSession = Depends(get_db), current: User = Depends(require_user_manager), ): if user_id == current.id: raise HTTPException(status_code=400, detail="不能删除当前登录账号") result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="用户不存在") await db.delete(user) await db.commit() return {"message": "用户已删除"}