import os import json import random import requests import asyncio from datetime import datetime from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, CacheMode, BrowserConfig from crawl4ai.async_configs import ProxyConfig # Overwrite print to log only errors and warnings (keeps minero_log.txt clean) import builtins def print(*args, **kwargs): msg = " ".join(str(arg) for arg in args) if any(term in msg.lower() for term in ["[!]", "error", "exception", "warning", "advertencia", "fallo"]): builtins.print(*args, **kwargs) # ====================================== # CONFIGURACION MAESTRA (ATLAS EL MINADOR) # ===================================== if os.path.exists(".env"): with open(".env") as f: for line in f: if "=" in line and not line.strip().startswith("#"): key, val = line.strip().split("=", 1) os.environ[key.strip()] = val.strip().strip('"').strip("'") PROXY_URL = os.environ.get("PROXY_URL", "http://uikPBRbczMQd8rBr:tuaE95tFQQa74nbO_country-us@geo.iproyal.com:12321") N8N_WEBHOOK_URL = os.environ.get("N8N_WEBHOOK_URL", "https://n8n.orozdesign.net/webhook/crawl4ai-inbound") NOCODB_API_URL = os.environ.get("NOCODB_API_URL", "https://db.orozdesign.net/api/v1/db/data/v1/p1rc5rxqribu59g") NOCODB_TOKEN = os.environ.get("NOCODB_TOKEN", "nc_pat_Tdip5x_XqJUoRkYAe7YBYhxDv4V1m7zM5iNQsAGz") NOCODB_TABLE_CIUDADES = os.environ.get("NOCODB_TABLE_CIUDADES", "meymgp6aqhxf7b5") BASE_NICHE = "Cleaning Services" NICHES = [ "Cleaning Services", "House Cleaning", "Housekeeping" ] import re # =============================================== # PARSER DE MARKDOWN (REEMPLAZA AL LLM — COSTO $0) # =============================================== def parse_leads_from_markdown(markdown_text, city_name, state): """ Extrae leads de Google Maps del contenido markdown sin usar LLM. Busca patrones de links a Google Maps, teléfonos US y websites. """ if not markdown_text: return [] leads = [] seen_phones = set() seen_names = set() # Links de Google Maps: [Nombre del Negocio](https://www.google.com/maps/place/...) o relative (/maps/place/...) gmb_link_pattern = re.compile( r'\[([^\]]{3,80})\]\(((?:https://(?:www\.)?google\.com)?/maps/place/[^)]{10,})\)', re.IGNORECASE ) # Teléfonos US: (XXX) XXX-XXXX | XXX-XXX-XXXX | XXX.XXX.XXXX phone_pattern = re.compile( r'(?:\+1[\s\-\.]?)?\(?\b(\d{3})\)?[\s\-\.](\d{3})[\s\-\.](\d{4})\b' ) gmb_matches = list(gmb_link_pattern.finditer(markdown_text)) import html for i, match in enumerate(gmb_matches): name = html.unescape(match.group(1).strip()) gmb_url = match.group(2).strip() if gmb_url.startswith('/'): gmb_url = "https://www.google.com" + gmb_url skip_terms = {'see all', 'more', 'website', 'directions', 'open', 'closed', 'call', 'photos', 'share', 'save'} if len(name) < 3 or name.lower() in skip_terms: continue if name in seen_names: continue seen_names.add(name) # Ventana de texto alrededor del match para buscar teléfono start_pos = match.start() end_pos = gmb_matches[i + 1].start() if i + 1 < len(gmb_matches) else start_pos + 600 window = markdown_text[max(0, start_pos - 100):end_pos] phone = "" phone_match = phone_pattern.search(window) if phone_match: raw_phone = f"({phone_match.group(1)}) {phone_match.group(2)}-{phone_match.group(3)}" if raw_phone not in seen_phones: seen_phones.add(raw_phone) phone = raw_phone # Website (excluye dominios de Google) website_match = re.search( r'https?://(?!(?:www\.)?google\.com|maps\.google\.com)[^\s\)"]{5,}', window ) website = website_match.group(0) if website_match else "" leads.append({ "business_name": name, "phone": phone, "city": city_name, "state": state, "gmb_url": gmb_url, "website": website }) return leads # ======================================= # PARSER DE HTML (RESPALDO ULTRA-ROBUSTO) # ======================================= def parse_leads_from_html(html_text, city_name, state): if not html_text: return [] leads = [] seen_names = set() seen_phones = set() # Buscamos href que contengan /maps/place/ y aria-label para el nombre del negocio pattern1 = re.compile(r'href="([^"]*/maps/place/[^"]*)"[^>]*aria-label="([^"]+)"', re.IGNORECASE) pattern2 = re.compile(r'aria-label="([^"]+)"[^>]*href="([^"]*/maps/place/[^"]*)"', re.IGNORECASE) matches = [] for match in pattern1.finditer(html_text): url, name = match.group(1), match.group(2) matches.append((name, url, match.start())) for match in pattern2.finditer(html_text): name, url = match.group(1), match.group(2) matches.append((name, url, match.start())) import html for name, gmb_url, start_pos in matches: name = html.unescape(name.strip()) gmb_url = gmb_url.strip() skip_terms = {'see all', 'more', 'website', 'directions', 'open', 'closed', 'call', 'photos', 'share', 'save'} if len(name) < 3 or name.lower() in skip_terms: continue if name in seen_names: continue seen_names.add(name) if gmb_url.startswith('/'): gmb_url = "https://www.google.com" + gmb_url # Ventana de búsqueda de teléfono en el HTML alrededor del link window = html_text[max(0, start_pos - 200):start_pos + 2500] phone_pattern = re.compile(r'(?:\+1[\s\-\.]?)?\(?\b(\d{3})\)?[\s\-\.](\d{3})[\s\-\.](\d{4})\b') phone = "" phone_match = phone_pattern.search(window) if phone_match: raw_phone = f"({phone_match.group(1)}) {phone_match.group(2)}-{phone_match.group(3)}" if raw_phone not in seen_phones: seen_phones.add(raw_phone) phone = raw_phone website_match = re.search(r'https?://(?!(?:www\.)?google\.com|maps\.google\.com|schema\.org)[^\s\)"]{5,200}', window) website = website_match.group(0) if website_match else "" leads.append({ "business_name": name, "phone": phone, "city": city_name, "state": state, "gmb_url": gmb_url, "website": website }) return leads # =============================================== # FILTROS NATIVOS EN PYTHON (EFICIENCIA Y AHORRO) # =============================================== EXCLUDED_KEYWORDS = [ "dry clean", "laundry", "laundromat", "lavanderia", "tintoreria", "wash & fold", "carpet", "rug", "tapiceria", "window", "pressure wash", "power wash", "merry maids", "molly maid", "the maids", "maidpro", "jani-king", "jan-pro", "corvus", "city wide", "vanguard", "office pride" ] # Apellidos comunes, nombres y términos hispanos para filtrado local rápido HISPANIC_KEYWORDS = [ # Apellidos comunes "garcia", "lopez", "ramirez", "martinez", "rodriguez", "hernandez", "perez", "gomez", "sanchez", "flores", "torres", "diaz", "cruz", "morales", "ortiz", "gonzalez", "quintana", "rivera", "chavez", "guzman", "alvarez", "romero", "alonso", "gutierrez", "castro", "ortega", "rubio", "ruiz", "serrano", "marquez", "munoz", "mendez", "castillo", "vargas", "ramos", "aguilar", "salazar", "delgado", "soto", "pena", "medina", "herrera", "aguirre", "valdez", "mora", "mendoza", "miranda", "arias", "robles", "fuentes", "velasquez", "montero", "castellanos", "camacho", "salinas", "cardona", "paredes", "figueroa", "silva", "suarez", "trejo", "trejos", "solis", # Nombres comunes "maria", "sofia", "monica", "dani", "anabelle", "juana", "julio", "carlos", "juan", "jose", "pedro", "luis", "rosa", "elena", "norma", "carmen", "laura", "leticia", "veronica", "patricia", "claudia", "marta", "adriana", "delia", "silvia", "beatriz", "gloria", "yolanda", "isabel", "teresa", "alicia", "miguel", "jorge", "alejandro", "francisco", "manuel", "santiago", "mateo", "lucas", "diego", "javier", "alberto", "antonio", "fernando", "gabriel", "melisa", "melissa", "mimi", "ana", "diana", # Conceptos y palabras clave de negocio "limpieza", "casa", "sol", "hogar", "brillante", "estrella", "el primo", "hermanos", "familia" ] def clean_and_validate_local(business_name): """ Realiza un pre-filtrado rápido. Retorna True si pasa el filtro residencial y tiene indicadores hispanos o nombre compatible. """ if not business_name: return False, "" name_lower = business_name.lower() # 1. Filtro estricto de exclusiones for kw in EXCLUDED_KEYWORDS: if kw in name_lower: return False, f"Excluido por palabra clave '{kw}'" # 2. Búsqueda del indicador hispano indicator_found = "" for kw in HISPANIC_KEYWORDS: # Buscamos límites de palabra para evitar matches parciales raros if re.search(r'\b' + re.escape(kw) + r'\b', name_lower): indicator_found = kw break if not indicator_found: return False, "Sin indicador hispano aparente" return True, indicator_found def get_headers(): return {"xc-token": NOCODB_TOKEN, "Content-Type": "application/json"} def get_next_city(): """Busca la primera ciudad en estado Pending en NocoDB""" if not NOCODB_TABLE_CIUDADES: return {"id": 0, "city": "Miami", "state": "FL"} url = f"{NOCODB_API_URL}/{NOCODB_TABLE_CIUDADES}?where=(mining_status,eq,Pending)&limit=1" try: res = requests.get(url, headers=get_headers()) if res.status_code == 200: data = res.json() if data.get('list') and len(data['list']) > 0: return data['list'][0] else: print(f"[!] Error de NocoDB ({res.status_code}): {res.text}") except Exception as e: print(f"[!] Error conectando a NocoDB: {e}") return None def update_city_status(city_id, status): """Actualiza el estado de la ciudad en NocoDB (Processing / Completed)""" if not NOCODB_TABLE_CIUDADES or city_id == 0: return url = f"{NOCODB_API_URL}/{NOCODB_TABLE_CIUDADES}/{city_id}" now_str = datetime.utcnow().isoformat()[:19].replace('T', ' ') data = {"mining_status": status, "last_mined": now_str} if status == "Processing": data["processing_started_at"] = now_str elif status in ("Completed", "Pending"): data["processing_started_at"] = None try: res = requests.patch(url, headers=get_headers(), json=data) if res.status_code not in (200, 204): print(f"[!] Error actualizando estado en NocoDB ({res.status_code}): {res.text}") except Exception as e: print(f"[!] Excepción actualizando NocoDB: {e}") def normalize_phone_e164(phone_raw): """ Convierte cualquier formato de teléfono al estándar E.164 de NocoDB: +1XXXXXXXXXX """ if not phone_raw: return "" digits_only = re.sub(r'[^0-9]', '', str(phone_raw)) if not digits_only or len(digits_only) < 7: return "" if len(digits_only) == 11 and digits_only.startswith('1'): return f"+1{digits_only[1:]}" if len(digits_only) == 10: return f"+1{digits_only}" return f"+1{digits_only[-10:]}" def send_to_n8n(lead_data, current_category): """Envía el lead estructurado al Webhook de recepción de n8n (Floox 01). La clasificación Latino/Americano ya NO ocurre aquí — la hace n8n Floox 02.""" try: phone_e164 = normalize_phone_e164(lead_data.get("phone", "")) if not phone_e164: print(f" [SKIP] Lead '{lead_data.get('business_name')}' descartado: teléfono irreconocible.") return payload = { "business_name": lead_data.get("business_name"), "phone": phone_e164, "city": lead_data.get("city"), "state": lead_data.get("state", ""), "website": lead_data.get("website", ""), "gmb_url": lead_data.get("gmb_url", ""), "category": current_category } headers = {"x-floox-key": "OrozDesign-Atlas-Team!"} response = requests.post(N8N_WEBHOOK_URL, json=payload, headers=headers) if response.status_code == 200: print(f" [OK] Lead '{payload['business_name']}' ({phone_e164}) inyectado a n8n.") else: print(f" [FAIL] Error {response.status_code} al enviar a n8n.") except Exception as e: print(f" [ERROR] Falló la conexión con n8n: {e}") async def extract_google_maps_leads(city_data): city_name = city_data.get('city', '') state = city_data.get('state', '') for variant in NICHES: search_query = f"{variant} in {city_name}, {state}" print(f"\n[*] INICIANDO EXTRACCIÓN: {search_query}") session_id = "".join(random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=8)) iproyal_pass = os.environ.get("IPROYAL_PASS", "tuaE95tFQQa74nbO_country-us") proxy_pass = f"{iproyal_pass}_session-{session_id}_lifetime-10m" proxy_user = os.environ.get("IPROYAL_USER", "uikPBRbczMQd8rBr") proxy_uri = f"http://{proxy_user}:{proxy_pass}@geo.iproyal.com:12321" print(f"[*] Proxy Seguro Activo: IPRoyal (Sesión: {session_id})") proxy = ProxyConfig( server="http://geo.iproyal.com:12321", username=proxy_user, password=proxy_pass ) browser_config = BrowserConfig( proxy_config=proxy, headless=True, avoid_ads=True, viewport_width=1920, viewport_height=1080, extra_args=[ "--disable-web-fonts", "--disable-extensions", "--disable-notifications", "--disable-popup-blocking" ] ) delay_seconds = round(random.uniform(7.0, 12.0), 2) print(f"[*] Pausa natural anti-bot: {delay_seconds}s") js_scroll_code = """ (async () => { // Set consent cookies to bypass Google's consent dialog document.cookie = "CONSENT=YES+shp.gws-20231127-0-RC1.en+FX+025; domain=.google.com; path=/"; document.cookie = "SOCS=JD50ZXN0XzIwMjMwODEwLTA5LXB0LmVuK0ZYKzk4Ng; domain=.google.com; path=/"; const clickAccept = () => { const buttons = Array.from(document.querySelectorAll('button')); for (const btn of buttons) { const txt = btn.textContent.trim(); if (/Accept all|Accept|Accepter|Aceptar/i.test(txt)) { btn.click(); return true; } } const inputs = Array.from(document.querySelectorAll('input[type="submit"]')); for (const input of inputs) { if (/Accept/i.test(input.value)) { input.click(); return true; } } return false; }; if (clickAccept()) { await new Promise(r => setTimeout(r, 4000)); } const findFeed = () => { const feed = document.querySelector('div[role="feed"]'); if (feed) return feed; const divs = document.querySelectorAll('div'); for (const div of divs) { if (div.scrollHeight > div.clientHeight && window.getComputedStyle(div).overflowY === 'auto') { if (div.querySelector('a[href*="/maps/place/"]')) { return div; } } } return null; }; const feed = findFeed(); if (!feed) return; let lastHeight = feed.scrollHeight; let unchangedRuns = 0; for (let i = 0; i < 30; i++) { feed.scrollTop = feed.scrollHeight; feed.dispatchEvent(new Event('scroll')); await new Promise(r => setTimeout(r, 2500)); const textContent = feed.textContent || ""; if (textContent.includes("You've reached the end of the list") || textContent.includes("reached the end") || textContent.includes("Fin de la lista")) { break; } let newHeight = feed.scrollHeight; if (newHeight === lastHeight) { unchangedRuns++; if (unchangedRuns >= 5) break; } else { unchangedRuns = 0; lastHeight = newHeight; } } })() """ config_args = { "cache_mode": CacheMode.BYPASS, "magic": True, "delay_before_return_html": delay_seconds, "js_code": js_scroll_code, "word_count_threshold": 0, "excluded_tags": ["script", "style", "head"], "exclude_domains": [ "fonts.gstatic.com", "fonts.googleapis.com", "accounts.google.com", "play.google.com", "mtalk.google.com", "android.clients.google.com", "ogads-pa.clients6.google.com", "feedback-pa.clients6.google.com", "tpc.googlesyndication.com" ], "extra_headers": { "Cookie": "CONSENT=YES+shp.gws-20231127-0-RC1.en+FX+025; SOCS=JD50ZXN0XzIwMjMwODEwLTA5LXB0LmVuK0ZYKzk4Ng" } } try: config = CrawlerRunConfig(**config_args) except TypeError: config_args.pop("extra_headers", None) config = CrawlerRunConfig(**config_args) target_url = f"https://www.google.com/maps/search/{search_query.replace(' ', '+')}?hl=en&gl=us" max_retries = 3 backoff_delay = 15.0 result = None for attempt in range(1, max_retries + 1): try: async with AsyncWebCrawler(config=browser_config) as crawler: result = await crawler.arun(url=target_url, config=config) if result.success: break else: error_msg = result.error_message or "Unknown error" print(f"[!] Intento {attempt} falló para {variant}: {error_msg}") if attempt < max_retries: sleep_time = backoff_delay * (2 ** (attempt - 1)) + random.uniform(1, 3) await asyncio.sleep(sleep_time) except Exception as ex: print(f"[!] Excepción en intento {attempt} para {variant}: {ex}") if attempt < max_retries: sleep_time = backoff_delay * (2 ** (attempt - 1)) + random.uniform(1, 3) await asyncio.sleep(sleep_time) else: raise ex if not result or not result.success: print(f"[!] Error definitivo en extracción para {variant}. Omitiendo nicho.") continue try: leads = parse_leads_from_markdown(result.markdown or "", city_name, state) if not leads: print(" [*] Markdown parser returned 0 leads. Falling back to HTML parser...") leads = parse_leads_from_html(result.html or "", city_name, state) print(f"[+] Crudo: Extraídos {len(leads)} leads iniciales por parser para {variant}.") tasks = [] for lead in leads: name = lead.get("business_name") phone = lead.get("phone") # Validación básica de estructura if not name or not phone or len(str(name).strip()) < 2 or len(str(phone).strip()) < 5: continue # Filtro estricto de exclusiones en Python name_lower = name.lower() is_excluded = False exclusion_reason = "" for kw in EXCLUDED_KEYWORDS: if kw in name_lower: is_excluded = True exclusion_reason = f"Excluido por palabra clave '{kw}'" break if is_excluded: print(f" [SKIP-LOCAL] '{name}' descartado: {exclusion_reason}") continue if not lead.get("city"): lead["city"] = city_name if not lead.get("state"): lead["state"] = state tasks.append(asyncio.to_thread(send_to_n8n, lead, variant)) if tasks: await asyncio.gather(*tasks) except Exception as e: print(f"[!] Error procesando leads para {variant}:", e) def check_n8n_status(): """Verifica si el workflow de n8n está encendido (Active) con tolerancia a fallos de red.""" try: headers = {"x-floox-key": "OrozDesign-Atlas-Team!"} res = requests.post(N8N_WEBHOOK_URL, json={"ping": True}, headers=headers, timeout=30) if res.status_code == 404: print("[!] ADVERTENCIA: La URL del webhook de n8n devolvió 404. Se asume ruta inactiva pero se continúa...") return True return True except Exception as e: print(f"[!] ADVERTENCIA: No se pudo verificar el estado de n8n por problemas de red/timeout ({e}). Continuando con tolerancia activa...") return True if __name__ == "__main__": print("====================================") print(" ATLAS RECOLECTOR - INICIANDO v2.1 ") print("====================================") import sys import time if "--now" in sys.argv or "-n" in sys.argv: print("[*] Modo manual detectado. Omitiendo retraso aleatorio (jitter).") else: jitter = random.randint(0, 900) print(f"[*] Aplicando retraso aleatorio (jitter) de {jitter}s...") time.sleep(jitter) check_n8n_status() print("[+] n8n está ONLINE (o en modo Tolerancia de Red). Permiso concedido para minar.") city = get_next_city() if city: city_id = city.get('Id', city.get('id', 0)) try: update_city_status(city_id, "Processing") asyncio.run(extract_google_maps_leads(city)) update_city_status(city_id, "Completed") print("\n[+] Ciclo finalizado exitosamente.") except BaseException as e: print(f"\n[!] Error crítico: {e}") try: update_city_status(city_id, "Pending") print("[+] Estado de la ciudad restaurado a 'Pending'.") except Exception as re: print(f"[!] No se pudo restaurar el estado: {re}") raise e else: print("[*] No hay ciudades en estado 'Pending'.")