Spaces:
Running
Running
| """ | |
| Admin Routes — Protected by admin role verification | |
| ────────────────────────────────────────────────────────────────────────────── | |
| All admin actions use Firebase Admin SDK (bypasses Firestore rules). | |
| The requesting user must have role: 'admin' in their Firestore document. | |
| Routes: | |
| GET /api/admin/users → list all users (paginated) | |
| PUT /api/admin/users/{uid} → update locked fields (fullName, phone, role) | |
| DELETE /api/admin/users/{uid} → delete user account + Firestore data | |
| GET /api/admin/stats → platform overview stats | |
| GET /api/admin/audit → audit trail of manual admin actions | |
| GET /api/admin/niche-requests → niches people couldn't find, most requested first | |
| PUT /api/admin/niche-requests/{id} → mark one added / dismissed / open | |
| """ | |
| from fastapi import APIRouter, HTTPException, Depends | |
| from pydantic import BaseModel | |
| from firebase_admin import auth as fb_auth, firestore | |
| from auth import verify_firebase_token | |
| from routes.account import delete_all_user_data | |
| from utils.supabase_client import get_supabase | |
| from utils import niche_requests | |
| from utils.audit import log_action | |
| router = APIRouter() | |
| def _verify_admin(user: dict) -> str: | |
| """Verifies the requesting user has role: 'admin'. Returns their UID.""" | |
| uid = user.get("uid") | |
| if not uid: | |
| raise HTTPException(status_code=401, detail="Unauthorized.") | |
| db = firestore.client() | |
| doc = db.collection("user_data").document(uid).get() | |
| role = doc.to_dict().get("role", "user") if doc.exists else "user" | |
| if role != "admin": | |
| raise HTTPException(status_code=403, detail="Admin access required.") | |
| return uid | |
| class UpdateUserRequest(BaseModel): | |
| fullName: str | None = None | |
| phone: str | None = None | |
| role: str | None = None # only 'user' or 'admin' | |
| def list_users( | |
| limit: int = 50, | |
| user: dict = Depends(verify_firebase_token), | |
| ): | |
| _verify_admin(user) | |
| db = firestore.client() | |
| docs = db.collection("user_data").limit(limit).stream() | |
| users = [] | |
| for doc in docs: | |
| d = doc.to_dict() | |
| users.append({ | |
| "uid": doc.id, | |
| "fullName": d.get("fullName", ""), | |
| "email": d.get("email", ""), | |
| "phone": d.get("phone", ""), | |
| "role": d.get("role", "user"), | |
| "xpTotal": d.get("xpTotal", 0), | |
| "createdAt": str(d.get("createdAt", "")), | |
| }) | |
| return {"users": users, "count": len(users)} | |
| def update_user( | |
| target_uid: str, | |
| body: UpdateUserRequest, | |
| user: dict = Depends(verify_firebase_token), | |
| ): | |
| _verify_admin(user) | |
| if body.role and body.role not in ("user", "admin"): | |
| raise HTTPException(status_code=400, detail="role must be 'user' or 'admin'") | |
| db = firestore.client() | |
| updates = {} | |
| if body.fullName is not None: updates["fullName"] = body.fullName | |
| if body.phone is not None: updates["phone"] = body.phone | |
| if body.role is not None: updates["role"] = body.role | |
| if not updates: | |
| raise HTTPException(status_code=400, detail="Nothing to update.") | |
| db.collection("user_data").document(target_uid).update(updates) | |
| # Sync display name to Firebase Auth if fullName changed | |
| if body.fullName: | |
| try: | |
| fb_auth.update_user(target_uid, display_name=body.fullName) | |
| except Exception: | |
| pass # non-critical | |
| return {"success": True, "updated": list(updates.keys())} | |
| def delete_user( | |
| target_uid: str, | |
| user: dict = Depends(verify_firebase_token), | |
| ): | |
| admin_uid = _verify_admin(user) | |
| if target_uid == admin_uid: | |
| # Self-deletion as an admin goes through DELETE /api/account/me instead, | |
| # same as any other user — this endpoint is for deleting *other* accounts. | |
| raise HTTPException(status_code=400, detail="Admins cannot delete their own account.") | |
| delete_all_user_data(target_uid) | |
| # Delete Firebase Auth account | |
| try: | |
| fb_auth.delete_user(target_uid) | |
| except fb_auth.UserNotFoundError: | |
| pass # already deleted | |
| return {"success": True, "deleted": target_uid} | |
| def audit_trail(kind: str | None = None, limit: int = 200, user: dict = Depends(verify_firebase_token)): | |
| """Every manual admin action (review decisions, source toggles, run starts), | |
| newest first — see backend/sql/admin_audit_schema.sql.""" | |
| _verify_admin(user) | |
| db = get_supabase() | |
| query = db.table("admin_audit_log").select("*").order("at", desc=True).limit(limit) | |
| if kind: | |
| query = query.eq("kind", kind) | |
| return {"entries": query.execute().data} | |
| def platform_stats(user: dict = Depends(verify_firebase_token)): | |
| _verify_admin(user) | |
| db = firestore.client() | |
| user_count = len(list(db.collection("user_data").stream())) | |
| # Count businesses across all users (approximate - stream limit applies) | |
| return { | |
| "totalUsers": user_count, | |
| "note": "Deep counts require Cloud Functions — this shows top-level totals only." | |
| } | |
| class NicheStatusRequest(BaseModel): | |
| status: str # open | added | dismissed | |
| def list_niche_requests(status: str | None = None, user: dict = Depends(verify_firebase_token)): | |
| _verify_admin(user) | |
| return {"requests": niche_requests.list_requests(status)} | |
| def update_niche_request(key: str, body: NicheStatusRequest, user: dict = Depends(verify_firebase_token)): | |
| admin_uid = _verify_admin(user) | |
| try: | |
| text = niche_requests.set_status(key, body.status) | |
| except ValueError as e: | |
| raise HTTPException(status_code=400, detail=str(e)) | |
| except KeyError: | |
| raise HTTPException(status_code=404, detail="Niche request not found.") | |
| log_action(admin_uid, "niche", body.status, text, target=key) | |
| return {"success": True} | |