From 53548d2788ca97f39da979838e1206a603536094 Mon Sep 17 00:00:00 2001 From: Mplan Date: Sun, 26 Jul 2026 01:37:15 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E7=AE=A1=E7=90=86=E5=91=98?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=20GET=20/profiles/{user=5Fid}/stats=EF=BC=8C?= =?UTF-8?q?=E8=BF=94=E5=9B=9E=E7=94=A8=E6=88=B7=E5=85=A8=E9=83=A8/?= =?UTF-8?q?=E5=85=AC=E5=BC=80=E4=B8=8A=E4=BC=A0=E6=95=B0=E5=92=8C=E6=94=B6?= =?UTF-8?q?=E8=97=8F=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/routers/profiles.py | 44 ++++++++++++++++++++++++++-- app/schemas.py | 10 +++++++ tests/test_profiles.py | 64 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 115 insertions(+), 3 deletions(-) diff --git a/app/routers/profiles.py b/app/routers/profiles.py index 9364e09..3345fd0 100644 --- a/app/routers/profiles.py +++ b/app/routers/profiles.py @@ -6,9 +6,9 @@ from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import joinedload -from app.deps import CurrentUser, DbSession, OptionalUser -from app.models import Cloud, Profile, User -from app.schemas import CloudOut, PageOut, ProfileOut, ProfileUpdateIn, PublicProfileOut +from app.deps import AdminUser, CurrentUser, DbSession, OptionalUser +from app.models import Cloud, Profile, User, UserCollection +from app.schemas import CloudOut, PageOut, ProfileOut, ProfileStatsOut, ProfileUpdateIn, PublicProfileOut from app.serializers import cloud_out, profile_out, public_profile_out @@ -38,6 +38,44 @@ async def update_my_profile( return profile_out(user.profile) +@router.get("/{user_id}/stats", response_model=ProfileStatsOut) +async def get_user_stats( + user_id: uuid.UUID, + _: AdminUser, + db: DbSession, +) -> ProfileStatsOut: + result = await db.execute( + select(User).options(joinedload(User.profile)).where(User.id == user_id) + ) + user = result.scalar_one_or_none() + if not user: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在") + + cloud_count = await db.scalar( + select(func.count()).select_from(Cloud).where(Cloud.user_id == user_id) + ) or 0 + 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( + select(func.count()).select_from(UserCollection).where(UserCollection.user_id == user_id) + ) or 0 + + return ProfileStatsOut( + user_id=user.id, + username=user.profile.username, + email=user.email, + created_at=user.created_at, + cloud_count=cloud_count, + public_cloud_count=public_cloud_count, + collection_count=collection_count, + ) + + @router.get("/{user_id}/clouds", response_model=PageOut[CloudOut]) async def get_profile_clouds( user_id: uuid.UUID, diff --git a/app/schemas.py b/app/schemas.py index 7d0e373..f650bbf 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -252,3 +252,13 @@ class AdminCloudVisibilityIn(BatchIdsIn): class BatchUpdateOut(BaseModel): updated: int + + +class ProfileStatsOut(BaseModel): + user_id: uuid.UUID + username: str + email: str + created_at: datetime + cloud_count: int + public_cloud_count: int + collection_count: int diff --git a/tests/test_profiles.py b/tests/test_profiles.py index 6ef1db2..e2818df 100644 --- a/tests/test_profiles.py +++ b/tests/test_profiles.py @@ -136,3 +136,67 @@ def test_get_user_clouds_as_owner(client: TestClient, monkeypatch) -> None: def test_get_nonexistent_user_clouds(client: TestClient) -> None: r = client.get(f"/api/v1/profiles/{uuid.uuid4()}/clouds") assert r.status_code == 404 + + +# --------------------------------------------------------------------------- +# stats (admin only) +# --------------------------------------------------------------------------- + + +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"]), + files={"image": ("cloud.png", png_bytes(), "image/png")}, + data={"cloud_type_id": "1", "is_hidden": "false"}, + ) + approve_cloud(client, admin_token(client), ur1.json()["cloud"]["id"]) + client.post( + "/api/v1/clouds", + headers=bearer(info["token"]), + files={"image": ("cloud.png", png_bytes(), "image/png")}, + data={"cloud_type_id": "2", "is_hidden": "false"}, + ) + + r = client.get( + f"/api/v1/profiles/{info['user_id']}/stats", + headers=bearer(admin_token(client)), + ) + assert r.status_code == 200 + body = r.json() + assert body["cloud_count"] == 2 + assert body["public_cloud_count"] == 1 + assert body["collection_count"] == 2 + assert body["username"] == "统计用户" + + +def test_get_user_stats_non_admin(client: TestClient, monkeypatch) -> None: + info = register_and_confirm( + client, monkeypatch, + f"stats-na-{uuid.uuid4().hex[:8]}@example.com", + "statsnap", "非管理员统计", + ) + r = client.get( + f"/api/v1/profiles/{info['user_id']}/stats", + headers=bearer(info["token"]), + ) + assert r.status_code == 403 + + +def test_get_user_stats_unauthenticated(client: TestClient) -> None: + r = client.get(f"/api/v1/profiles/{uuid.uuid4()}/stats") + assert r.status_code == 401 + + +def test_get_user_stats_nonexistent(client: TestClient) -> None: + r = client.get( + f"/api/v1/profiles/{uuid.uuid4()}/stats", + headers=bearer(admin_token(client)), + ) + assert r.status_code == 404