""" Google Ads Generator & Auditor - Interface Streamlit Aplicação que extrai conteúdo de Landing Pages e gera automaticamente ativos de campanha para Google Ads usando IA (OpenAI ou Gemini). Inclui módulo de Auditoria (Google Ads Scorecard) para avaliação de qualidade. Suporta estrutura multi-grupo (3 Ad Groups temáticos por campanha). """ import os import re import json import time import threading import streamlit as st from dotenv import load_dotenv from src.scraper import scrape_landing_page from src.ai_generator import generate_google_ads_assets, MODELS from src.auditor import audit_campaign from src.exporter import ( create_keywords_df, create_negative_keywords_df, create_ads_df, create_sitelinks_df, create_callouts_df, export_all_to_excel, ) # ─── Carregar variáveis de ambiente ─────────────────────────────── load_dotenv() def _run_with_progress(fn, progress_bar, status_text, msg: str, start_pct: int = 30, end_pct: int = 95): """ Executa fn() em thread separada enquanto anima a barra de progresso com indicador numérico crescente (ex: 30%, 35%, 40%...). Não altera a lógica — apenas dá feedback visual durante chamadas longas. """ result = [None] error = [None] def _worker(): try: result[0] = fn() except Exception as e: error[0] = e thread = threading.Thread(target=_worker) thread.start() pct = start_pct while thread.is_alive() and pct < end_pct: pct = min(pct + 5, end_pct) progress_bar.progress(pct, text=f"{msg} ({pct}%)") status_text.info(f"🤖 {msg} ({pct}%)") time.sleep(1.5) thread.join() if error[0]: raise error[0] return result[0] def _build_json_export(assets: dict) -> str: """Monta o JSON de exportação no formato padronizado para Google Ads (multi-grupo).""" # Ad Groups formatados groups_formatted = [] for group in assets.get("ad_groups", []): # Palavras-chave formatadas com tipo de correspondência keywords_formatted = [] for kw in group.get("keywords", []): keyword = kw.get("keyword", "") match_type = kw.get("match_type", "Phrase") if match_type == "Exact": keywords_formatted.append(f"[{keyword}]") elif match_type == "Phrase": keywords_formatted.append(f'"{keyword}"') else: keywords_formatted.append(keyword) groups_formatted.append({ "nome_grupo": group.get("name", "Grupo"), "palavras_chave_fundo_de_funil": keywords_formatted, "titulos_headlines": group.get("headlines", []), "descricoes": group.get("descriptions", []), }) # Palavras-chave negativas com motivo (nível campanha) negative_kw_formatted = [] for nkw in assets.get("negative_keywords", []): keyword = nkw.get("keyword", "") reason = nkw.get("reason", "") if reason: negative_kw_formatted.append(f"{keyword} ({reason})") else: negative_kw_formatted.append(keyword) # Sitelinks formatados (nível campanha) sitelinks_formatted = [] for sl in assets.get("sitelinks", []): sitelinks_formatted.append({ "titulo": sl.get("title", ""), "desc_linha_1": sl.get("description1", ""), "desc_linha_2": sl.get("description2", ""), }) export_data = [ { "campanha_google_ads": { "ad_groups": groups_formatted, "palavras_chave_negativas": negative_kw_formatted, "sitelinks": sitelinks_formatted, "callouts": assets.get("callouts", []), } } ] return json.dumps(export_data, ensure_ascii=False, indent=2) def _highlight_keywords(text: str, keywords: list[str]) -> str: """Destaca em negrito as palavras-chave encontradas no texto.""" result = text for kw in keywords: pattern = re.compile(re.escape(kw), re.IGNORECASE) match = pattern.search(result) if match: result = result[:match.start()] + f"**{match.group()}**" + result[match.end():] break # Destacar apenas a primeira keyword encontrada return result # ─── Configuração da Página ─────────────────────────────────────── st.set_page_config( page_title="Google Ads Generator & Auditor", page_icon="📊", layout="wide", ) # ─── CSS Customizado ────────────────────────────────────────────── st.markdown(""" """, unsafe_allow_html=True) # ─── Sidebar ────────────────────────────────────────────────────── with st.sidebar: st.header("Navegação") page = st.radio( "Modo", ["Gerador de Campanha", "Auditor (Scorecard)"], index=0, help="Alterne entre gerar ativos e auditar campanhas existentes.", ) st.divider() st.header("Configurações") provider = st.selectbox( "Provider de IA", ["OpenAI", "Gemini"], index=0, help="Escolha o provider de IA. As chaves devem estar configuradas no arquivo .env", ) model = st.selectbox( "Modelo", MODELS.get(provider, []), index=0, help="Modelo a ser utilizado para geração/auditoria.", ) # Status das chaves configuradas st.divider() openai_ok = bool(os.environ.get("OPENAI_API_KEY")) gemini_ok = bool(os.environ.get("GEMINI_API_KEY")) st.caption("Status das API Keys (.env):") st.markdown(f"- OpenAI: {'✅ Configurada' if openai_ok else '❌ Não encontrada'}") st.markdown(f"- Gemini: {'✅ Configurada' if gemini_ok else '❌ Não encontrada'}") st.divider() campaign_name = st.text_input( "Nome da Campanha", value="Campanha LP", help="Nome que aparecerá na coluna 'Campaign' do CSV.", ) st.divider() st.caption("Desenvolvido com Streamlit + OpenAI + Gemini") # ===================================================================== # MODO: GERADOR DE CAMPANHA # ===================================================================== def render_generator(): """Renderiza a interface do Gerador de Campanha (multi-grupo).""" st.markdown('
Google Ads Generator
', unsafe_allow_html=True) st.markdown( 'Gere ativos de campanha automaticamente a partir da sua Landing Page (3 Grupos de Anúncio)
', unsafe_allow_html=True, ) # ─── Área Principal ─────────────────────────────────────────── url = st.text_input( "URL da Landing Page", placeholder="https://www.seusite.com.br/landing-page", help="Cole a URL completa da sua Landing Page aqui.", ) col_btn, col_status = st.columns([1, 3]) with col_btn: generate_btn = st.button("Gerar Campanha", type="primary", use_container_width=True) # ─── Lógica Principal ──────────────────────────────────────── if generate_btn: # Validações if not url: st.error("Por favor, insira a URL da Landing Page.") st.stop() if not url.startswith(("http://", "https://")): st.error("A URL deve começar com http:// ou https://") st.stop() # Verificar se a chave do provider selecionado está configurada if provider == "OpenAI" and not os.environ.get("OPENAI_API_KEY"): st.error("OPENAI_API_KEY não encontrada no arquivo .env. Configure antes de continuar.") st.stop() elif provider == "Gemini" and not os.environ.get("GEMINI_API_KEY"): st.error("GEMINI_API_KEY não encontrada no arquivo .env. Configure antes de continuar.") st.stop() # Barra de progresso visual progress_bar = st.progress(0, text="Iniciando...") status_text = st.empty() # Step 1: Scraping progress_bar.progress(10, text="Extraindo conteúdo da Landing Page...") status_text.info("🔍 Acessando a URL e extraindo conteúdo...") try: lp_data = scrape_landing_page(url) except Exception as e: progress_bar.empty() status_text.empty() st.error(f"Erro ao acessar a URL: {e}") st.stop() progress_bar.progress(30, text="Conteúdo extraído com sucesso!") status_text.info( f"📄 Extraído: {len(lp_data['paragraphs'])} parágrafos, " f"{len(lp_data['h1'])} H1, {len(lp_data['h2'])} H2, " f"{len(lp_data.get('list_items', []))} itens de lista, " f"{len(lp_data['ctas'])} CTAs" ) # Step 2: IA — executa em thread com progresso animado try: assets, prompts = _run_with_progress( fn=lambda: generate_google_ads_assets( lp_content=lp_data["full_text"], provider=provider, model=model, ), progress_bar=progress_bar, status_text=status_text, msg=f"Gerando ativos com {provider} ({model})", start_pct=35, end_pct=95, ) except ValueError as e: progress_bar.empty() status_text.empty() st.error(str(e)) st.stop() except Exception as e: progress_bar.empty() status_text.empty() st.error(f"Erro ao gerar ativos: {e}") st.stop() progress_bar.progress(100, text="Campanha gerada com sucesso! (100%)") status_text.success("✅ Campanha gerada com sucesso!") # Salvar no session_state st.session_state["assets"] = assets st.session_state["prompts"] = prompts st.session_state["lp_data"] = lp_data st.session_state["campaign_name"] = campaign_name st.session_state["provider_used"] = provider st.session_state["model_used"] = model # ─── Exibição dos Resultados ───────────────────────────────── if "assets" in st.session_state: assets = st.session_state["assets"] prompts = st.session_state.get("prompts", {}) camp = st.session_state.get("campaign_name", "Campanha LP") prov_used = st.session_state.get("provider_used", "") model_used = st.session_state.get("model_used", "") ad_groups = assets.get("ad_groups", []) st.divider() # Métricas resumo (totais agregados) total_kw = sum(len(g.get("keywords", [])) for g in ad_groups) total_headlines = sum(len(g.get("headlines", [])) for g in ad_groups) total_desc = sum(len(g.get("descriptions", [])) for g in ad_groups) col1, col2, col3, col4, col5, col6, col7 = st.columns(7) with col1: st.metric("Grupos", len(ad_groups)) with col2: st.metric("Keywords", total_kw) with col3: st.metric("Negativas", len(assets.get("negative_keywords", []))) with col4: st.metric("Títulos", total_headlines) with col5: st.metric("Descrições", total_desc) with col6: st.metric("Sitelinks", len(assets.get("sitelinks", []))) with col7: st.metric("Callouts", len(assets.get("callouts", []))) st.divider() # Abas com resultados tab_groups, tab_neg, tab_sl, tab_co, tab_prompts, tab_raw = st.tabs([ "Grupos de Anúncio", "Negativas", "Sitelinks", "Callouts", "Prompts Utilizados", "Dados da LP", ]) with tab_groups: st.subheader("Grupos de Anúncio (Ad Groups)") if not ad_groups: st.info("Nenhum grupo de anúncio gerado.") else: # Sub-abas para cada grupo group_tab_names = [f"Grupo {i+1}: {g.get('name', '')}" for i, g in enumerate(ad_groups)] group_tabs = st.tabs(group_tab_names) for idx, (gtab, group) in enumerate(zip(group_tabs, ad_groups)): with gtab: group_name = group.get("name", f"Grupo {idx+1}") st.markdown(f'{group_name}
', unsafe_allow_html=True) # Keywords do grupo st.write("**Palavras-chave:**") group_kw = group.get("keywords", []) if group_kw: for i, kw in enumerate(group_kw, 1): keyword = kw.get("keyword", "") match_type = kw.get("match_type", "Phrase") if match_type == "Exact": formatted = f"[{keyword}]" elif match_type == "Phrase": formatted = f'"{keyword}"' else: formatted = keyword st.markdown(f"{i}. `{formatted}` ({match_type})") else: st.info("Nenhuma keyword neste grupo.") st.markdown("---") # Headlines do grupo st.write("**Títulos (Headlines):**") kw_list = sorted( [kw["keyword"].lower() for kw in group_kw if kw.get("keyword")], key=len, reverse=True, ) for i, h in enumerate(group.get("headlines", []), 1): char_count = len(h) color = "green" if char_count <= 30 else "red" h_highlighted = _highlight_keywords(h, kw_list) st.markdown(f"{i}. {h_highlighted} — :{color}[{char_count} chars]") st.markdown("---") # Descriptions do grupo st.write("**Descrições:**") for i, d in enumerate(group.get("descriptions", []), 1): char_count = len(d) color = "green" if char_count <= 90 else "red" st.markdown(f"{i}. {d} — :{color}[{char_count} chars]") st.markdown("---") # DataFrame RSA do grupo st.write("**Tabela RSA (formato Google Ads Editor):**") # Montar assets temporário para o grupo individual single_group_assets = { "ad_groups": [group], "negative_keywords": [], "sitelinks": [], "callouts": [], } ads_df = create_ads_df(single_group_assets, camp) if not ads_df.empty: st.dataframe(ads_df, use_container_width=True, hide_index=True) # DataFrame consolidado de todas as keywords st.markdown("---") st.write("**Tabela Consolidada de Keywords (todos os grupos):**") kw_df = create_keywords_df(assets, camp) if not kw_df.empty: st.dataframe(kw_df, use_container_width=True, hide_index=True) with tab_neg: st.subheader("Palavras-chave Negativas (Nível de Campanha)") nkw_df = create_negative_keywords_df(assets, camp) if not nkw_df.empty: st.dataframe(nkw_df, use_container_width=True, hide_index=True) else: st.info("Nenhuma palavra-chave negativa gerada.") with tab_sl: st.subheader("Sitelinks (Nível de Campanha)") sl_df = create_sitelinks_df(assets, camp) if not sl_df.empty: st.dataframe(sl_df, use_container_width=True, hide_index=True) else: st.info("Nenhum sitelink gerado.") with tab_co: st.subheader("Callouts — Frases de Destaque (Nível de Campanha)") co_df = create_callouts_df(assets, camp) if not co_df.empty: st.dataframe(co_df, use_container_width=True, hide_index=True) else: st.info("Nenhum callout gerado.") with tab_prompts: st.subheader("Prompts Utilizados") st.caption(f"Provider: **{prov_used}** | Modelo: **{model_used}**") st.markdown("#### Prompt de Sistema (System Prompt)") st.code(prompts.get("system_prompt", "N/A"), language=None) st.markdown("#### Prompt do Usuário (User Prompt)") st.code(prompts.get("user_prompt", "N/A"), language=None) with tab_raw: st.subheader("Dados Extraídos da Landing Page") lp_data = st.session_state.get("lp_data", {}) if lp_data: st.write(f"**URL:** {lp_data.get('url', '')}") st.write(f"**Título:** {lp_data.get('title', '')}") st.write(f"**Meta Description:** {lp_data.get('meta_description', '')}") with st.expander("Headings (H1, H2, H3)"): for h in lp_data.get("h1", []): st.write(f"**H1:** {h}") for h in lp_data.get("h2", []): st.write(f"**H2:** {h}") for h in lp_data.get("h3", []): st.write(f"**H3:** {h}") with st.expander("Itens de Lista / Benefícios"): for li in lp_data.get("list_items", []): st.write(f"- {li}") with st.expander("CTAs Encontrados"): for c in lp_data.get("ctas", []): st.write(f"- {c}") with st.expander("Texto Completo Enviado à IA"): st.code(lp_data.get("full_text", ""), language=None) # ─── Botões de Download ─────────────────────────────────── st.divider() st.subheader("Download dos Ativos") col_dl1, col_dl2, col_dl3 = st.columns(3) with col_dl1: excel_data = export_all_to_excel(assets, camp) st.download_button( label="Baixar Excel (.xlsx)", data=excel_data, file_name="google_ads_assets.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", use_container_width=True, ) with col_dl2: kw_csv = create_keywords_df(assets, camp).to_csv(index=False) st.download_button( label="Baixar Keywords (.csv)", data=kw_csv, file_name="google_ads_keywords.csv", mime="text/csv", use_container_width=True, ) with col_dl3: json_export = _build_json_export(assets) st.download_button( label="Baixar JSON (.json)", data=json_export, file_name="google_ads_assets.json", mime="application/json", use_container_width=True, ) # ===================================================================== # MODO: AUDITOR (SCORECARD) # ===================================================================== def _get_score_color(score: float) -> str: """Retorna a cor com base na nota.""" if score >= 8.0: return "#34a853" # verde elif score >= 5.0: return "#fbbc04" # amarelo else: return "#ea4335" # vermelho def _get_criterion_emoji(nota: float, nota_maxima: float) -> str: """Retorna emoji com base na porcentagem atingida do critério.""" if nota_maxima == 0: return "⚪" pct = nota / nota_maxima if pct >= 0.9: return "✅" elif pct >= 0.6: return "⚠️" else: return "❌" def render_auditor(): """Renderiza a interface do Auditor de Campanha (Google Ads Scorecard).""" st.markdown('Google Ads Scorecard
', unsafe_allow_html=True) st.markdown( 'Audite a qualidade dos seus ativos de campanha com IA
', unsafe_allow_html=True, ) # ─── Fonte do JSON ──────────────────────────────────────────── has_session_json = "assets" in st.session_state json_source = st.radio( "Fonte do JSON para auditoria", [ "Usar campanha da sessão atual" if has_session_json else "Usar campanha da sessão atual (nenhuma disponível)", "Colar JSON manualmente", ], index=0 if has_session_json else 1, horizontal=True, help="Escolha de onde vem o JSON da campanha a ser auditada.", ) campaign_json = "" if "sessão atual" in json_source and has_session_json: # Construir o JSON a partir dos assets na sessão campaign_json = _build_json_export(st.session_state["assets"]) with st.expander("Ver JSON da sessão atual", expanded=False): st.code(campaign_json, language="json") elif "sessão atual" in json_source and not has_session_json: st.warning("Nenhuma campanha gerada na sessão atual. Gere uma campanha primeiro no modo 'Gerador' ou cole o JSON manualmente.") return else: campaign_json = st.text_area( "Cole o JSON da campanha aqui", height=300, placeholder='[{"campanha_google_ads": {"ad_groups": [...], "palavras_chave_negativas": [...], ...}}]', help="Cole o JSON completo da campanha no formato de exportação.", ) # ─── Botão de Auditoria ─────────────────────────────────────── col_audit_btn, _ = st.columns([1, 3]) with col_audit_btn: audit_btn = st.button("Auditar Campanha", type="primary", use_container_width=True) # ─── Lógica de Auditoria ───────────────────────────────────── if audit_btn: if not campaign_json or not campaign_json.strip(): st.error("Por favor, forneça o JSON da campanha para auditoria.") st.stop() # Validar JSON try: json.loads(campaign_json) except json.JSONDecodeError: st.error("O JSON fornecido é inválido. Verifique a formatação e tente novamente.") st.stop() # Verificar chave da API if provider == "OpenAI" and not os.environ.get("OPENAI_API_KEY"): st.error("OPENAI_API_KEY não encontrada no arquivo .env. Configure antes de continuar.") st.stop() elif provider == "Gemini" and not os.environ.get("GEMINI_API_KEY"): st.error("GEMINI_API_KEY não encontrada no arquivo .env. Configure antes de continuar.") st.stop() # Barra de progresso visual progress_bar = st.progress(0, text="Iniciando auditoria...") status_text = st.empty() progress_bar.progress(15, text="Validando JSON da campanha...") status_text.info("📋 JSON validado. Preparando envio para IA...") # Executa auditoria em thread com progresso animado try: audit_result, audit_prompts = _run_with_progress( fn=lambda: audit_campaign( campaign_json=campaign_json, provider=provider, model=model, ), progress_bar=progress_bar, status_text=status_text, msg=f"Analisando campanha com {provider} ({model})", start_pct=25, end_pct=95, ) except ValueError as e: progress_bar.empty() status_text.empty() st.error(str(e)) st.stop() except json.JSONDecodeError: progress_bar.empty() status_text.empty() st.error("A IA retornou uma resposta inválida. Tente novamente ou troque de modelo.") st.stop() except Exception as e: progress_bar.empty() status_text.empty() st.error(f"Erro durante auditoria: {e}") st.stop() progress_bar.progress(100, text="Auditoria concluída! (100%)") status_text.success("✅ Auditoria concluída com sucesso!") # Salvar no session_state st.session_state["audit_result"] = audit_result st.session_state["audit_prompts"] = audit_prompts # ─── Exibição dos Resultados da Auditoria ──────────────────── if "audit_result" in st.session_state: audit_result = st.session_state["audit_result"] audit_prompts = st.session_state.get("audit_prompts", {}) st.divider() # ─── Score Principal ────────────────────────────────────── nota = audit_result["nota_final"] cor = _get_score_color(nota) col_score, col_resumo = st.columns([1, 3]) with col_score: st.markdown( f'{nota:.1f}
' f'de 10.0 pontos
' f'{grupo_nome}
', unsafe_allow_html=True) st.markdown( f"""