Create users.py

routes/users.py

@@ -0,0 +1,120 @@
+import os
+import logging
+import asyncio
+import aiohttp
+from fastapi import APIRouter, HTTPException, Header, Depends
+from functools import lru_cache
+from typing import List, Dict, Any
+
+router = APIRouter()
+
+# Configuração do Supabase
+SUPABASE_URL = "https://ussxqnifefkgkaumjann.supabase.co"
+SUPABASE_KEY = os.getenv("SUPA_KEY")
+
+if not SUPABASE_KEY:
+    raise ValueError("❌ SUPA_KEY não foi definido no ambiente!")
+
+SUPABASE_HEADERS = {
+    "apikey": SUPABASE_KEY,
+    "Authorization": f"Bearer {SUPABASE_KEY}",
+    "Content-Type": "application/json"
+}
+
+# Configuração do logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Cache para reduzir chamadas repetidas
+@lru_cache(maxsize=128)
+def get_cached_admin_status(user_id: str) -> bool:
+    """Obtém e armazena em cache se um usuário é admin"""
+    import requests
+    user_data_url = f"{SUPABASE_URL}/rest/v1/User?id=eq.{user_id}"
+    response = requests.get(user_data_url, headers=SUPABASE_HEADERS)
+    
+    if response.status_code != 200 or not response.json():
+        return False
+    
+    user_info = response.json()[0]
+    return user_info.get("is_admin", False)
+
+async def verify_admin_token(user_token: str) -> str:
+    """Verifica se o token pertence a um administrador de forma assíncrona"""
+    headers = {
+        "Authorization": f"Bearer {user_token}",
+        "apikey": SUPABASE_KEY,
+        "Content-Type": "application/json"
+    }
+    
+    async with aiohttp.ClientSession() as session:
+        # Verificar se o token é válido
+        async with session.get(f"{SUPABASE_URL}/auth/v1/user", headers=headers) as response:
+            if response.status != 200:
+                raise HTTPException(status_code=401, detail="Token inválido ou expirado")
+            
+            user_data = await response.json()
+            user_id = user_data.get("id")
+            if not user_id:
+                raise HTTPException(status_code=400, detail="ID do usuário não encontrado")
+    
+    # Usar cache para verificar se é admin
+    is_admin = await asyncio.to_thread(get_cached_admin_status, user_id)
+    
+    if not is_admin:
+        raise HTTPException(status_code=403, detail="Acesso negado: privilégios de administrador necessários")
+    
+    return user_id
+
+async def get_recent_users(limit: int = 10) -> List[Dict[str, Any]]:
+    """Obtém os usuários mais recentes da plataforma"""
+    try:
+        # URL para obter os usuários mais recentes, ordenados por data de criação (decrescente)
+        users_url = f"{SUPABASE_URL}/rest/v1/User?select=name&order=created_at.desc&limit={limit}"
+        
+        # Modificar os headers para garantir que a resposta seja tratada como UTF-8
+        headers = SUPABASE_HEADERS.copy()
+        headers["Accept"] = "application/json; charset=utf-8"
+        
+        async with aiohttp.ClientSession() as session:
+            async with session.get(users_url, headers=headers) as response:
+                if response.status != 200:
+                    logger.error(f"❌ Erro ao obter usuários recentes: {response.status}")
+                    return []
+                
+                # Forçar a decodificação como UTF-8
+                text = await response.text(encoding='utf-8')
+                import json
+                users_data = json.loads(text)
+                
+        return users_data
+        
+    except Exception as e:
+        logger.error(f"❌ Erro ao obter usuários recentes: {str(e)}")
+        return []
+
+@router.get("/admin/users/recent")
+async def get_recent_users_endpoint(
+    user_token: str = Header(None, alias="User-key")
+):
+    """
+    Endpoint para obter os 10 usuários mais recentes da plataforma
+    """
+    try:
+        # Verificar se é um administrador
+        user_id = await verify_admin_token(user_token)
+        
+        # Obter os 10 usuários mais recentes
+        recent_users = await get_recent_users(10)
+        
+        return {
+            "recent_users": recent_users,
+            "count": len(recent_users)
+        }
+        
+    except HTTPException as he:
+        raise he
+        
+    except Exception as e:
+        logger.error(f"❌ Erro ao obter usuários recentes: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))