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 _extract_card_rating_and_reviews(card, detail_panel=None) -> tuple: """ Extrai a nota (rating) e a quantidade total de avaliações do container exclusivo do estabelecimento, com fallback para o painel de detalhes. """ rating = 0.0 total_avaliacoes = 0 if not card: return rating, total_avaliacoes try: # 1. Obter o elemento container pai do lugar no feed (div.Nv251d ou ancestral do card link) container = card try: parent_handle = card.evaluate_handle( "el => el.closest('div.Nv251d') || el.closest('div[role=\"article\"]') || el.closest('div.THtW3d') || el.parentElement.parentElement" ) if parent_handle: elem = parent_handle.as_element() if elem: container = elem except Exception: container = card # 2. Procurar aria-labels no container do lugar aria_elems = container.query_selector_all("[aria-label*='estrela'], [aria-label*='star'], [aria-label*='avaliaç']") container_label = container.get_attribute("aria-label") or "" card_label = card.get_attribute("aria-label") or "" combined_aria = " ".join([e.get_attribute("aria-label") or "" for e in aria_elems] + [container_label, card_label]) r_match = re.search(r'([1-5][.,]\d)\s*(?:estrelas?|stars?)', combined_aria, re.IGNORECASE) if r_match: try: rating = float(r_match.group(1).replace(',', '.')) except ValueError: pass rev_match = re.search(r'([\d.,]+)\s*avaliaç', combined_aria, re.IGNORECASE) if rev_match: try: total_avaliacoes = int(re.sub(r'\D', '', rev_match.group(1))) except ValueError: pass # 3. Ler o texto bruto (inner_text) do container do lugar container_text = container.inner_text() or card.inner_text() or "" if rating == 0.0: # O rating no card do Maps vem como "4,8 (120)" ou "4.8 ★" m = re.search(r'([1-5][.,]\d)\s*(?:\(|\s*★|\s*estrelas?)', container_text, re.IGNORECASE) if not m: m = re.search(r'([1-5][.,]\d)', container_text) if m: try: val = float(m.group(1).replace(',', '.')) if 1.0 <= val <= 5.0: rating = val except ValueError: pass if total_avaliacoes == 0: # O total de avaliações no card vem entre parênteses "(120)" ou "(1.250)" m = re.search(r'\(([\d.,]+)\)', container_text) if m: try: total_avaliacoes = int(re.sub(r'\D', '', m.group(1))) except ValueError: pass else: m = re.search(r'([\d.,]+)\s*avaliaç', container_text, re.IGNORECASE) if m: try: total_avaliacoes = int(re.sub(r'\D', '', m.group(1))) except ValueError: pass # 4. Fallback para o detail_panel se o container do card não possuir a nota if (rating == 0.0 or total_avaliacoes == 0) and detail_panel: detail_text = detail_panel.inner_text() or "" if rating == 0.0: star_elem = detail_panel.query_selector("div.F7L8d span, span.ceYyJf, [aria-label*='estrelas']") if star_elem: st_txt = star_elem.get_attribute("aria-label") or star_elem.inner_text() st_m = re.search(r'([1-5][.,]\d)', st_txt) if st_m: try: rating = float(st_m.group(1).replace(',', '.')) except ValueError: pass if total_avaliacoes == 0: rev_elem = detail_panel.query_selector("button[aria-label*='avaliaç'], span[aria-label*='avaliaç']") if rev_elem: rv_txt = rev_elem.get_attribute("aria-label") or rev_elem.inner_text() rv_m = re.search(r'([\d.,]+)\s*avaliaç', rv_txt, re.IGNORECASE) if rv_m: try: cnt = int(re.sub(r'\D', '', rv_m.group(1))) if cnt < 20000: total_avaliacoes = cnt except ValueError: pass except Exception as e: logger.error(f"Erro ao extrair rating/avaliacoes do card: {e}") return rating, total_avaliacoes @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 extrai dados completos: Telefone, Endereço completo, Website e Rating individual. """ 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.") 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. Extração da Nota e Avaliações estritamente escopadas no Container do lugar rating, total_avaliacoes = ScraperService._extract_card_rating_and_reviews(card, detail_panel) # 2. Nome da Empresa (priorizar aria-label do card ou h1 dentro do detail_panel) nome_empresa = (card.get_attribute("aria-label") or "").strip() if nome_empresa.lower() in ('resultados', 'resultado', ''): nome_empresa = "" if not nome_empresa and detail_panel: h1_elem = detail_panel.query_selector("h1") if h1_elem: h_txt = h1_elem.inner_text().strip() if h_txt and h_txt.lower() not in ('resultados', 'resultado'): nome_empresa = h_txt if not nome_empresa: lines = [l.strip() for l in card.inner_text().split('\n') if l.strip()] for line in lines: if line and line.lower() not in ('resultados', 'resultado') and not re.search(r'^[1-5][.,]\d', line): nome_empresa = line break if not nome_empresa or nome_empresa.lower() in ('resultados', 'resultado'): continue # 3. 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) # 4. 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 # 5. 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 "" 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: detail_panel = page.query_selector("div[role='main']") or page.query_selector("div.m6QEbd") h1_elem = detail_panel.query_selector("h1") if detail_panel else None nome_empresa = h1_elem.inner_text().strip() if h1_elem else "" if nome_empresa.lower() in ('resultados', 'resultado', ''): title_elem = page.query_selector(".fontHeadlineLarge, .DUwif, div.qBF1Pd") nome_empresa = title_elem.inner_text().strip() if title_elem else "" if nome_empresa and nome_empresa.lower() not in ('resultados', 'resultado'): 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, total_avaliacoes = ScraperService._extract_card_rating_and_reviews(page.query_selector("h1"), detail_panel) 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] }