import tweepy from transformers import pipeline, GPT2LMHeadModel, GPT2Tokenizer, AutoModelForSequenceClassification, AutoTokenizer import os import streamlit as st from datetime import datetime import time from tenacity import retry, stop_after_attempt, wait_exponential def debug_print(message): """Função para imprimir mensagens de debug tanto no console quanto no Streamlit""" print(message) st.text(message) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), retry=lambda e: isinstance(e, tweepy.errors.TooManyRequests) ) def fetch_tweets(client, query, tweet_fields): try: debug_print(f"Iniciando busca com query: {query}") debug_print(f"Campos solicitados: {tweet_fields}") tweets = client.search_recent_tweets( query=query, max_results=10, tweet_fields=tweet_fields ) if tweets is None: debug_print("Nenhum resultado retornado da API") return None if not hasattr(tweets, 'data'): debug_print("Resposta não contém dados") return None debug_print(f"Tweets encontrados: {len(tweets.data) if tweets.data else 0}") return tweets except tweepy.errors.TooManyRequests as e: debug_print(f"Rate limit atingido: {str(e)}") raise e except tweepy.errors.TwitterServerError as e: debug_print(f"Erro do servidor Twitter: {str(e)}") raise e except tweepy.errors.BadRequest as e: debug_print(f"Erro na requisição: {str(e)}") raise e except Exception as e: debug_print(f"Erro inesperado na busca: {str(e)}") raise e @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), retry=lambda e: isinstance(e, tweepy.errors.TooManyRequests) ) def post_tweet(api, text): return api.update_status(status=text) def main(): try: st.title("Análise de Sentimentos - BBB25") # Verificar variáveis de ambiente debug_print("Verificando variáveis de ambiente...") required_vars = [ 'TWITTER_API_KEY', 'TWITTER_API_SECRET_KEY', 'TWITTER_ACCESS_TOKEN', 'TWITTER_ACCESS_TOKEN_SECRET', 'TWITTER_BEARER_TOKEN' ] missing_vars = [] for var in required_vars: if os.getenv(var) is None: missing_vars.append(var) debug_print(f"Erro: A variável de ambiente '{var}' não está definida.") else: debug_print(f"{var} carregada com sucesso.") if missing_vars: raise ValueError(f"Variáveis de ambiente faltando: {', '.join(missing_vars)}") debug_print("Iniciando autenticação com Twitter...") # Autenticação com Twitter para leitura client = tweepy.Client( bearer_token=os.getenv('TWITTER_BEARER_TOKEN'), wait_on_rate_limit=True ) # Autenticação com Twitter para postagem auth = tweepy.OAuth1UserHandler( os.getenv('TWITTER_API_KEY'), os.getenv('TWITTER_API_SECRET_KEY'), os.getenv('TWITTER_ACCESS_TOKEN'), os.getenv('TWITTER_ACCESS_TOKEN_SECRET') ) api = tweepy.API(auth, wait_on_rate_limit=True) # Vamos testar a autenticação com uma query simples debug_print("Testando autenticação...") try: test_query = "test" test_response = client.search_recent_tweets(query=test_query, max_results=10) debug_print("Teste de autenticação bem sucedido") except Exception as e: debug_print(f"Erro no teste de autenticação: {str(e)}") raise e # Query principal query = 'BBB25 lang:pt -is:retweet -is:reply' tweet_fields = ['text', 'created_at', 'lang', 'public_metrics'] debug_print("Iniciando busca principal de tweets...") with st.spinner('Buscando tweets...'): tweets = fetch_tweets(client, query, tweet_fields) if tweets is None: st.error("Não foi possível obter tweets") return if not tweets.data: st.warning("Nenhum tweet encontrado com os critérios especificados") debug_print("Busca retornou vazia") return debug_print(f"Encontrados {len(tweets.data)} tweets") # Mostrar alguns tweets encontrados para debug st.subheader("Tweets encontrados (preview):") for i, tweet in enumerate(tweets.data[:3]): st.text(f"Tweet {i+1}: {tweet.text[:100]}...") # Análise de sentimentos with st.spinner('Analisando sentimentos...'): debug_print("Iniciando análise de sentimentos...") # Inicializando o pipeline com modelo em português sentiment_pipeline = pipeline( "sentiment-analysis", model="pierreguillou/bert-base-cased-gs-sentiment-pt", tokenizer="pierreguillou/bert-base-cased-gs-sentiment-pt" ) sentiments = [] for tweet in tweets.data: if hasattr(tweet, 'lang') and tweet.lang == 'pt': result = sentiment_pipeline(tweet.text) # Mapeando os resultados para positive/negative/neutral label = result[0]['label'] if label == 'POSITIVE': sentiments.append('positive') elif label == 'NEGATIVE': sentiments.append('negative') else: sentiments.append('neutral') debug_print(f"Sentimento analisado: {label}") time.sleep(1) # Calcular taxas if sentiments: positive = sentiments.count('positive') negative = sentiments.count('negative') neutral = sentiments.count('neutral') total = len(sentiments) debug_print(f"Total de sentimentos analisados: {total}") positive_ratio = positive / total negative_ratio = negative / total neutral_ratio = neutral / total # Gerar mensagem com IA with st.spinner('Gerando novo tweet...'): debug_print("Iniciando geração de texto...") tokenizer = GPT2Tokenizer.from_pretrained('gpt2') model = GPT2LMHeadModel.from_pretrained('gpt2') if positive_ratio > 0.6: prompt = "Write an exciting tweet about BBB25 with a positive tone in Portuguese." elif negative_ratio > 0.6: prompt = "Write an informative tweet about BBB25 with a neutral tone in Portuguese." else: prompt = "Write a buzzing tweet about BBB25 with an engaging tone in Portuguese." debug_print(f"Usando prompt: {prompt}") input_ids = tokenizer.encode(prompt, return_tensors='pt') outputs = model.generate( input_ids, max_length=25, do_sample=True, pad_token_id=tokenizer.eos_token_id ) generated_text = tokenizer.decode(outputs[0], skip_special_tokens=True) generated_text = generated_text[:280] debug_print(f"Texto gerado: {generated_text}") # Postar tweet with st.spinner('Postando tweet...'): debug_print("Tentando postar tweet...") post_tweet(api, generated_text) st.success("Tweet postado com sucesso!") debug_print("Tweet postado com sucesso") # Interface Streamlit st.title("Resultados da Análise") # Mostrar estatísticas col1, col2, col3 = st.columns(3) with col1: st.metric("Sentimento Positivo", f"{positive_ratio:.1%}") with col2: st.metric("Sentimento Neutro", f"{neutral_ratio:.1%}") with col3: st.metric("Sentimento Negativo", f"{negative_ratio:.1%}") # Mostrar tweet gerado st.subheader("Tweet Gerado e Postado") st.write(generated_text) # Logging debug_print("Salvando log...") log_entry = { 'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'positive_ratio': positive_ratio, 'negative_ratio': negative_ratio, 'neutral_ratio': neutral_ratio, 'tweet': generated_text } with open('posting_log.txt', 'a') as f: f.write(f"{str(log_entry)}\n") debug_print("Log salvo com sucesso") except Exception as e: st.error(f"Erro: {str(e)}") debug_print(f"Erro fatal: {str(e)}") raise e finally: st.markdown("---") st.markdown( """
Desenvolvido com ❤️ usando Streamlit e Transformers
""", unsafe_allow_html=True ) if __name__ == "__main__": main()