import os import logging import aiohttp from pydantic import BaseModel from fastapi import APIRouter, HTTPException, Body, Query, Header from typing import List, Dict, Any, Optional, Union router = APIRouter() class UpdateOnboardingQuestion(BaseModel): id: int title: Optional[str] = None question_type: Optional[str] = None optional: Optional[bool] = None options: Optional[List[str]] = None class CreateOnboardingQuestion(BaseModel): title: Optional[str] = None question_type: Optional[str] = None optional: Optional[bool] = None options: Optional[Union[List[str], str, dict]] = None target_type: str # Supabase configs SUPABASE_URL = "https://ussxqnifefkgkaumjann.supabase.co" SUPABASE_KEY = os.getenv("SUPA_KEY") SUPABASE_ROLE_KEY = os.getenv("SUPA_SERVICE_KEY") if not SUPABASE_KEY or not SUPABASE_ROLE_KEY: raise ValueError("❌ SUPA_KEY ou SUPA_SERVICE_KEY não foram definidos no ambiente!") SUPABASE_HEADERS = { "apikey": SUPABASE_KEY, "Authorization": f"Bearer {SUPABASE_KEY}", "Content-Type": "application/json" } SUPABASE_ROLE_HEADERS = { "apikey": SUPABASE_ROLE_KEY, "Authorization": f"Bearer {SUPABASE_ROLE_KEY}", "Content-Type": "application/json" } # Logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Função para verificar token e permissões async def verify_token_with_permissions(user_token: str, required_permission: Optional[str] = None) -> Dict[str, Any]: """Verifica o token e retorna ID do usuário e suas permissões""" headers = { "Authorization": f"Bearer {user_token}", "apikey": SUPABASE_KEY, "Content-Type": "application/json" } async with aiohttp.ClientSession() as session: 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") # Obter permissões do usuário user_data_url = f"{SUPABASE_URL}/rest/v1/User?id=eq.{user_id}&select=is_admin,edit_onboarding" async with aiohttp.ClientSession() as session: async with session.get(user_data_url, headers=SUPABASE_HEADERS) as response: if response.status != 200 or not await response.json(): raise HTTPException(status_code=403, detail="Acesso negado: não foi possível verificar permissões") user_info = (await response.json())[0] is_admin = user_info.get("is_admin", False) if not is_admin: raise HTTPException(status_code=403, detail="Acesso negado: privilégios de administrador necessários") # Verificar permissão específica, se requisitada if required_permission: has_permission = user_info.get(required_permission, False) if not has_permission: raise HTTPException( status_code=403, detail=f"Acesso negado: permissão '{required_permission}' necessária" ) return { "user_id": user_id, "is_admin": is_admin, "permissions": user_info } @router.patch("/onboarding/update-question") async def update_onboarding_question( payload: UpdateOnboardingQuestion = Body(...), user_token: str = Header(None, alias="User-key") ): """ Atualiza uma pergunta de onboarding com base no ID. Apenas os campos enviados no payload serão atualizados. Requer permissão de admin e edit_onboarding=true. Retorna a pergunta atualizada com todos os seus campos. """ try: # Verificar se o usuário é admin e tem permissão para editar onboarding await verify_token_with_permissions(user_token, "edit_onboarding") update_data = {} if payload.title is not None: update_data["title"] = payload.title if payload.question_type is not None: update_data["question_type"] = payload.question_type if payload.optional is not None: update_data["optional"] = payload.optional if payload.options is not None: update_data["options"] = payload.options if not update_data: raise HTTPException(status_code=400, detail="Nenhum campo para atualizar foi fornecido.") query_url = f"{SUPABASE_URL}/rest/v1/Onboarding?id=eq.{payload.id}" headers = SUPABASE_ROLE_HEADERS.copy() headers["Prefer"] = "return=representation" # Atualiza os dados async with aiohttp.ClientSession() as session: async with session.patch(query_url, json=update_data, headers=headers) as response: if response.status != 200: detail = await response.text() logger.error(f"❌ Erro ao atualizar pergunta: {detail}") raise HTTPException(status_code=response.status, detail="Erro ao atualizar pergunta") # Buscar os dados atualizados completos fetch_url = f"{SUPABASE_URL}/rest/v1/Onboarding?id=eq.{payload.id}&select=id,title,description,question_type,options,target_type,optional,lock" async with aiohttp.ClientSession() as session: async with session.get(fetch_url, headers=SUPABASE_HEADERS) as fetch_response: if fetch_response.status != 200: logger.error(f"❌ Erro ao buscar pergunta atualizada: {fetch_response.status}") raise HTTPException(status_code=fetch_response.status, detail="Erro ao buscar pergunta atualizada") updated_data = await fetch_response.json() if not updated_data: raise HTTPException(status_code=404, detail="Pergunta atualizada não encontrada") question = updated_data[0] formatted = { "id": question["id"], "title": question["title"], "description": question.get("description"), "question_type": question["question_type"], "options": question.get("options", []), "optional": question.get("optional", False), "lock": question.get("lock", False) } return {"message": "✅ Pergunta atualizada com sucesso!", "updated": formatted} except HTTPException as he: raise he except Exception as e: logger.error(f"❌ Erro interno ao atualizar pergunta: {str(e)}") raise HTTPException(status_code=500, detail="Erro interno do servidor") @router.post("/onboarding/add-question") async def add_onboarding_question( payload: CreateOnboardingQuestion = Body(...), user_token: str = Header(None, alias="User-key") ): """ Adiciona uma nova pergunta de onboarding. Trata casos onde `options` vem como `{}` ou string. Requer permissão de admin e edit_onboarding=true. Retorna a pergunta recém-criada com todos os seus campos. """ try: # Verificar se o usuário é admin e tem permissão para editar onboarding await verify_token_with_permissions(user_token, "edit_onboarding") # Tratamento de `options` options = payload.options if isinstance(options, dict) and not options: options = None elif isinstance(options, str): options = [options.strip().capitalize()] new_question = { "title": payload.title, "question_type": payload.question_type, "optional": payload.optional, "options": options, "target_type": payload.target_type } query_url = f"{SUPABASE_URL}/rest/v1/Onboarding" headers = SUPABASE_ROLE_HEADERS.copy() headers["Prefer"] = "return=representation" async with aiohttp.ClientSession() as session: async with session.post(query_url, json=new_question, headers=headers) as response: if response.status not in (200, 201): detail = await response.text() logger.error(f"❌ Erro ao adicionar pergunta: {detail}") raise HTTPException(status_code=response.status, detail="Erro ao adicionar pergunta") created = await response.json() if not created or not created[0].get("id"): raise HTTPException(status_code=500, detail="Erro ao recuperar ID da pergunta criada") question_id = created[0]["id"] # Buscar os dados completos da pergunta recém-criada fetch_url = f"{SUPABASE_URL}/rest/v1/Onboarding?id=eq.{question_id}&select=id,title,description,question_type,options,target_type,optional,lock" async with aiohttp.ClientSession() as session: async with session.get(fetch_url, headers=SUPABASE_HEADERS) as fetch_response: if fetch_response.status != 200: logger.error(f"❌ Erro ao buscar pergunta criada: {fetch_response.status}") raise HTTPException(status_code=fetch_response.status, detail="Erro ao buscar pergunta criada") data = await fetch_response.json() if not data: raise HTTPException(status_code=404, detail="Pergunta criada não encontrada") question = data[0] formatted = { "id": question["id"], "title": question["title"], "description": question.get("description"), "question_type": question["question_type"], "options": question.get("options", []), "target_type": question["target_type"], "optional": question.get("optional", False), "lock": question.get("lock", False) } return {"message": "✅ Pergunta adicionada com sucesso!", "created": formatted} except HTTPException as he: raise he except Exception as e: logger.error(f"❌ Erro interno ao adicionar pergunta: {str(e)}") raise HTTPException(status_code=500, detail="Erro interno do servidor") @router.delete("/onboarding/delete-question") async def delete_onboarding_question( id: int = Query(..., description="ID da pergunta a ser deletada"), user_token: str = Header(None, alias="User-key") ): """ Deleta uma pergunta de onboarding com base no ID (passado via query parameter). Requer permissão de admin e edit_onboarding=true. """ try: # Verificar se o usuário é admin e tem permissão para editar onboarding await verify_token_with_permissions(user_token, "edit_onboarding") query_url = f"{SUPABASE_URL}/rest/v1/Onboarding?id=eq.{id}" headers = SUPABASE_ROLE_HEADERS.copy() headers["Prefer"] = "return=representation" async with aiohttp.ClientSession() as session: async with session.delete(query_url, headers=headers) as response: if response.status != 200: detail = await response.text() logger.error(f"❌ Erro ao deletar pergunta: {detail}") raise HTTPException(status_code=response.status, detail="Erro ao deletar pergunta") deleted = await response.json() if not deleted: raise HTTPException(status_code=404, detail="Pergunta não encontrada ou já deletada.") return {"message": "🗑️ Pergunta deletada com sucesso!", "deleted": deleted} except HTTPException as he: raise he except Exception as e: logger.error(f"❌ Erro interno ao deletar pergunta: {str(e)}") raise HTTPException(status_code=500, detail="Erro interno do servidor") @router.get("/onboarding/questions") async def get_onboarding_questions(user_token: str = Header(None, alias="User-key")) -> Dict[str, List[Dict[str, Any]]]: """ Retorna todas as perguntas de onboarding, separadas por target_type (client e stylist). Requer permissão de admin e edit_onboarding=true. """ try: # Verificar se o usuário é admin e tem permissão para editar onboarding await verify_token_with_permissions(user_token, "edit_onboarding") query_url = f"{SUPABASE_URL}/rest/v1/Onboarding?select=id,title,description,question_type,options,target_type,optional,lock&order=created_at.asc" headers = SUPABASE_HEADERS.copy() headers["Accept"] = "application/json" async with aiohttp.ClientSession() as session: async with session.get(query_url, headers=headers) as response: if response.status != 200: logger.error(f"❌ Erro ao buscar onboarding: {response.status}") raise HTTPException(status_code=response.status, detail="Erro ao buscar onboarding") data = await response.json() # Organizar perguntas por tipo de usuário client_questions = [] stylist_questions = [] for question in data: formatted = { "id": question["id"], "title": question["title"], "description": question.get("description"), "question_type": question["question_type"], "options": question.get("options", []), "optional": question.get("optional", False), "lock": question.get("lock", False) } if question["target_type"] == "client": client_questions.append(formatted) elif question["target_type"] == "stylist": stylist_questions.append(formatted) return { "client_questions": client_questions, "stylist_questions": stylist_questions } except HTTPException as he: raise he except Exception as e: logger.error(f"❌ Erro ao obter perguntas de onboarding: {str(e)}") raise HTTPException(status_code=500, detail="Erro interno do servidor")