新增 /profiles/me、/me/stats、/me/clouds 快捷路由，放开 stats 本人查看权限

app/routers/profiles.py

@@ -7,9 +7,16 @@ from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import joinedload
 
-from app.deps import AdminUser, CurrentUser, DbSession, OptionalUser
+from app.deps import CurrentUser, DbSession, OptionalUser
 from app.models import Cloud, Profile, User, UserCollection
-from app.schemas import CloudOut, PageOut, ProfileOut, ProfileStatsOut, ProfileUpdateIn, PublicProfileOut
+from app.schemas import (
+    CloudOut,
+    PageOut,
+    ProfileOut,
+    ProfileStatsOut,
+    ProfileUpdateIn,
+    PublicProfileOut,
+)
 from app.serializers import cloud_out, profile_out, public_profile_out
 from app.services.storage import delete_files, save_avatar
 
@@ -17,12 +24,14 @@ from app.services.storage import delete_files, save_avatar
 router = APIRouter(prefix="/profiles", tags=["用户资料"])
 
 
-@router.get("/{user_id}", response_model=PublicProfileOut)
+# ---------------------------------------------------------------------------
-async def get_profile(user_id: uuid.UUID, db: DbSession) -> PublicProfileOut:
+# /me 路由（必须定义在 /{user_id} 之前，避免 "me" 被当作 UUID 解析）
-    profile = await db.get(Profile, user_id)
+# ---------------------------------------------------------------------------
-    if not profile:
+
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
+
-    return public_profile_out(profile)
+@router.get("/me", response_model=PublicProfileOut)
+async def get_my_profile(user: CurrentUser, db: DbSession) -> PublicProfileOut:
+    return public_profile_out(user.profile)
 
 
 @router.patch("/me", response_model=ProfileOut)
@@ -71,12 +80,64 @@ async def delete_avatar(
     return profile_out(user.profile)
 
 
+@router.get("/me/stats", response_model=ProfileStatsOut)
+async def get_my_stats(user: CurrentUser, db: DbSession) -> ProfileStatsOut:
+    return await _build_stats(user.id, db)
+
+
+@router.get("/me/clouds", response_model=PageOut[CloudOut])
+async def get_my_clouds(
+    user: CurrentUser,
+    db: DbSession,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=100),
+) -> PageOut[CloudOut]:
+    return await _list_profile_clouds(user.id, db, viewer_role=user.profile.role, viewer_id=user.id, page=page, page_size=page_size)
+
+
+# ---------------------------------------------------------------------------
+# /{user_id} 路由
+# ---------------------------------------------------------------------------
+
+
+@router.get("/{user_id}", response_model=PublicProfileOut)
+async def get_profile(user_id: uuid.UUID, db: DbSession) -> PublicProfileOut:
+    profile = await db.get(Profile, user_id)
+    if not profile:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
+    return public_profile_out(profile)
+
+
 @router.get("/{user_id}/stats", response_model=ProfileStatsOut)
 async def get_user_stats(
     user_id: uuid.UUID,
-    _: AdminUser,
+    viewer: CurrentUser,
     db: DbSession,
 ) -> ProfileStatsOut:
+    if viewer.id != user_id and viewer.profile.role != "admin":
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只能查看自己的统计数据")
+    return await _build_stats(user_id, db)
+
+
+@router.get("/{user_id}/clouds", response_model=PageOut[CloudOut])
+async def get_profile_clouds(
+    user_id: uuid.UUID,
+    db: DbSession,
+    viewer: OptionalUser,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=100),
+) -> PageOut[CloudOut]:
+    viewer_role = viewer.profile.role if viewer else None
+    viewer_id = viewer.id if viewer else None
+    return await _list_profile_clouds(user_id, db, viewer_role=viewer_role, viewer_id=viewer_id, page=page, page_size=page_size)
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+
+async def _build_stats(user_id: uuid.UUID, db: DbSession) -> ProfileStatsOut:
     result = await db.execute(
         select(User).options(joinedload(User.profile)).where(User.id == user_id)
     )
