silas.brito/leadradar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: selecao individual de leads antes de salvar, dark mode global e extracao 100% real do Google Maps

Browse Source
This commit is contained in:
Silas Andrade de Brito
2026-08-26 17:33:07 -03:00
parent 934046f2cc
commit 4c8f1f8299
8 changed files with 314 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/routes/leads.py
+59 -24
View File
@@ -11,9 +11,13 @@ from app.utils.rbac import admin_required, get_current_user
leads_bp = Blueprint('leads', __name__, url_prefix='/api/v1/leads')
@leads_bp.route('/search-maps', methods=['POST'])
@leads_bp.route('/scrape-preview', methods=['POST'])
@jwt_required()
def search_maps():
def scrape_preview():
"""
Realiza a varredura no Google Maps e retorna os estabelecimentos encontrados
para visualização prévia e seleção ANTES de salvar no banco de dados.
"""
data = request.get_json() or {}
cep = data.get('cep')
ramo = data.get('ramo')
@@ -22,31 +26,66 @@ def search_maps():
if not cep or not ramo:
return jsonify({'error': 'CEP e Ramo de Atividade são campos obrigatórios.'}), 400
# 1. Consulta CEP via ViaCEP
location_info = GeocodingService.get_location_by_cep(cep)
if 'error' in location_info and not location_info.get('cidade'):
return jsonify({'error': location_info['error']}), 400
# 2. Monta query de busca e dispara Playwright Scraper
search_query = GeocodingService.build_search_query(ramo, location_info)
scraped_data = ScraperService.scrape_google_maps(search_query, max_results=max_results)
# Se Playwright não retornar dados em ambiente sem display ou se for mock em desenvolvimento
if not scraped_data:
# Tenta fallback estruturado mock para garantir funcionalidade se não houver display
scraped_data = [
{
'nome_empresa': f"{ramo.title()} {location_info.get('bairro', 'Centro').title()}",
'telefone': '(11) 98888-7777',
'endereco': f"{location_info.get('logradouro', 'Rua Principal')}, {location_info.get('bairro', 'Bairro')}",
'google_rating': 4.8,
'total_avaliacoes': 42,
'google_maps_url': 'https://maps.google.com',
'website': ''
}
]
return jsonify({
'search_query': search_query,
'location': location_info,
'count': len(scraped_data),
'results': scraped_data
}), 200
@leads_bp.route('/import-selected', methods=['POST'])
@jwt_required()
def import_selected():
"""
Persiste no banco de dados APENAS os leads selecionados pelo usuário na tela.
"""
data = request.get_json() or {}
cep = data.get('cep')
ramo = data.get('ramo')
selected_leads = data.get('selected_leads', [])
if not cep or not ramo:
return jsonify({'error': 'CEP e Ramo de Atividade são obrigatórios.'}), 400
if not selected_leads or not isinstance(selected_leads, list):
return jsonify({'error': 'Nenhum lead selecionado para importação.'}), 400
user_id = get_jwt_identity()
summary = ScraperService.process_and_persist_leads(cep, ramo, selected_leads, usuario_id=user_id)
return jsonify({
'message': f'{summary["created_count"]} novos leads importados e {summary["updated_count"]} atualizados com sucesso!',
'summary': summary
}), 200
@leads_bp.route('/search-maps', methods=['POST'])
@jwt_required()
def search_maps():
"""
Endpoint legados de busca direta com salvamento automático.
"""
data = request.get_json() or {}
cep = data.get('cep')
ramo = data.get('ramo')
max_results = int(data.get('max_results', 15))
if not cep or not ramo:
return jsonify({'error': 'CEP e Ramo de Atividade são campos obrigatórios.'}), 400
location_info = GeocodingService.get_location_by_cep(cep)
if 'error' in location_info and not location_info.get('cidade'):
return jsonify({'error': location_info['error']}), 400
search_query = GeocodingService.build_search_query(ramo, location_info)
scraped_data = ScraperService.scrape_google_maps(search_query, max_results=max_results)
# 3. Processa e salva/deduplica no banco de dados
user_id = get_jwt_identity()
summary = ScraperService.process_and_persist_leads(cep, ramo, scraped_data, usuario_id=user_id)
@@ -63,7 +102,7 @@ def list_leads():
ramo = request.args.get('ramo')
search = request.args.get('search')
page = int(request.args.get('page', 1))
per_page = int(request.args.get('per_page', 50))
per_page = int(request.args.get('per_page', 200))
query = Lead.query
@@ -197,7 +236,6 @@ def export_leads():
if format_type == 'json':
return jsonify([l.to_dict() for l in leads]), 200
# Formato CSV
output = io.StringIO()
writer = csv.writer(output, delimiter=';')
writer.writerow([
@@ -235,9 +273,6 @@ def delete_lead(lead_id):
@leads_bp.route('/bulk', methods=['DELETE'])
@admin_required()
def expunge_leads():
"""
Expurgo / Limpeza em lote do banco de dados (Exclusivo Admin).
"""
LeadInteracao.query.delete()
num_deleted = Lead.query.delete()
db.session.commit()
backend/app/services/scraper_service.py
+53 -46
View File
@@ -15,37 +15,45 @@ class ScraperService:
@staticmethod
def scrape_google_maps(search_query: str, max_results: int = 15, headless: bool = True) -> List[Dict[str, Any]]:
"""
Executa o Playwright em modo headless para buscar no Google Maps.
Retorna uma lista de dicionários com dados dos estabelecimentos encontrados.
Executa o Playwright em modo headless para buscar no Google Maps com extração 100% real de estabelecimentos.
"""
results = []
try:
from playwright.sync_api import sync_playwright
except ImportError:
logger.warning("Playwright não está instalado. Retornando lista vazia.")
logger.error("Playwright não está instalado no ambiente.")
return results
try:
with sync_playwright() as p:
browser = p.chromium.launch(
headless=headless,
args=['--no-sandbox', '--disable-setuid-sandbox', '--disable-dev-shm-usage', '--lang=pt-BR']
args=[
'--no-sandbox',
'--disable-setuid-sandbox',
'--disable-dev-shm-usage',
'--disable-blink-features=AutomationControlled',
'--lang=pt-BR'
]
)
context = browser.new_context(
viewport={'width': 1280, 'height': 800},
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
viewport={'width': 1366, 'height': 768},
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
locale='pt-BR'
)
page = context.new_page()
# Prevenir detecção do webdriver
page.add_init_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
encoded_query = search_query.replace(' ', '+')
url = f"https://www.google.com/maps/search/{encoded_query}"
logger.info(f"Navegando para: {url}")
logger.info(f"Prospecção no Maps: {url}")
page.goto(url, wait_until='networkidle', timeout=30000)
time.sleep(2)
page.goto(url, wait_until='domcontentloaded', timeout=40000)
time.sleep(3)
# Aceitar cookies se houver modal
# Fechar dialogs de consentimento de cookies do Google caso apareçam
try:
accept_btn = page.query_selector("button[aria-label*='Aceitar'], button[aria-label*='Accept']")
if accept_btn:
@@ -54,70 +62,72 @@ class ScraperService:
except Exception:
pass
# Tentar encontrar o container do feed de resultados no Google Maps
# Aguardar feed de resultados ou container principal de busca
feed_selector = "div[role='feed']"
page.wait_for_selector(feed_selector, timeout=10000)
try:
page.wait_for_selector(feed_selector, timeout=12000)
except Exception:
# Se for redirecionado direto para 1 único resultado
logger.info("Verificando se a busca caiu diretamente no painel de 1 lugar.")
# Scroll progressivo para carregar mais estabelecimentos
for _ in range(min(5, max_results // 3 + 1)):
# Scroll progressivo no feed de resultados
for scroll_step in range(min(8, (max_results // 2) + 2)):
page.evaluate(f"""
const feed = document.querySelector("{feed_selector}");
if (feed) {{
feed.scrollBy(0, 1000);
feed.scrollBy(0, 1200);
}}
""")
time.sleep(1.5)
# Selecionar os elementos de resultado
cards = page.query_selector_all("div[role='feed'] > div > div[a]")
if not cards:
cards = page.query_selector_all("a[href*='/maps/place/']")
time.sleep(1.2)
# Extrair os links dos lugares listados
card_elements = page.query_selector_all("a[href*='/maps/place/']")
seen_urls = set()
for card in cards:
for card in card_elements:
if len(results) >= max_results:
break
try:
# Link do Maps
href = card.get_attribute("href")
if not href or href in seen_urls:
continue
seen_urls.add(href)
# aria-label geralmente contem o nome da empresa
# Extrair o nome da empresa via aria-label ou texto interno
aria_label = card.get_attribute("aria-label") or ""
# Extração de texto dentro do card
card_text = card.inner_text()
card_text = card.inner_text() or ""
lines = [l.strip() for l in card_text.split('\n') if l.strip()]
nome_empresa = aria_label.strip() if aria_label else (lines[0] if lines else "Estabelecimento Comercial")
# Extrair rating e contagem de avaliações
nome_empresa = aria_label.strip() if aria_label else (lines[0] if lines else "")
if not nome_empresa:
continue
# Extrair avaliação e total de reviews
rating = 0.0
total_avaliacoes = 0
# Exemplo de regex para extrair "4,8 (120)" ou "4.8 (120)"
rating_match = re.search(r'([1-5][.,]\d)\s*\(([\d.,]+)\)', card_text)
# Procura regex por padroes como "4,8 (150)" ou "4.8"
rating_match = re.search(r'([1-5][.,]\d)\s*(?:\(([\d.,]+)\))?', card_text)
if rating_match:
try:
rating = float(rating_match.group(1).replace(',', '.'))
total_avaliacoes = int(re.sub(r'\D', '', rating_match.group(2)))
if rating_match.group(2):
total_avaliacoes = int(re.sub(r'\D', '', rating_match.group(2)))
except ValueError:
pass
# Extrair telefone
# Extrair número de telefone do card text se presente
phone_match = re.search(r'(\(?\d{2}\)?\s*9?\d{4}[-.\s]?\d{4})', card_text)
telefone_raw = phone_match.group(1) if phone_match else ""
# Extrair endereço ou trecho do card
# Extrair trecho de endereço ou bairro das linhas
endereco = ""
for line in lines:
if any(term in line.lower() for term in ['r.', 'rua', 'av.', 'avenida', 'alameda', 'bairro', 'praça', 'nº']):
endereco = line
break
if any(term in line.lower() for term in ['r.', 'rua', 'av.', 'avenida', 'alameda', 'bairro', 'praça', 'nº', 'dr.', 'doutor', 'centro']):
if line != nome_empresa:
endereco = line
break
results.append({
'nome_empresa': nome_empresa,
@@ -129,12 +139,12 @@ class ScraperService:
'website': ''
})
except Exception as item_err:
logger.error(f"Erro ao extrair item do Maps: {item_err}")
logger.error(f"Erro ao processar item do Maps: {item_err}")
continue
browser.close()
except Exception as err:
logger.error(f"Erro na execução do Playwright Scraper: {err}")
logger.error(f"Erro na execução do Playwright Scraper real: {err}")
return results
@@ -145,7 +155,6 @@ class ScraperService:
"""
location = GeocodingService.get_location_by_cep(cep)
if 'error' in location:
# Fallback se o CEP não retornar endereço detalhado
location = {
'cep': sanitize_cep(cep),
'logradouro': '',
@@ -168,7 +177,7 @@ class ScraperService:
tel_sanitizado = sanitize_phone(tel_raw)
cidade = location.get('cidade') or item.get('cidade') or ''
# Checar regra de deduplicação: (nome_empresa, telefone_sanitizado, cidade)
# Deduplicação: (nome_empresa, telefone_sanitizado, cidade)
query = Lead.query.filter(
Lead.nome_empresa.ilike(nome_empresa),
Lead.cidade.ilike(cidade)
@@ -179,7 +188,6 @@ class ScraperService:
existing_lead = query.first()
if existing_lead:
# Atualiza dados sem resetar status_funil
if tel_raw and not existing_lead.telefone:
existing_lead.telefone = tel_raw
existing_lead.telefone_sanitizado = tel_sanitizado
@@ -203,7 +211,6 @@ class ScraperService:
updated_count += 1
persisted_leads.append(existing_lead)
else:
# Novo Lead
new_lead = Lead(
nome_empresa=nome_empresa,
ramo_atividade=ramo,
@@ -221,7 +228,7 @@ class ScraperService:
google_maps_url=item.get('google_maps_url', ''),
status_funil='novo',
tags=[ramo],
notas=f"Capturado via Radar de Busca em {cep}.",
notas=f"Capturado no Google Maps em {cep}.",
usuario_responsavel_id=usuario_id
)
db.session.add(new_lead)
backend/tests/test_leads.py
+60 -31
View File
@@ -1,37 +1,75 @@
from unittest.mock import patch
def test_search_maps_endpoint(client, user_headers):
with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo:
mock_geo.return_value = {
'cep': '01310100',
'logradouro': 'Avenida Paulista',
'bairro': 'Bela Vista',
'cidade': 'São Paulo',
'uf': 'SP',
'formatted_cep': '01310-100'
}
def test_scrape_preview_endpoint(client, user_headers):
with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo, \
patch('app.services.scraper_service.ScraperService.scrape_google_maps') as mock_scrape:
response = client.post('/api/v1/leads/search-maps', headers=user_headers, json={
mock_geo.return_value = {
'cep': '01310100', 'logradouro': 'Avenida Paulista', 'bairro': 'Bela Vista',
'cidade': 'São Paulo', 'uf': 'SP', 'formatted_cep': '01310-100'
}
mock_scrape.return_value = [
{
'nome_empresa': 'Padaria Paulista Real',
'telefone': '(11) 98888-7777',
'endereco': 'Av. Paulista, 1000',
'google_rating': 4.7,
'total_avaliacoes': 120,
'google_maps_url': 'https://maps.google.com'
}
]
response = client.post('/api/v1/leads/scrape-preview', headers=user_headers, json={
'cep': '01310-100',
'ramo': 'Padaria',
'max_results': 5
})
assert response.status_code == 200
data = response.get_json()
assert 'summary' in data
assert data['summary']['created_count'] >= 1
assert data['count'] == 1
assert data['results'][0]['nome_empresa'] == 'Padaria Paulista Real'
def test_import_selected_leads_endpoint(client, user_headers):
selected_items = [
{
'nome_empresa': 'Padaria Paulista Real',
'telefone': '(11) 98888-7777',
'endereco': 'Av. Paulista, 1000',
'google_rating': 4.7,
'total_avaliacoes': 120,
'google_maps_url': 'https://maps.google.com'
}
]
def test_update_lead_status_and_interaction(client, user_headers):
# Primeiro dispara busca para criar lead
with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo:
mock_geo.return_value = {
'cep': '01310100', 'logradouro': 'Avenida Paulista', 'bairro': 'Bela Vista',
'cidade': 'São Paulo', 'uf': 'SP', 'formatted_cep': '01310-100'
}
client.post('/api/v1/leads/search-maps', headers=user_headers, json={'cep': '01310-100', 'ramo': 'Padaria'})
response = client.post('/api/v1/leads/import-selected', headers=user_headers, json={
'cep': '01310-100',
'ramo': 'Padaria',
'selected_leads': selected_items
})
assert response.status_code == 200
data = response.get_json()
assert data['summary']['created_count'] == 1
def test_update_lead_status_and_interaction(client, user_headers):
# Criar lead via import-selected
with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo:
mock_geo.return_value = {
'cep': '01310100', 'logradouro': 'Avenida Paulista', 'bairro': 'Bela Vista',
'cidade': 'São Paulo', 'uf': 'SP', 'formatted_cep': '01310-100'
}
client.post('/api/v1/leads/import-selected', headers=user_headers, json={
'cep': '01310-100', 'ramo': 'Padaria',
'selected_leads': [{'nome_empresa': 'Padaria Teste', 'telefone': '(11) 98888-7777'}]
})
# Listar leads
list_res = client.get('/api/v1/leads', headers=user_headers)
assert list_res.status_code == 200
leads = list_res.get_json()['leads']
@@ -46,38 +84,29 @@ def test_update_lead_status_and_interaction(client, user_headers):
assert update_res.status_code == 200
assert update_res.get_json()['status_funil'] == 'contatado'
# Verificar historico de interacoes
detail_res = client.get(f'/api/v1/leads/{lead_id}', headers=user_headers)
assert detail_res.status_code == 200
detail = detail_res.get_json()
assert len(detail['interacoes']) >= 2
types = [i['tipo'] for i in detail['interacoes']]
assert 'status_change' in types
def test_opt_out_lgpd(client, user_headers):
# Criar lead
with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo:
mock_geo.return_value = {
'cep': '01310100', 'logradouro': 'Rua Teste', 'bairro': 'Centro',
'cidade': 'São Paulo', 'uf': 'SP', 'formatted_cep': '01310-100'
}
client.post('/api/v1/leads/search-maps', headers=user_headers, json={'cep': '01310-100', 'ramo': 'Farmácia'})
client.post('/api/v1/leads/import-selected', headers=user_headers, json={
'cep': '01310-100', 'ramo': 'Farmácia',
'selected_leads': [{'nome_empresa': 'Farmácia Teste', 'telefone': '(11) 97777-6666'}]
})
list_res = client.get('/api/v1/leads', headers=user_headers)
lead_id = list_res.get_json()['leads'][0]['id']
# Opt-out
opt_res = client.post(f'/api/v1/leads/{lead_id}/opt-out', headers=user_headers)
assert opt_res.status_code == 200
assert opt_res.get_json()['lead']['status_funil'] == 'opt_out'
def test_export_leads(client, user_headers):
# Exportar CSV
csv_res = client.get('/api/v1/leads/export?format=csv', headers=user_headers)
assert csv_res.status_code == 200
assert 'text/csv' in csv_res.content_type
# Exportar JSON
json_res = client.get('/api/v1/leads/export?format=json', headers=user_headers)
assert json_res.status_code == 200
assert isinstance(json_res.get_json(), list)
backend/tests/test_webhooks.py
+6 -2
View File
@@ -1,15 +1,19 @@
from unittest.mock import patch
def test_n8n_webhook_flow(client, user_headers):
# Criar lead
# Criar lead via import-selected
with patch('app.services.geocoding_service.GeocodingService.get_location_by_cep') as mock_geo:
mock_geo.return_value = {
'cep': '01310100', 'logradouro': 'Rua Teste', 'bairro': 'Centro',
'cidade': 'São Paulo', 'uf': 'SP', 'formatted_cep': '01310-100'
}
client.post('/api/v1/leads/search-maps', headers=user_headers, json={'cep': '01310-100', 'ramo': 'Supermercado'})
client.post('/api/v1/leads/import-selected', headers=user_headers, json={
'cep': '01310-100', 'ramo': 'Supermercado',
'selected_leads': [{'nome_empresa': 'Supermercado Teste', 'telefone': '(11) 99999-0000'}]
})
leads = client.get('/api/v1/leads', headers=user_headers).get_json()['leads']
assert len(leads) > 0
lead_id = leads[0]['id']
# Disparar webhook n8n