204 lines
8.7 KiB
Python
204 lines
8.7 KiB
Python
import os
|
|
import httpx
|
|
import streamlit as st
|
|
|
|
BACKEND_URL = os.environ.get('BACKEND_URL', 'http://127.0.0.1:5000')
|
|
|
|
class APIClient:
|
|
def __init__(self):
|
|
self.base_url = BACKEND_URL.rstrip('/')
|
|
|
|
def _get_headers(self) -> dict:
|
|
headers = {'Content-Type': 'application/json'}
|
|
token = st.session_state.get('access_token')
|
|
if token:
|
|
headers['Authorization'] = f'Bearer {token}'
|
|
return headers
|
|
|
|
def login(self, email: str, password: str) -> tuple[bool, str]:
|
|
url = f"{self.base_url}/api/v1/auth/login"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.post(url, json={'email': email, 'password': password})
|
|
data = res.json()
|
|
if res.status_code == 200:
|
|
st.session_state['access_token'] = data.get('access_token')
|
|
st.session_state['refresh_token'] = data.get('refresh_token')
|
|
st.session_state['user'] = data.get('user')
|
|
return True, "Login realizado com sucesso!"
|
|
return False, data.get('error', 'Falha ao autenticar.')
|
|
except Exception as e:
|
|
return False, f"Erro de conexão com o servidor API: {str(e)}"
|
|
|
|
def logout(self):
|
|
st.session_state.pop('access_token', None)
|
|
st.session_state.pop('refresh_token', None)
|
|
st.session_state.pop('user', None)
|
|
|
|
def change_password(self, old_password: str, new_password: str) -> tuple[bool, str]:
|
|
url = f"{self.base_url}/api/v1/auth/change-password"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.post(url, headers=self._get_headers(), json={
|
|
'old_password': old_password,
|
|
'new_password': new_password
|
|
})
|
|
data = res.json()
|
|
if res.status_code == 200:
|
|
return True, data.get('message', 'Senha alterada com sucesso.')
|
|
return False, data.get('error', 'Falha ao alterar senha.')
|
|
except Exception as e:
|
|
return False, f"Erro ao conectar com API: {str(e)}"
|
|
|
|
def scrape_preview(self, cep: str, ramo: str, max_results: int = 15) -> tuple[bool, dict]:
|
|
"""
|
|
Solicita a varredura no Maps sem salvar no banco de dados ainda.
|
|
"""
|
|
url = f"{self.base_url}/api/v1/leads/scrape-preview"
|
|
try:
|
|
with httpx.Client(timeout=60.0) as client:
|
|
res = client.post(url, headers=self._get_headers(), json={
|
|
'cep': cep,
|
|
'ramo': ramo,
|
|
'max_results': max_results
|
|
})
|
|
data = res.json()
|
|
if res.status_code == 200:
|
|
return True, data
|
|
return False, {'error': data.get('error', 'Erro durante prospecção no Google Maps.')}
|
|
except Exception as e:
|
|
return False, {'error': f"Erro ao conectar com API: {str(e)}"}
|
|
|
|
def import_selected_leads(self, cep: str, ramo: str, selected_leads: list) -> tuple[bool, str]:
|
|
"""
|
|
Envia a lista de leads selecionados para serem salvos no banco.
|
|
"""
|
|
url = f"{self.base_url}/api/v1/leads/import-selected"
|
|
try:
|
|
with httpx.Client(timeout=30.0) as client:
|
|
res = client.post(url, headers=self._get_headers(), json={
|
|
'cep': cep,
|
|
'ramo': ramo,
|
|
'selected_leads': selected_leads
|
|
})
|
|
data = res.json()
|
|
if res.status_code == 200:
|
|
return True, data.get('message', 'Leads importados com sucesso!')
|
|
return False, data.get('error', 'Falha ao importar leads.')
|
|
except Exception as e:
|
|
return False, f"Erro ao conectar com API: {str(e)}"
|
|
|
|
def get_leads(self, status_funil: str = None, cidade: str = None, ramo: str = None, search: str = None) -> list:
|
|
url = f"{self.base_url}/api/v1/leads"
|
|
params = {}
|
|
if status_funil: params['status_funil'] = status_funil
|
|
if cidade: params['cidade'] = cidade
|
|
if ramo: params['ramo'] = ramo
|
|
if search: params['search'] = search
|
|
params['per_page'] = 200
|
|
|
|
try:
|
|
with httpx.Client(timeout=15.0) as client:
|
|
res = client.get(url, headers=self._get_headers(), params=params)
|
|
if res.status_code == 200:
|
|
return res.json().get('leads', [])
|
|
return []
|
|
except Exception:
|
|
return []
|
|
|
|
def get_lead_details(self, lead_id: str) -> dict:
|
|
url = f"{self.base_url}/api/v1/leads/{lead_id}"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.get(url, headers=self._get_headers())
|
|
if res.status_code == 200:
|
|
return res.json()
|
|
return {}
|
|
except Exception:
|
|
return {}
|
|
|
|
def update_lead(self, lead_id: str, payload: dict) -> tuple[bool, str]:
|
|
url = f"{self.base_url}/api/v1/leads/{lead_id}"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.put(url, headers=self._get_headers(), json=payload)
|
|
if res.status_code == 200:
|
|
return True, "Lead atualizado."
|
|
return False, res.json().get('error', 'Falha ao atualizar lead.')
|
|
except Exception as e:
|
|
return False, f"Erro ao comunicar com servidor: {str(e)}"
|
|
|
|
def opt_out_lead(self, lead_id: str) -> tuple[bool, str]:
|
|
url = f"{self.base_url}/api/v1/leads/{lead_id}/opt-out"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.post(url, headers=self._get_headers())
|
|
if res.status_code == 200:
|
|
return True, res.json().get('message', 'Opt-out registrado com sucesso.')
|
|
return False, res.json().get('error', 'Erro ao processar opt-out.')
|
|
except Exception as e:
|
|
return False, f"Erro: {str(e)}"
|
|
|
|
def delete_lead(self, lead_id: str) -> tuple[bool, str]:
|
|
url = f"{self.base_url}/api/v1/leads/{lead_id}"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.delete(url, headers=self._get_headers())
|
|
data = res.json()
|
|
if res.status_code == 200:
|
|
return True, data.get('message', 'Lead excluído com sucesso.')
|
|
return False, data.get('error', 'Erro ao excluir lead.')
|
|
except Exception as e:
|
|
return False, f"Erro ao comunicar com o servidor: {str(e)}"
|
|
|
|
def get_users(self) -> list:
|
|
url = f"{self.base_url}/api/v1/users"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.get(url, headers=self._get_headers())
|
|
if res.status_code == 200:
|
|
return res.json()
|
|
return []
|
|
except Exception:
|
|
return []
|
|
|
|
def create_user(self, nome: str, email: str, password: str, role: str) -> tuple[bool, str]:
|
|
url = f"{self.base_url}/api/v1/users"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.post(url, headers=self._get_headers(), json={
|
|
'nome': nome,
|
|
'email': email,
|
|
'password': password,
|
|
'role': role
|
|
})
|
|
data = res.json()
|
|
if res.status_code == 201:
|
|
return True, f"Usuário {email} criado com sucesso!"
|
|
return False, data.get('error', 'Erro ao criar usuário.')
|
|
except Exception as e:
|
|
return False, f"Erro ao conectar com API: {str(e)}"
|
|
|
|
def delete_user(self, user_id: str) -> tuple[bool, str]:
|
|
url = f"{self.base_url}/api/v1/users/{user_id}"
|
|
try:
|
|
with httpx.Client(timeout=10.0) as client:
|
|
res = client.delete(url, headers=self._get_headers())
|
|
data = res.json()
|
|
if res.status_code == 200:
|
|
return True, data.get('message', 'Usuário excluído.')
|
|
return False, data.get('error', 'Erro ao excluir usuário.')
|
|
except Exception as e:
|
|
return False, f"Erro ao conectar com API: {str(e)}"
|
|
|
|
def export_leads_data(self, format_type: str = 'csv') -> bytes:
|
|
url = f"{self.base_url}/api/v1/leads/export?format={format_type}"
|
|
try:
|
|
with httpx.Client(timeout=15.0) as client:
|
|
res = client.get(url, headers=self._get_headers())
|
|
if res.status_code == 200:
|
|
return res.content
|
|
return b""
|
|
except Exception:
|
|
return b""
|