#!/usr/bin/env python3 """ Prospecção automática agendada. Lê prospeccao_schedule.json, roda scrape+envio pra CRM em cada combinação palavra-chave x cidade (na ordem: para cada cidade -> para cada palavra-chave), enriquece os leads (opcional), evita duplicatas e reporta ao Renato no Telegram via outbox do bot da Mia. Compatibilidade: mantém os campos legados "nichos" no JSON (tratados como palavras-chave). Novo campo preferido: "palavras_chave". Novos campos suportados no JSON: - palavras_chave: list[str] (alternativa a "nichos") - pais: "BR" | "US" - enriquecer: bool - crm: {"token","location_id","pipeline_id","stage_id","conta_id"} (se ausente, cai no default do config.py = Linkia) """ import json import os import re import sys import time import traceback from datetime import datetime from pathlib import Path # Garante que os módulos do diretório sejam importáveis quando o service roda BASE_DIR = Path(__file__).resolve().parent if str(BASE_DIR) not in sys.path: sys.path.insert(0, str(BASE_DIR)) from prospector import scrape_google_maps # fallback scraping Playwright CONFIG_PATH = BASE_DIR / "prospeccao_schedule.json" PROCESSED_IDS_PATH = BASE_DIR / "linkia_processed_ids.json" PROSPECTADOS_PATH = BASE_DIR / "prospectados.json" PROGRESS_PATH = BASE_DIR / "prospeccao_auto_progress.json" OUTBOX_DIR = Path("/opt/mia-bot/outbox") LOG_DIR = Path("/opt/mia/logs") LOG_DIR.mkdir(parents=True, exist_ok=True) # ----------------------------- helpers ----------------------------- def log(msg: str) -> None: stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{stamp}] {msg}", flush=True) def load_json(path: Path, default): try: if path.exists(): with path.open("r", encoding="utf-8") as f: return json.load(f) except Exception as e: log(f"WARN: falha ao ler {path}: {e}") return default def save_json(path: Path, data) -> None: try: tmp = path.with_suffix(path.suffix + ".tmp") with tmp.open("w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) tmp.replace(path) except Exception as e: log(f"ERRO ao salvar {path}: {e}") def normalize_phone(raw: str) -> str: """Extrai só dígitos do telefone (comparável entre formatos).""" if not raw: return "" return re.sub(r"\D", "", raw) def dedup_key(nome: str, cidade: str) -> str: return f"{(nome or '').strip().lower()}|{(cidade or '').strip().lower()}" def enviar_telegram(texto: str) -> None: """Cria um JSON no outbox pra o bot da Mia mandar ao Renato.""" try: OUTBOX_DIR.mkdir(parents=True, exist_ok=True) fname = OUTBOX_DIR / f"{time.time_ns()}.json" with fname.open("w", encoding="utf-8") as f: json.dump({"text": texto}, f, ensure_ascii=False) log(f"outbox: {fname}") except Exception as e: log(f"ERRO ao enviar telegram: {e}") # ----------------------------- progresso ----------------------------- _progress_state = { "status": "aguardando", # aguardando|em_andamento|concluido|erro "inicio": None, "fim": None, "combinacoes_total": 0, "combinacoes_feitas": 0, "combinacao_atual": None, # "palavra em cidade" "encontrados": 0, "enviados": 0, "duplicados": 0, "leads": [], # últimos leads capturados (nome, telefone, cidade, palavra_chave) "erro": None, "pais": "BR", "log_tail": [], # últimas linhas do log } def _persist_progress() -> None: try: save_json(PROGRESS_PATH, _progress_state) except Exception as e: log(f"WARN: falha ao persistir progresso: {e}") def _progress_add_lead(lead: dict, palavra_chave: str) -> None: _progress_state["leads"].append({ "nome": lead.get("nome", "")[:120], "telefone": lead.get("telefone", "")[:40], "cidade": lead.get("cidade", "")[:80], "palavra_chave": palavra_chave[:80], "enviado_crm": bool(lead.get("enviado_crm")), }) # Cap em 100 leads pra evitar payload gigante if len(_progress_state["leads"]) > 100: _progress_state["leads"] = _progress_state["leads"][-100:] def _progress_log(msg: str) -> None: stamp = datetime.now().strftime("%H:%M:%S") _progress_state["log_tail"].append(f"[{stamp}] {msg}") if len(_progress_state["log_tail"]) > 40: _progress_state["log_tail"] = _progress_state["log_tail"][-40:] # ----------------------------- envio ao CRM ----------------------------- def _obter_config_crm(config: dict) -> dict: """ Prioridade: 1. Bloco "crm" no schedule (novo formato) 2. Fallback: config.py (Linkia default) """ crm_cfg = config.get("crm") or {} if crm_cfg.get("token") and crm_cfg.get("location_id"): return { "token": crm_cfg.get("token"), "location_id": crm_cfg.get("location_id"), "pipeline_id": crm_cfg.get("pipeline_id") or "", "stage_id": crm_cfg.get("stage_id") or "", "conta_id": crm_cfg.get("conta_id"), } try: from config import ( LINKIA_TOKEN, LINKIA_LOCATION_ID, LINKIA_PIPELINE_ID, LINKIA_STAGE_ID ) return { "token": LINKIA_TOKEN, "location_id": LINKIA_LOCATION_ID, "pipeline_id": LINKIA_PIPELINE_ID, "stage_id": LINKIA_STAGE_ID, "conta_id": None, } except Exception as e: log(f"ERRO: nenhuma config de CRM disponível: {e}") return {} def _enviar_lead_ao_crm(lead: dict, crm: dict, enriquecer_flag: bool, pais: str) -> tuple[bool, str]: """ Reusa a lógica de criar_contato_ghl / criar_oportunidade_ghl / enriquecimento do app.py. Retorna (ok, mensagem). """ try: # import atrasado pra evitar carregar Flask no worker se não for necessário from app import ( criar_contato_ghl, criar_oportunidade_ghl, enriquecer_empresa, enriquecer_empresa_us, _adicionar_nota_ghl, ) except Exception as e: return False, f"import app.py falhou: {e}" token = crm.get("token") or "" location_id = crm.get("location_id") or "" pipeline_id = crm.get("pipeline_id") or "" stage_id = crm.get("stage_id") or "" if not (token and location_id and pipeline_id and stage_id): return False, "config CRM incompleta" # ── Enriquecimento (opcional) ── dados_enriq: dict = {} if enriquecer_flag: try: if (pais or "BR").upper() == "US": dados_enriq = enriquecer_empresa_us( lead.get("nome", ""), website=lead.get("site", ""), cidade=lead.get("cidade", ""), ) or {} else: dados_enriq = enriquecer_empresa(lead.get("nome", "")) or {} # Populaç no lead antes de mandar ao CRM if dados_enriq.get("telefone") and not lead.get("telefone"): lead["telefone"] = dados_enriq["telefone"] if dados_enriq.get("email") and not lead.get("email"): lead["email"] = dados_enriq["email"] if dados_enriq.get("cnpj"): lead["cnpj"] = dados_enriq["cnpj"] if dados_enriq.get("decisor_nome"): lead["decisor_nome"] = dados_enriq["decisor_nome"] lead["decisor_cargo"] = dados_enriq.get("decisor_cargo", "") if dados_enriq.get("linkedin_url"): lead["linkedin_url"] = dados_enriq["linkedin_url"] except Exception as e: log(f" WARN enriquecimento falhou p/ {lead.get('nome')}: {e}") # ── Contato ── try: contact_id, err = criar_contato_ghl(lead, token, location_id, source="Google Maps (auto)") except Exception as e: return False, f"contato: {e}" if not contact_id: return False, f"contato: {err or 'falhou'}" # ── Oportunidade ── try: ok, opportunity_id, err_oport = criar_oportunidade_ghl( lead, contact_id, token, location_id, pipeline_id, stage_id, source="Google Maps (auto)" ) except Exception as e: return False, f"oportunidade: {e}" is_duplicate = err_oport and "duplicate" in err_oport.lower() if not (ok or is_duplicate): return False, f"oportunidade: {err_oport}" lead["enviado_crm"] = True lead["contact_id"] = contact_id # ── Campos enriquecidos na oportunidade + nota ── if enriquecer_flag and dados_enriq and any(dados_enriq.get(k) for k in ("email", "cnpj", "decisor_nome", "linkedin_url")): try: _adicionar_nota_ghl(contact_id, token, location_id, dados_enriq, opportunity_id=opportunity_id) except Exception as e: log(f" WARN nota/campos enriquecidos: {e}") return True, "ok" # ----------------------------- core ----------------------------- def rodar_prospeccao( palavras_override: list[str] | None = None, cidades_override: list[str] | None = None, limite_override: int | None = None, notificar: bool = True, filtro_sem_site_override: bool | None = None, ) -> dict: """ Executa a rodada completa. Retorna dicionário com resumo. Overrides existem pra suportar teste manual sem editar o JSON. Ordem de execução: para cada cidade → para cada palavra-chave. """ inicio = datetime.now() config = load_json(CONFIG_PATH, {}) if not config: log("ERRO: prospeccao_schedule.json não encontrado ou vazio") if notificar: enviar_telegram("Prospecção automática falhou: config ausente.") return {"ok": False, "erro": "config ausente"} if not config.get("ativo", True) and not (palavras_override or cidades_override): log("Automação marcada como inativa. Encerrando.") return {"ok": True, "skipped": True} # Aceita "palavras_chave" (novo) ou "nichos" (legado) palavras = palavras_override or config.get("palavras_chave") or config.get("nichos") or [] cidades = cidades_override or config.get("cidades", []) limite = limite_override or int(config.get("limite_por_busca", 20)) pais = (config.get("pais") or "BR").upper() enriquecer_flag = bool(config.get("enriquecer", False)) if filtro_sem_site_override is not None: filtro_sem_site = bool(filtro_sem_site_override) else: filtro_sem_site = bool(config.get("filtro_sem_site", False)) if not palavras or not cidades: log("ERRO: palavras_chave ou cidades vazios na config") if notificar: enviar_telegram("Prospecção automática: config sem palavras-chave ou cidades.") return {"ok": False, "erro": "config incompleta"} crm = _obter_config_crm(config) if not crm.get("token"): log("ERRO: sem config de CRM válida") if notificar: enviar_telegram("Prospecção automática: CRM não configurado.") return {"ok": False, "erro": "sem CRM"} # Inicializa progresso _progress_state.update({ "status": "em_andamento", "inicio": inicio.isoformat(), "fim": None, "combinacoes_total": len(cidades) * len(palavras), "combinacoes_feitas": 0, "combinacao_atual": None, "encontrados": 0, "enviados": 0, "duplicados": 0, "leads": [], "erro": None, "pais": pais, "log_tail": [], }) _persist_progress() _progress_log(f"Início | {len(cidades)} cidades × {len(palavras)} palavras-chave") prospectados = load_json(PROSPECTADOS_PATH, []) if not isinstance(prospectados, list): prospectados = [] prospectados_set = {p.strip().lower() for p in prospectados if isinstance(p, str)} telefones_ja_prospectados: set[str] = set() processed_ids = load_json(PROCESSED_IDS_PATH, []) if not isinstance(processed_ids, list): processed_ids = [] processed_ids_set = set(processed_ids) resultados = [] total_encontrados = 0 total_enviados = 0 total_duplicados = 0 total_com_site_pulados = 0 # Ordem pedida: para cada cidade → para cada palavra-chave for cidade in cidades: for palavra in palavras: combo_label = f"{palavra} em {cidade}" log(f"===== {combo_label} =====") _progress_state["combinacao_atual"] = combo_label _persist_progress() _progress_log(f"Rodando: {combo_label}") encontrados = 0 enviados = 0 duplicados = 0 com_site_pulados = 0 erro_busca = None try: leads = _buscar_leads(palavra, cidade, limite, pais) encontrados = len(leads) log(f"scrape/api retornou {encontrados} leads") _progress_log(f"{combo_label}: {encontrados} encontrado(s)") except Exception as e: erro_busca = str(e)[:200] log(f"ERRO busca {palavra}/{cidade}: {erro_busca}") log(traceback.format_exc()) leads = [] for lead in leads: nome = (lead.get("nome") or "").strip() tel_norm = normalize_phone(lead.get("telefone", "")) key = dedup_key(nome, cidade) if not nome: continue # Marca cidade e "nicho" (palavra-chave) no lead lead["cidade"] = cidade lead["nicho"] = palavra if key in prospectados_set: duplicados += 1 log(f" dup (nome+cidade): {nome}") continue if tel_norm and tel_norm in telefones_ja_prospectados: duplicados += 1 log(f" dup (telefone): {nome} {tel_norm}") continue # Filtro "somente leads sem site" (aplica ANTES do envio ao CRM) if filtro_sem_site: site_val = (lead.get("site") or "").strip() if site_val: com_site_pulados += 1 log(f" filtro sem-site: pulando {nome} (site: {site_val[:60]})") continue ok_env, msg_env = _enviar_lead_ao_crm(lead, crm, enriquecer_flag, pais) if ok_env: contact_id = lead.get("contact_id") if contact_id and contact_id not in processed_ids_set: processed_ids.append(contact_id) processed_ids_set.add(contact_id) prospectados_set.add(key) prospectados.append(key) if tel_norm: telefones_ja_prospectados.add(tel_norm) enviados += 1 _progress_state["enviados"] += 1 log(f" ✓ CRM: {nome}") else: log(f" ✗ CRM erro {nome}: {msg_env}") _progress_log(f"CRM erro {nome}: {msg_env[:80]}") _progress_state["encontrados"] += 1 _progress_add_lead(lead, palavra) _persist_progress() time.sleep(0.5) resultados.append({ "cidade": cidade, "palavra_chave": palavra, "encontrados": encontrados, "enviados": enviados, "duplicados": duplicados, "com_site_pulados": com_site_pulados, "erro": erro_busca, }) total_encontrados += encontrados total_enviados += enviados total_duplicados += duplicados total_com_site_pulados += com_site_pulados _progress_state["duplicados"] += duplicados _progress_state["combinacoes_feitas"] += 1 extra_site = f", {com_site_pulados} c/ site pulado(s)" if com_site_pulados else "" _progress_log( f"OK {combo_label}: {encontrados} encontrado(s), {enviados} enviado(s), {duplicados} dup{extra_site}" ) _persist_progress() # persiste após cada combinação pra não perder progresso em caso de crash save_json(PROSPECTADOS_PATH, prospectados) save_json(PROCESSED_IDS_PATH, processed_ids) duracao = (datetime.now() - inicio).total_seconds() log( f"Rodada concluída em {duracao:.1f}s | " f"encontrados={total_encontrados} enviados={total_enviados} dup={total_duplicados}" ) _progress_state.update({ "status": "concluido", "fim": datetime.now().isoformat(), "combinacao_atual": None, }) _persist_progress() # Monta relatório linhas = ["Prospecção automática concluída 🎯", ""] if filtro_sem_site: linhas.append("Filtro ativo: apenas leads SEM site") linhas.append("") linhas.append(f"Buscas realizadas: {len(resultados)}") for r in resultados: if r["erro"]: linhas.append(f"• {r['palavra_chave']} em {r['cidade']} → erro: {r['erro']}") else: extras = [] if r["duplicados"]: extras.append(f"{r['duplicados']} dup") if r.get("com_site_pulados"): extras.append(f"{r['com_site_pulados']} c/ site pulados") extra = f" | {' | '.join(extras)}" if extras else "" linhas.append( f"• {r['palavra_chave']} em {r['cidade']} → {r['encontrados']} encontrados, " f"{r['enviados']} enviados ao CRM{extra}" ) linhas.append("") resumo_total = ( f"Total: {total_encontrados} leads encontrados | " f"{total_enviados} enviados ao CRM | {total_duplicados} duplicados evitados" ) if total_com_site_pulados: resumo_total += f" | {total_com_site_pulados} com site pulados" linhas.append(resumo_total) linhas.append(f"Duração: {duracao:.0f}s") relatorio = "\n".join(linhas) log(relatorio) if notificar: enviar_telegram(relatorio) return { "ok": True, "resultados": resultados, "total_encontrados": total_encontrados, "total_enviados": total_enviados, "total_duplicados": total_duplicados, "total_com_site_pulados": total_com_site_pulados, "filtro_sem_site": filtro_sem_site, "duracao_s": duracao, } # ----------------------------- busca (Places API / Playwright) ----------------------------- def _buscar_leads(palavra: str, cidade: str, limite: int, pais: str) -> list[dict]: """ Preferência: Google Places API (se GOOGLE_PLACES_KEY estiver no config.py). Fallback: scrape_google_maps do prospector (Playwright), que não suporta 'pais'. """ try: from config import GOOGLE_PLACES_KEY except Exception: GOOGLE_PLACES_KEY = "" if GOOGLE_PLACES_KEY: return _buscar_via_places_api(palavra, cidade, limite, pais, GOOGLE_PLACES_KEY) log("Sem GOOGLE_PLACES_KEY, usando Playwright (país 'BR' fixo)") return scrape_google_maps(palavra, cidade, limite) def _buscar_via_places_api(palavra: str, cidade: str, limite: int, pais: str, api_key: str) -> list[dict]: """Google Places API (New) — sem risco de bloqueio.""" import requests PLACES_URL = "https://places.googleapis.com/v1/places:searchText" FIELD_MASK = ( "places.displayName,places.formattedAddress,places.nationalPhoneNumber," "places.websiteUri,places.rating,places.userRatingCount,places.id" ) preposicao = "in" if pais == "US" else "em" query = f"{palavra} {preposicao} {cidade}" log(f"Places API: '{query}'") leads: list[dict] = [] page_token = None coletados = 0 while coletados < limite: body = {"textQuery": query, "maxResultCount": 20} if page_token: body["pageToken"] = page_token try: r = requests.post( PLACES_URL, headers={ "Content-Type": "application/json", "X-Goog-Api-Key": api_key, "X-Goog-FieldMask": FIELD_MASK, }, json=body, timeout=15, ) if r.status_code != 200: log(f"Places API erro: {r.status_code} {r.text[:120]}") break data = r.json() places = data.get("places", []) if not places: break for place in places: if coletados >= limite: break nome = place.get("displayName", {}).get("text", "").strip() if not nome or len(nome) < 2: continue lead = { "nome": nome, "telefone": place.get("nationalPhoneNumber", "").strip(), "site": place.get("websiteUri", "").strip(), "endereco": place.get("formattedAddress", "").strip(), "avaliacao": str(place.get("rating", "")), "nicho": palavra, "cidade": cidade, "enviado_crm": False, "contact_id": None, } leads.append(lead) coletados += 1 page_token = data.get("nextPageToken") if not page_token: break time.sleep(2) except Exception as e: log(f"Places API exceção: {e}") break return leads # ----------------------------- CLI ----------------------------- def _parse_args(argv): """CLI leve pra permitir teste manual.""" import argparse parser = argparse.ArgumentParser(description="Prospecção automática agendada") parser.add_argument("--palavra", "--nicho", action="append", dest="palavras", help="Sobrescreve palavra-chave (pode repetir)") parser.add_argument("--cidade", action="append", help="Sobrescreve cidade (pode repetir)") parser.add_argument("--limite", type=int, help="Sobrescreve limite por busca") parser.add_argument("--sem-telegram", action="store_true", help="Não envia relatório ao Telegram") parser.add_argument("--filtro-sem-site", action="store_true", help="Só envia ao CRM leads que NÃO tenham site preenchido") return parser.parse_args(argv) if __name__ == "__main__": args = _parse_args(sys.argv[1:]) try: rodar_prospeccao( palavras_override=args.palavras, cidades_override=args.cidade, limite_override=args.limite, notificar=not args.sem_telegram, filtro_sem_site_override=True if args.filtro_sem_site else None, ) except Exception as e: log(f"FATAL: {e}") log(traceback.format_exc()) _progress_state.update({ "status": "erro", "erro": str(e)[:300], "fim": datetime.now().isoformat(), }) _persist_progress() enviar_telegram(f"Prospecção automática falhou: {str(e)[:300]}") sys.exit(1)