feat(mgr-api): 添加可用性检查接口 & 更新认证白名单

- 新增 GET /user/availability/mail（免认证，邮箱可用性）
- 新增 GET /user/availability/username（免认证，用户名可用性）
- 新增 GET /account/availability/slug（需登录，Slug 可用性）
- 更新 AdminAuthMiddleware 免认证前缀白名单
- ruff + black 格式化

Author: kaokao221

app/api/management/account_availability_slug.py

@@ -0,0 +1,37 @@
+"""账户 Slug 可用性检查端点。
+
+供已登录用户创建账户时校验 Slug 是否可用，需身份认证。
+校验流程：格式合法性 → 保留名检查 → 唯一性查询。
+"""
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.account import Account
+from app.models.session import get_db
+from app.services.user.name_validator import (
+    is_name_reserved,
+    validate_slug_format,
+)
+
+router = APIRouter()
+
+
+@router.get("/slug")
+async def check_slug_availability(
+    value: str = Query(..., min_length=3, max_length=64, description="待检查的 Slug"),
+    db: AsyncSession = Depends(get_db),
+):
+    """检查账户 Slug 是否可用。
+
+    依次校验格式、保留名和唯一性，返回结果与不可用原因。
+    """
+    if not validate_slug_format(value):
+        return {"available": False, "reason": "Slug 格式不合法"}
+    if await is_name_reserved(value, db):
+        return {"available": False, "reason": "该名称为系统保留名"}
+    result = await db.execute(select(Account.id).where(Account.slug == value))
+    if result.scalar_one_or_none() is not None:
+        return {"available": False, "reason": "该 Slug 已被占用"}
+    return {"available": True}

app/api/management/account_pre_reg_crud.py

@@ -1,6 +1,5 @@
 """预注册客户端 CRUD 路由。"""
 
-
 from fastapi import APIRouter, Depends
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession

app/api/management/account_router.py

@@ -25,10 +25,16 @@
 from .account_invitation import router as inv_r
 from .account_invitation_create import router as inv_c_r
 from .account_invitation_revoke import router as inv_v_r
+from .account_availability_slug import router as avail_slug_r
 from .bulk import router as bulk_r
 
 router = APIRouter()
 
+# /account/availability/*（需认证）
+router.include_router(
+    avail_slug_r, prefix="/account/availability", tags=["Availability"]
+)
+
 # /account 顶层
 router.include_router(list_r, prefix="/account", tags=["Account"])
 router.include_router(search_r, prefix="/account", tags=["Account"])

app/api/management/router.py

@@ -10,6 +10,8 @@
 from .token_deactivate import router as deact_r
 from .user_apply import router as apply_r
 from .user_auth import router as auth_r
+from .user_availability_mail import router as avail_mail_r
+from .user_availability_username import router as avail_uname_r
 from .user_info import router as info_r
 from .user_totp import router as totp_r
 from .user_info_email import router as email_r
@@ -29,6 +31,10 @@
 router.include_router(info_r, prefix="/user", tags=["User"])
 router.include_router(totp_r, prefix="/user/2fa/totp", tags=["2FA"])
 
+# /user/availability/*（无需认证）
+router.include_router(avail_mail_r, prefix="/user/availability", tags=["Availability"])
+router.include_router(avail_uname_r, prefix="/user/availability", tags=["Availability"])
+
 # /user/info/*
 router.include_router(email_r, prefix="/user/info", tags=["UserInfo"])
 router.include_router(uname_r, prefix="/user/info", tags=["UserInfo"])

app/api/management/user_availability_mail.py

@@ -0,0 +1,26 @@
+"""邮箱可用性检查端点。
+
+供注册页前端实时校验邮箱是否已被占用，无需身份认证。
+"""
+
+from fastapi import APIRouter, Query
+from pydantic import EmailStr
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from fastapi import Depends
+
+from app.models.session import get_db
+from app.models.user import User
+
+router = APIRouter()
+
+
+@router.get("/mail")
+async def check_mail_availability(
+    value: EmailStr = Query(..., description="待检查的邮箱地址"),
+    db: AsyncSession = Depends(get_db),
+):
+    """检查邮箱是否可用（未被注册）。"""
+    result = await db.execute(select(User.id).where(User.email == value))
+    exists = result.scalar_one_or_none() is not None
+    return {"available": not exists}

app/api/management/user_availability_username.py

@@ -0,0 +1,37 @@
+"""用户名可用性检查端点。
+
+供注册页前端实时校验用户名是否可用，无需身份认证。
+校验流程：格式合法性 → 保留名检查 → 唯一性查询。
+"""
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.session import get_db
+from app.models.user import User
+from app.services.user.name_validator import (
+    is_name_reserved,
+    validate_username_format,
+)
+
+router = APIRouter()
+
+
+@router.get("/username")
+async def check_username_availability(
+    value: str = Query(..., min_length=3, max_length=64, description="待检查的用户名"),
+    db: AsyncSession = Depends(get_db),
+):
+    """检查用户名是否可用。
+
+    依次校验格式、保留名和唯一性，返回结果与不可用原因。
+    """
+    if not validate_username_format(value):
+        return {"available": False, "reason": "用户名格式不合法"}
+    if await is_name_reserved(value, db):
+        return {"available": False, "reason": "该名称为系统保留名"}
+    result = await db.execute(select(User.id).where(User.username == value))
+    if result.scalar_one_or_none() is not None:
+        return {"available": False, "reason": "该用户名已被占用"}
+    return {"available": True}

app/api/schemas/account.py

@@ -24,9 +24,7 @@ class AccountCreate(BaseModel):
     def _check_slug(cls, v: Optional[str]) -> Optional[str]:
         """Slug 格式校验（若提供）。"""
         if v is not None and not _SLUG_RE.match(v):
-            raise ValueError(
-                "Slug 需 3~64 位小写字母数字连字符，" "不以连字符开头/结尾"
-            )
+            raise ValueError("Slug 需 3~64 位小写字母数字连字符，不以连字符开头/结尾")
         return v
 
 

app/core/auth/admin_middleware.py

@@ -20,8 +20,10 @@
 
 logger = logging.getLogger(__name__)
 
-# 免认证路径白名单
+# 免认证路径白名单（精确匹配）
 _EXEMPT = {"/", "/user/auth", "/user/apply"}
+# 免认证路径前缀（前缀匹配）
+_EXEMPT_PREFIXES = ("/user/availability/",)
 _DENY = JSONResponse(status_code=403, content={"detail": "权限不足"})
 _NO_AUTH = JSONResponse(status_code=401, content={"detail": "未认证"})
 
@@ -37,7 +39,7 @@ async def dispatch(
         if request.method == "OPTIONS":
             return await call_next(request)
         path = request.url.path
-        if path in _EXEMPT:
+        if path in _EXEMPT or path.startswith(_EXEMPT_PREFIXES):
             return await call_next(request)
         # 提取 Bearer Token
         auth = request.headers.get("authorization", "")