@@ -84,19 +145,28 @@ async def get_user_stats(
     if not user:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
 
-    cloud_count = await db.scalar(
+    cloud_count = (
-        select(func.count()).select_from(Cloud).where(Cloud.user_id == user_id)
+        await db.scalar(select(func.count()).select_from(Cloud).where(Cloud.user_id == user_id))
-    ) or 0
+        or 0
-    public_cloud_count = await db.scalar(
+    )
-        select(func.count()).select_from(Cloud).where(
+    public_cloud_count = (
+        await db.scalar(
+            select(func.count())
+            .select_from(Cloud)
+            .where(
                 Cloud.user_id == user_id,
                 Cloud.status == "approved",
                 Cloud.is_hidden.is_(False),
             )
-    ) or 0
+        )
-    collection_count = await db.scalar(
+        or 0
+    )
+    collection_count = (
+        await db.scalar(
             select(func.count()).select_from(UserCollection).where(UserCollection.user_id == user_id)
-    ) or 0
+        )
+        or 0
+    )
 
     return ProfileStatsOut(
         user_id=user.id,
@@ -109,18 +179,19 @@ async def get_user_stats(
     )
 
 
-@router.get("/{user_id}/clouds", response_model=PageOut[CloudOut])
+async def _list_profile_clouds(
-async def get_profile_clouds(
     user_id: uuid.UUID,
     db: DbSession,
-    viewer: OptionalUser,
+    *,
-    page: int = Query(1, ge=1),
+    viewer_role: str | None,
-    page_size: int = Query(50, ge=1, le=100),
+    viewer_id: uuid.UUID | None,
+    page: int,
+    page_size: int,
 ) -> PageOut[CloudOut]:
     if not await db.get(Profile, user_id):
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
 
-    can_manage = bool(viewer and (viewer.id == user_id or viewer.profile.role == "admin"))
+    can_manage = bool(viewer_id and (viewer_id == user_id or viewer_role == "admin"))
     filters = [Cloud.user_id == user_id]
     if not can_manage:
         filters.extend([Cloud.status == "approved", Cloud.is_hidden.is_(False)])

tests/test_profiles.py

@@ -112,6 +112,30 @@ def test_get_user_clouds_public(client: TestClient, monkeypatch) -> None:
         assert item["status"] == "approved"
 
 
+def test_get_my_clouds_via_me(client: TestClient, monkeypatch) -> None:
+    info = register_and_confirm(
+        client, monkeypatch,
+        f"myclouds-{uuid.uuid4().hex[:8]}@example.com",
+        "myclouds1", "me 云图",
+    )
+    client.post(
+        "/api/v1/clouds",
+        headers=bearer(info["token"]),
+        files={"image": ("cloud.png", png_bytes(), "image/png")},
+        data={"cloud_type_id": "1", "is_hidden": "false"},
+    )
+    r = client.get("/api/v1/profiles/me/clouds", headers=bearer(info["token"]))
+    assert r.status_code == 200
+    assert r.json()["total"] >= 1
+    # 本人可以看到 pending 状态
+    assert r.json()["items"][0]["status"] == "pending"
+
+
+def test_get_my_clouds_via_me_unauthenticated(client: TestClient) -> None:
+    r = client.get("/api/v1/profiles/me/clouds")
+    assert r.status_code == 401
+
+
 def test_get_user_clouds_as_owner(client: TestClient, monkeypatch) -> None:
     info = register_and_confirm(
         client, monkeypatch,
@@ -139,17 +163,72 @@ def test_get_nonexistent_user_clouds(client: TestClient) -> None:
 
 
 # ---------------------------------------------------------------------------
-# stats (admin only)
+# /me profile shortcut
 # ---------------------------------------------------------------------------
 
 
+def test_get_my_profile_via_me(client: TestClient, monkeypatch) -> None:
+    info = register_and_confirm(
+        client, monkeypatch,
+        f"me-prof-{uuid.uuid4().hex[:8]}@example.com",
+        "meprof12", "me 资料",
+    )
+    r = client.get("/api/v1/profiles/me", headers=bearer(info["token"]))
+    assert r.status_code == 200
+    assert r.json()["username"] == "me 资料"
+    assert r.json()["id"] == info["user_id"]
+
+
+def test_get_my_profile_via_me_unauthenticated(client: TestClient) -> None:
+    r = client.get("/api/v1/profiles/me")
+    assert r.status_code == 401
+
+
+# ---------------------------------------------------------------------------
+# stats
+# ---------------------------------------------------------------------------
+
+
+def test_get_my_stats_via_me(client: TestClient, monkeypatch) -> None:
+    info = register_and_confirm(
+        client, monkeypatch,
+        f"mestats-{uuid.uuid4().hex[:8]}@example.com",
+        "mestatsp", "me 统计",
+    )
+    client.post(
+        "/api/v1/clouds",
+        headers=bearer(info["token"]),
+        files={"image": ("cloud.png", png_bytes(), "image/png")},
+        data={"cloud_type_id": "1", "is_hidden": "false"},
+    )
+    r = client.get("/api/v1/profiles/me/stats", headers=bearer(info["token"]))
+    assert r.status_code == 200
+    body = r.json()
+    assert body["cloud_count"] == 1
+    assert body["username"] == "me 统计"
+    assert body["user_id"] == info["user_id"]
+
+
+def test_get_my_stats_via_user_id(client: TestClient, monkeypatch) -> None:
+    info = register_and_confirm(
+        client, monkeypatch,
+        f"selfstats-{uuid.uuid4().hex[:8]}@example.com",
+        "selfstat", "自统计",
+    )
+    r = client.get(
+        f"/api/v1/profiles/{info['user_id']}/stats",
+        headers=bearer(info["token"]),
+    )
+    assert r.status_code == 200
+    assert r.json()["user_id"] == info["user_id"]
+
+
 def test_get_user_stats_admin(client: TestClient, monkeypatch) -> None:
     info = register_and_confirm(
         client, monkeypatch,
         f"stats-{uuid.uuid4().hex[:8]}@example.com",
         "statspass", "统计用户",
     )
-    # upload one cloud and approve it, upload another pending
     ur1 = client.post(
         "/api/v1/clouds",
         headers=bearer(info["token"]),
@@ -176,15 +255,20 @@ def test_get_user_stats_admin(client: TestClient, monkeypatch) -> None:
     assert body["username"] == "统计用户"
 
 
-def test_get_user_stats_non_admin(client: TestClient, monkeypatch) -> None:
+def test_get_user_stats_other_user_forbidden(client: TestClient, monkeypatch) -> None:
-    info = register_and_confirm(
+    info_a = register_and_confirm(
         client, monkeypatch,
-        f"stats-na-{uuid.uuid4().hex[:8]}@example.com",
+        f"stats-a-{uuid.uuid4().hex[:8]}@example.com",
-        "statsnap", "非管理员统计",
+        "statsapp1", "A 的统计",
+    )
+    info_b = register_and_confirm(
+        client, monkeypatch,
+        f"stats-b-{uuid.uuid4().hex[:8]}@example.com",
+        "statsbpp2", "B 的统计",
     )
     r = client.get(
-        f"/api/v1/profiles/{info['user_id']}/stats",
+        f"/api/v1/profiles/{info_a['user_id']}/stats",
-        headers=bearer(info["token"]),
+        headers=bearer(info_b["token"]),
     )
     assert r.status_code == 403