#!/usr/bin/env python3 """ Google Alerts Email Reporter Automatiza a geração de relatórios de crise a partir de Google Alerts via Gmail API. Integra com Grok 4 (xAI API) para análise de sentimento avançada. """ import os import base64 import re import json from datetime import datetime, timedelta from pathlib import Path from bs4 import BeautifulSoup import requests from dotenv import load_dotenv from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient.errors import HttpError # Carrega variáveis de ambiente load_dotenv() # Configurações SCOPES = ['https://www.googleapis.com/auth/gmail.readonly'] ALERTS_SENDER = os.getenv('ALERTS_SENDER', 'alerts-noreply@google.com') XAI_API_KEY = os.getenv('XAI_API_KEY', '') XAI_API_URL = os.getenv('XAI_API_URL', 'https://api.x.ai/v1/chat/completions') NUM_EMAILS = int(os.getenv('NUM_EMAILS', '2')) SEARCH_QUERY = os.getenv('SEARCH_QUERY', 'Banco Master') DEBUG_MODE = os.getenv('DEBUG_MODE', 'false').lower() == 'true' # Cores para logs no terminal class Colors: HEADER = '\033[95m' OKBLUE = '\033[94m' OKCYAN = '\033[96m' OKGREEN = '\033[92m' WARNING = '\033[93m' FAIL = '\033[91m' ENDC = '\033[0m' BOLD = '\033[1m' def log(message, level='info'): """Log colorido no terminal.""" prefix = { 'info': f"{Colors.OKBLUE}ℹ{Colors.ENDC}", 'success': f"{Colors.OKGREEN}✓{Colors.ENDC}", 'warning': f"{Colors.WARNING}⚠{Colors.ENDC}", 'error': f"{Colors.FAIL}✗{Colors.ENDC}", 'debug': f"{Colors.OKCYAN}⚙{Colors.ENDC}" } timestamp = datetime.now().strftime('%H:%M:%S') print(f"[{timestamp}] {prefix.get(level, '')} {message}") def authenticate_gmail(): """Autentica no Gmail API (gera token.json na primeira vez).""" log("Iniciando autenticação Gmail...", 'info') creds = None if os.path.exists('token.json'): creds = Credentials.from_authorized_user_file('token.json', SCOPES) log("Token existente encontrado", 'debug') if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: log("Renovando token expirado...", 'info') creds.refresh(Request()) else: if not os.path.exists('credentials.json'): log("ERRO: credentials.json não encontrado!", 'error') log("Baixe de: https://console.cloud.google.com/apis/credentials", 'warning') raise FileNotFoundError("credentials.json não encontrado") log("Iniciando fluxo OAuth (browser abrirá)...", 'warning') flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES) creds = flow.run_local_server(port=0) with open('token.json', 'w') as token: token.write(creds.to_json()) log("Token salvo com sucesso", 'success') service = build('gmail', 'v1', credentials=creds) log("Autenticação concluída", 'success') return service def get_email_body(payload): """Extrai o body do email (recursivo para MIME).""" if 'parts' in payload: for part in payload['parts']: if part['mimeType'] == 'text/html': data = part['body'].get('data', '') if data: return base64.urlsafe_b64decode(data).decode('utf-8', errors='ignore') elif part['mimeType'] == 'text/plain': data = part['body'].get('data', '') if data: return base64.urlsafe_b64decode(data).decode('utf-8', errors='ignore') elif 'parts' in part: result = get_email_body(part) if result: return result elif payload.get('mimeType') in ['text/html', 'text/plain']: data = payload['body'].get('data', '') if data: return base64.urlsafe_b64decode(data).decode('utf-8', errors='ignore') return '' def get_last_alerts_emails(service, num_emails=2): """Busca os N últimos emails de Google Alerts.""" log(f"Buscando {num_emails} emails de {ALERTS_SENDER}...", 'info') # Query: busca todos (não só não lidos) para pegar histórico recente query = f'from:{ALERTS_SENDER}' try: results = service.users().messages().list( userId='me', q=query, maxResults=num_emails ).execute() messages = results.get('messages', []) if not messages: log("Nenhum email de Google Alerts encontrado!", 'warning') return [] log(f"Encontrados {len(messages)} emails", 'success') alerts_data = [] for idx, msg in enumerate(messages, 1): log(f"Processando email {idx}/{len(messages)}...", 'info') msg_data = service.users().messages().get( userId='me', id=msg['id'], format='full' ).execute() # Extrai headers headers = {h['name']: h['value'] for h in msg_data['payload']['headers']} subject = headers.get('Subject', 'Sem assunto') date_str = headers.get('Date', '') log(f" Subject: {subject[:60]}...", 'debug') log(f" Date: {date_str}", 'debug') # Extrai body body = get_email_body(msg_data['payload']) if not body: log(" Body vazio, pulando...", 'warning') continue if DEBUG_MODE: # Salva HTML para debug debug_file = f"debug_email_{idx}.html" with open(debug_file, 'w', encoding='utf-8') as f: f.write(body) log(f" HTML salvo em {debug_file}", 'debug') # Scrape de alertas soup = BeautifulSoup(body, 'html.parser') items = [] # Google Alerts usa estrutura variável; tentamos múltiplos seletores selectors = [ ('table[role="article"]', 'tr'), # Estrutura comum ('div[class*="alert"]', 'div'), # Fallback 1 ('table', 'tr'), # Fallback 2 (mais genérico) ] for container_selector, item_selector in selectors: containers = soup.find_all(container_selector) if containers: log(f" Usando seletor: {container_selector}", 'debug') break # Parse items parsed_items = 0 for container in containers[:5]: # Max 5 containers por email item_elements = container.find_all(item_selector) for elem in item_elements: # Busca link/título link_elem = elem.find('a', href=True) if not link_elem: continue title = link_elem.get_text(strip=True) link = link_elem['href'] # Remove Google redirect if 'google.com/url?q=' in link: link = re.search(r'q=(https?://[^&]+)', link) link = link.group(1) if link else link_elem['href'] # Busca fonte (texto próximo ao link) source_elem = elem.find('font', color=True) or elem.find('span', class_=re.compile(r'source|cite')) source = source_elem.get_text(strip=True) if source_elem else 'Fonte desconhecida' if title and len(title) > 10: # Filtra títulos muito curtos items.append({ 'title': title, 'source': source, 'link': link, 'date': date_str }) parsed_items += 1 log(f" Extraídos {parsed_items} alertas", 'success' if parsed_items > 0 else 'warning') if items: alerts_data.append({ 'subject': subject, 'date': date_str, 'items': items }) total_items = sum(len(a['items']) for a in alerts_data) log(f"Total: {total_items} alertas de {len(alerts_data)} emails", 'success') return alerts_data except HttpError as error: log(f"Erro na API Gmail: {error}", 'error') return [] def analyze_sentiment_with_grok(items, mode='heavy'): """Analisa sentimento usando Grok 4 via xAI API.""" log(f"Iniciando análise de sentimento (modo: {mode})...", 'info') if not XAI_API_KEY or XAI_API_KEY == 'SEU_API_KEY_AQUI': log("API Key xAI não configurada, usando análise mock", 'warning') return analyze_sentiment_mock(items) sentiments = {'neg': 0, 'pos': 0, 'neu': 0} analyzed_items = [] # Batch processing (max 10 por vez para evitar timeout) batch_size = 10 for i in range(0, len(items), batch_size): batch = items[i:i+batch_size] log(f"Analisando batch {i//batch_size + 1}/{(len(items)-1)//batch_size + 1}...", 'info') # Monta prompt para batch prompts = [] for idx, item in enumerate(batch): prompts.append(f""" [ITEM {idx+1}] Título: {item['title']} Fonte: {item['source']} Link: {item['link'][:80]}... """) full_prompt = f""" Analise o sentimento de cada notícia sobre "{SEARCH_QUERY}" em relação à reputação/imagem da empresa. Critérios: - NEGATIVO: Crise, reclamações, processos, problemas, críticas - POSITIVO: Crescimento, prêmios, melhorias, elogios - NEUTRO: Fatos, dados, informações sem viés Para cada item, responda APENAS no formato: ITEM X: NEGATIVO/POSITIVO/NEUTRO | Razão em 1 frase. {"".join(prompts)} Responda de forma objetiva e direta. """ headers = { 'Authorization': f'Bearer {XAI_API_KEY}', 'Content-Type': 'application/json' } data = { 'model': 'grok-beta' if mode == 'heavy' else 'grok-2-latest', 'messages': [{'role': 'user', 'content': full_prompt}], 'max_tokens': 1000 if mode == 'heavy' else 500, 'temperature': 0.3 # Mais determinístico } try: response = requests.post(XAI_API_URL, json=data, headers=headers, timeout=30) response.raise_for_status() result = response.json()['choices'][0]['message']['content'] if DEBUG_MODE: log(f"Resposta Grok:\n{result}", 'debug') # Parse respostas for idx, line in enumerate(result.split('\n')): if not line.strip(): continue item_idx = i + idx if item_idx >= len(items): break sentiment = 'neu' reason = '' if 'NEGATIVO' in line.upper(): sentiment = 'neg' sentiments['neg'] += 1 elif 'POSITIVO' in line.upper(): sentiment = 'pos' sentiments['pos'] += 1 else: sentiments['neu'] += 1 # Extrai razão if '|' in line: reason = line.split('|')[1].strip() analyzed_items.append({ **items[item_idx], 'sentiment': sentiment, 'reason': reason }) log(f"Batch analisado: {sentiments}", 'success') except requests.exceptions.Timeout: log("Timeout na API Grok, usando fallback", 'warning') return analyze_sentiment_mock(items) except requests.exceptions.RequestException as e: log(f"Erro na API Grok: {e}", 'error') return analyze_sentiment_mock(items) except KeyError as e: log(f"Erro ao parsear resposta Grok: {e}", 'error') return analyze_sentiment_mock(items) total = len(items) log(f"Análise concluída: {sentiments} de {total} items", 'success') return sentiments, total, analyzed_items def analyze_sentiment_mock(items): """Análise de sentimento mock (fallback sem API).""" log("Usando análise mock (sem API)", 'warning') # Palavras-chave para heurística simples negative_keywords = ['problema', 'crise', 'reclamação', 'processo', 'fraude', 'falha', 'erro', 'bloqueio', 'bloqueado', 'golpe'] positive_keywords = ['crescimento', 'prêmio', 'expansão', 'sucesso', 'inovação', 'liderança'] sentiments = {'neg': 0, 'pos': 0, 'neu': 0} analyzed_items = [] for item in items: text = (item['title'] + ' ' + item['source']).lower() neg_score = sum(1 for kw in negative_keywords if kw in text) pos_score = sum(1 for kw in positive_keywords if kw in text) if neg_score > pos_score: sentiment = 'neg' sentiments['neg'] += 1 reason = f"Detectadas {neg_score} palavras negativas" elif pos_score > neg_score: sentiment = 'pos' sentiments['pos'] += 1 reason = f"Detectadas {pos_score} palavras positivas" else: sentiment = 'neu' sentiments['neu'] += 1 reason = "Sem indicadores claros" analyzed_items.append({ **item, 'sentiment': sentiment, 'reason': reason }) return sentiments, len(items), analyzed_items def generate_html_report(alerts_data, sentiments, total_items, analyzed_items): """Gera relatório HTML compacto e profissional.""" log("Gerando relatório HTML...", 'info') percent_neg = (sentiments['neg'] / total_items * 100) if total_items > 0 else 0 percent_pos = (sentiments['pos'] / total_items * 100) if total_items > 0 else 0 percent_neu = (sentiments['neu'] / total_items * 100) if total_items > 0 else 0 # Filtrar top 10 negativos negative_items = [item for item in analyzed_items if item.get('sentiment') == 'neg'][:10] html = f""" Relatório de Crise: {SEARCH_QUERY}

