connect / routes /onboarding.py
habulaj's picture
Update routes/onboarding.py
7bdb090 verified
raw
history blame
14.4 kB
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")