345 lines
16 KiB
Python
345 lines
16 KiB
Python
import logging
|
|
import re
|
|
import time
|
|
from typing import List, Dict, Any
|
|
from app.extensions import db
|
|
from app.models.lead import Lead
|
|
from app.models.lead_interacao import LeadInteracao
|
|
from app.utils.sanitizers import sanitize_phone, sanitize_cep
|
|
from app.services.geocoding_service import GeocodingService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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.
|
|
Clica em cada estabelecimento para abrir o painel lateral de detalhes
|
|
e extrair dados completos: Telefone, Endereço completo, Website e Rating.
|
|
"""
|
|
results = []
|
|
try:
|
|
from playwright.sync_api import sync_playwright
|
|
except ImportError:
|
|
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',
|
|
'--disable-blink-features=AutomationControlled',
|
|
'--lang=pt-BR'
|
|
]
|
|
)
|
|
context = browser.new_context(
|
|
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()
|
|
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"Iniciando varredura detalhada no Maps: {url}")
|
|
|
|
page.goto(url, wait_until='domcontentloaded', timeout=40000)
|
|
time.sleep(3)
|
|
|
|
# Aceitar cookies se houver modal
|
|
try:
|
|
accept_btn = page.query_selector("button[aria-label*='Aceitar'], button[aria-label*='Accept']")
|
|
if accept_btn:
|
|
accept_btn.click()
|
|
time.sleep(1)
|
|
except Exception:
|
|
pass
|
|
|
|
feed_selector = "div[role='feed']"
|
|
has_feed = True
|
|
try:
|
|
page.wait_for_selector(feed_selector, timeout=8000)
|
|
except Exception:
|
|
has_feed = False
|
|
logger.info("Feed de múltiplos resultados não localizado. Verificando painel de resultado único.")
|
|
|
|
# Se houver feed de múltiplos resultados, fazer scroll progressivo
|
|
if has_feed:
|
|
for _ in range(min(6, (max_results // 3) + 2)):
|
|
page.evaluate(f"""
|
|
const feed = document.querySelector("{feed_selector}");
|
|
if (feed) {{
|
|
feed.scrollBy(0, 1000);
|
|
}}
|
|
""")
|
|
time.sleep(1.0)
|
|
|
|
card_elements = page.query_selector_all("a[href*='/maps/place/']")
|
|
seen_urls = set()
|
|
|
|
for card in card_elements:
|
|
if len(results) >= max_results:
|
|
break
|
|
|
|
try:
|
|
href = card.get_attribute("href")
|
|
if not href or href in seen_urls:
|
|
continue
|
|
seen_urls.add(href)
|
|
|
|
# Clicar no card para abrir o painel lateral de detalhes no Maps
|
|
try:
|
|
card.click()
|
|
time.sleep(1.2)
|
|
except Exception:
|
|
pass
|
|
|
|
detail_panel = page.query_selector("div[role='main']") or page.query_selector("div.m6QEbd")
|
|
detail_text = detail_panel.inner_text() if detail_panel else card.inner_text()
|
|
|
|
# 1. Nome da Empresa (h1 do painel ou aria-label do card)
|
|
h1_elem = page.query_selector("h1")
|
|
nome_empresa = ""
|
|
if h1_elem:
|
|
nome_empresa = h1_elem.inner_text().strip()
|
|
if not nome_empresa:
|
|
nome_empresa = (card.get_attribute("aria-label") or "").strip()
|
|
if not nome_empresa:
|
|
lines = [l.strip() for l in card.inner_text().split('\n') if l.strip()]
|
|
nome_empresa = lines[0] if lines else ""
|
|
|
|
if not nome_empresa:
|
|
continue
|
|
|
|
# 2. Telefone detalhado
|
|
telefone_raw = ""
|
|
phone_btn = page.query_selector("button[data-tooltip*='telefone'], button[data-item-id*='phone'], [aria-label*='Telefone'], [aria-label*='Ligar']")
|
|
if phone_btn:
|
|
phone_attr = phone_btn.get_attribute("aria-label") or phone_btn.inner_text()
|
|
phone_match = re.search(r'(\(?\d{2}\)?\s*9?\d{4}[-.\s]?\d{4})', phone_attr)
|
|
if phone_match:
|
|
telefone_raw = phone_match.group(1)
|
|
|
|
if not telefone_raw:
|
|
phone_match = re.search(r'(\(?\d{2}\)?\s*9?\d{4}[-.\s]?\d{4})', detail_text)
|
|
if phone_match:
|
|
telefone_raw = phone_match.group(1)
|
|
|
|
# 3. Endereço Completo
|
|
endereco = ""
|
|
addr_btn = page.query_selector("button[data-tooltip*='endereço'], button[data-item-id*='address'], [aria-label*='Endereço']")
|
|
if addr_btn:
|
|
addr_text = addr_btn.get_attribute("aria-label") or addr_btn.inner_text()
|
|
addr_clean = re.sub(r'^[Ee]ndereço:\s*', '', addr_text).strip()
|
|
if addr_clean:
|
|
endereco = addr_clean
|
|
|
|
if not endereco:
|
|
for line in detail_text.split('\n'):
|
|
line_str = line.strip()
|
|
if any(term in line_str.lower() for term in ['r.', 'rua', 'av.', 'avenida', 'alameda', 'bairro', 'praça', 'nº', 'dr.', 'centro', 'sp', 'rj', 'mg']):
|
|
if line_str != nome_empresa and not re.search(r'^\d\.\d', line_str):
|
|
endereco = line_str
|
|
break
|
|
|
|
# 4. Website
|
|
website = ""
|
|
site_btn = page.query_selector("a[data-tooltip*='site'], a[data-item-id*='authority']")
|
|
if site_btn:
|
|
website = site_btn.get_attribute("href") or ""
|
|
|
|
# 5. Rating e Avaliações
|
|
rating = 0.0
|
|
total_avaliacoes = 0
|
|
rating_match = re.search(r'([1-5][.,]\d)\s*(?:\(([\d.,]+)\))?', detail_text)
|
|
if rating_match:
|
|
try:
|
|
rating = float(rating_match.group(1).replace(',', '.'))
|
|
if rating_match.group(2):
|
|
total_avaliacoes = int(re.sub(r'\D', '', rating_match.group(2)))
|
|
except ValueError:
|
|
pass
|
|
|
|
results.append({
|
|
'nome_empresa': nome_empresa,
|
|
'telefone': telefone_raw,
|
|
'endereco': endereco,
|
|
'google_rating': rating,
|
|
'total_avaliacoes': total_avaliacoes,
|
|
'google_maps_url': href,
|
|
'website': website
|
|
})
|
|
except Exception as item_err:
|
|
logger.error(f"Erro ao detalhar lugar do Maps: {item_err}")
|
|
continue
|
|
else:
|
|
# Se for resultado único direto
|
|
try:
|
|
h1_elem = page.query_selector("h1")
|
|
if h1_elem:
|
|
nome_empresa = h1_elem.inner_text().strip()
|
|
detail_panel = page.query_selector("div[role='main']") or page.query_selector("div.m6QEbd")
|
|
detail_text = detail_panel.inner_text() if detail_panel else ""
|
|
|
|
telefone_raw = ""
|
|
phone_btn = page.query_selector("button[data-tooltip*='telefone'], button[data-item-id*='phone'], [aria-label*='Telefone']")
|
|
if phone_btn:
|
|
phone_match = re.search(r'(\(?\d{2}\)?\s*9?\d{4}[-.\s]?\d{4})', phone_btn.get_attribute("aria-label") or phone_btn.inner_text())
|
|
if phone_match:
|
|
telefone_raw = phone_match.group(1)
|
|
|
|
endereco = ""
|
|
addr_btn = page.query_selector("button[data-tooltip*='endereço'], button[data-item-id*='address']")
|
|
if addr_btn:
|
|
endereco = re.sub(r'^[Ee]ndereço:\s*', '', addr_btn.get_attribute("aria-label") or addr_btn.inner_text()).strip()
|
|
|
|
website = ""
|
|
site_btn = page.query_selector("a[data-tooltip*='site'], a[data-item-id*='authority']")
|
|
if site_btn:
|
|
website = site_btn.get_attribute("href") or ""
|
|
|
|
rating = 0.0
|
|
total_avaliacoes = 0
|
|
rating_match = re.search(r'([1-5][.,]\d)\s*(?:\(([\d.,]+)\))?', detail_text)
|
|
if rating_match:
|
|
try:
|
|
rating = float(rating_match.group(1).replace(',', '.'))
|
|
if rating_match.group(2):
|
|
total_avaliacoes = int(re.sub(r'\D', '', rating_match.group(2)))
|
|
except ValueError:
|
|
pass
|
|
|
|
results.append({
|
|
'nome_empresa': nome_empresa,
|
|
'telefone': telefone_raw,
|
|
'endereco': endereco,
|
|
'google_rating': rating,
|
|
'total_avaliacoes': total_avaliacoes,
|
|
'google_maps_url': page.url,
|
|
'website': website
|
|
})
|
|
except Exception as single_err:
|
|
logger.error(f"Erro ao processar resultado único no Maps: {single_err}")
|
|
|
|
browser.close()
|
|
except Exception as err:
|
|
logger.error(f"Erro na execução do Playwright Scraper detalhado: {err}")
|
|
|
|
return results
|
|
|
|
@staticmethod
|
|
def process_and_persist_leads(cep: str, ramo: str, scraped_items: List[Dict[str, Any]], usuario_id: str = None) -> Dict[str, Any]:
|
|
"""
|
|
Gera leads enriquecidos a partir dos dados do geocoding + scraping, aplicando deduplicação.
|
|
"""
|
|
location = GeocodingService.get_location_by_cep(cep)
|
|
if 'error' in location:
|
|
location = {
|
|
'cep': sanitize_cep(cep),
|
|
'logradouro': '',
|
|
'bairro': '',
|
|
'cidade': 'Não especificada',
|
|
'uf': '',
|
|
'formatted_cep': cep
|
|
}
|
|
|
|
created_count = 0
|
|
updated_count = 0
|
|
persisted_leads = []
|
|
|
|
for item in scraped_items:
|
|
nome_empresa = item.get('nome_empresa', '').strip()
|
|
if not nome_empresa:
|
|
continue
|
|
|
|
tel_raw = item.get('telefone', '')
|
|
tel_sanitizado = sanitize_phone(tel_raw)
|
|
cidade = location.get('cidade') or item.get('cidade') or ''
|
|
|
|
# Deduplicação: (nome_empresa, telefone_sanitizado, cidade)
|
|
query = Lead.query.filter(
|
|
Lead.nome_empresa.ilike(nome_empresa),
|
|
Lead.cidade.ilike(cidade)
|
|
)
|
|
if tel_sanitizado:
|
|
query = query.filter(Lead.telefone_sanitizado == tel_sanitizado)
|
|
|
|
existing_lead = query.first()
|
|
|
|
if existing_lead:
|
|
if tel_raw and not existing_lead.telefone:
|
|
existing_lead.telefone = tel_raw
|
|
existing_lead.telefone_sanitizado = tel_sanitizado
|
|
if item.get('google_rating') and item['google_rating'] > 0:
|
|
existing_lead.google_rating = item['google_rating']
|
|
if item.get('total_avaliacoes'):
|
|
existing_lead.total_avaliacoes = item['total_avaliacoes']
|
|
if item.get('google_maps_url'):
|
|
existing_lead.google_maps_url = item['google_maps_url']
|
|
if item.get('website') and not existing_lead.website:
|
|
existing_lead.website = item['website']
|
|
|
|
existing_lead.atualizado_em = db.func.now()
|
|
|
|
interacao = LeadInteracao(
|
|
lead_id=existing_lead.id,
|
|
usuario_id=usuario_id,
|
|
tipo='lead_atualizado',
|
|
descricao='Lead re-encontrado em nova busca e atualizado.',
|
|
metadados={'ramo_busca': ramo, 'cep_busca': cep}
|
|
)
|
|
db.session.add(interacao)
|
|
updated_count += 1
|
|
persisted_leads.append(existing_lead)
|
|
else:
|
|
new_lead = Lead(
|
|
nome_empresa=nome_empresa,
|
|
ramo_atividade=ramo,
|
|
cep_busca=location.get('formatted_cep', cep),
|
|
logradouro=location.get('logradouro') or item.get('endereco', ''),
|
|
bairro=location.get('bairro', ''),
|
|
cidade=cidade,
|
|
uf=location.get('uf', ''),
|
|
telefone=tel_raw,
|
|
telefone_sanitizado=tel_sanitizado,
|
|
whatsapp_valido=True if tel_sanitizado else False,
|
|
website=item.get('website', ''),
|
|
google_rating=item.get('google_rating', 0.0),
|
|
total_avaliacoes=item.get('total_avaliacoes', 0),
|
|
google_maps_url=item.get('google_maps_url', ''),
|
|
status_funil='novo',
|
|
tags=[ramo],
|
|
notas=f"Capturado no Google Maps em {cep}.",
|
|
usuario_responsavel_id=usuario_id
|
|
)
|
|
db.session.add(new_lead)
|
|
db.session.flush()
|
|
|
|
interacao = LeadInteracao(
|
|
lead_id=new_lead.id,
|
|
usuario_id=usuario_id,
|
|
tipo='lead_criado',
|
|
descricao=f'Lead capturado no Maps para ramo "{ramo}" e CEP {cep}.',
|
|
metadados={'query_cep': cep, 'ramo': ramo}
|
|
)
|
|
db.session.add(interacao)
|
|
created_count += 1
|
|
persisted_leads.append(new_lead)
|
|
|
|
db.session.commit()
|
|
|
|
return {
|
|
'location': location,
|
|
'created_count': created_count,
|
|
'updated_count': updated_count,
|
|
'total_processed': len(scraped_items),
|
|
'leads': [l.to_dict() for l in persisted_leads]
|
|
}
|