📊 RELATÓRIO DE CRISE: {SEARCH_QUERY}

Período: Últimos {NUM_EMAILS} emails de Google Alerts
Total de alertas: {total_items} notícias analisadas
Gerado em: {datetime.now().strftime('%d/%m/%Y às %H:%M:%S')}

{sentiments['neg']}

Negativos

{percent_neg:.1f}%

{sentiments['pos']}

Positivos

{percent_pos:.1f}%

{sentiments['neu']}

Neutros

{percent_neu:.1f}%
{percent_neg:.0f}% DOS ALERTAS SÃO NEGATIVOS → Crise de imagem em curso. Ação imediata necessária.

PLANO DE AÇÃO URGENTE

⚠️ TOP 10 ALERTAS NEGATIVOS

""" if not negative_items: html += '

Nenhum alerta negativo identificado.

' for idx, item in enumerate(negative_items, 1): date_display = item['date'].split(',')[0] if ',' in item['date'] else item['date'][:20] html += f"""
#{idx} - {item['title'][:150]}{'...' if len(item['title']) > 150 else ''}
NEGATIVO
📅 {date_display} | 🌐 {item['source']} | Ver notícia →
{f'
💡 {item.get("reason", "")}
' if item.get("reason") else ''}
""" html += f"""
""" output_file = 'relatorio_crise.html' with open(output_file, 'w', encoding='utf-8') as f: f.write(html) log(f"Relatório salvo em '{output_file}'", 'success') # Também salva JSON para análise json_file = 'relatorio_dados.json' with open(json_file, 'w', encoding='utf-8') as f: json.dump({ 'timestamp': datetime.now().isoformat(), 'search_query': SEARCH_QUERY, 'total_items': total_items, 'sentiments': sentiments, 'items': analyzed_items }, f, ensure_ascii=False, indent=2) log(f"Dados salvos em '{json_file}'", 'success') return output_file # Execução Principal if __name__ == '__main__': print(f""" {Colors.BOLD}{Colors.HEADER} ╔═══════════════════════════════════════════════════╗ ║ GOOGLE ALERTS REPORTER - Análise de Crise ║ ║ Powered by Gmail API + Grok 4 ║ ╚═══════════════════════════════════════════════════╝ {Colors.ENDC} """) try: # Autenticação service = authenticate_gmail() # Busca emails alerts = get_last_alerts_emails(service, num_emails=NUM_EMAILS) if not alerts: log("Nenhum dado para processar. Verifique a query de busca.", 'error') exit(1) # Extrai todos os items all_items = [item for alert in alerts for item in alert['items']] if not all_items: log("Nenhum alerta encontrado nos emails.", 'error') exit(1) # Análise de sentimento sentiments, total, analyzed_items = analyze_sentiment_with_grok(all_items, mode='heavy') # Gera relatório output_file = generate_html_report(alerts, sentiments, total, analyzed_items) print(f""" {Colors.OKGREEN} ═══════════════════════════════════════════════════ ✓ PROCESSAMENTO CONCLUÍDO ═══════════════════════════════════════════════════ {Colors.ENDC} 📧 Emails processados: {len(alerts)} 📰 Alertas analisados: {total} 🔴 Negativos: {sentiments['neg']} ({sentiments['neg']/total*100:.1f}%) 🟢 Positivos: {sentiments['pos']} ({sentiments['pos']/total*100:.1f}%) 🟡 Neutros: {sentiments['neu']} ({sentiments['neu']/total*100:.1f}%) 📄 Relatório: {output_file} {Colors.OKCYAN}→ Abra o arquivo no navegador para visualizar{Colors.ENDC} """) # Auto-open no browser (macOS) import subprocess try: subprocess.run(['open', output_file], check=True) log("Abrindo relatório no navegador...", 'info') except: pass except KeyboardInterrupt: print(f"\n{Colors.WARNING}Processo interrompido pelo usuário{Colors.ENDC}") exit(0) except Exception as e: log(f"Erro fatal: {e}", 'error') if DEBUG_MODE: import traceback traceback.print_exc() exit(1)