Spaces:
Sleeping
Sleeping
import tweepy | |
from transformers import pipeline, GPT2LMHeadModel, GPT2Tokenizer | |
import os | |
import streamlit as st | |
from datetime import datetime | |
# Verificar variáveis de ambiente | |
required_vars = [ | |
'TWITTER_API_KEY', | |
'TWITTER_API_SECRET_KEY', | |
'TWITTER_ACCESS_TOKEN', | |
'TWITTER_ACCESS_TOKEN_SECRET', | |
'TWITTER_BEARER_TOKEN' | |
] | |
# Verificação inicial das variáveis de ambiente | |
missing_vars = [] | |
for var in required_vars: | |
if os.getenv(var) is None: | |
missing_vars.append(var) | |
print(f"Erro: A variável de ambiente '{var}' não está definida.") | |
else: | |
print(f"{var} carregada com sucesso.") | |
if missing_vars: | |
raise ValueError(f"As seguintes variáveis de ambiente são necessárias: {', '.join(missing_vars)}") | |
# Autenticação com Twitter para leitura | |
client = tweepy.Client( | |
bearer_token=os.getenv('TWITTER_BEARER_TOKEN') | |
) | |
# 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) | |
# Configuração da query e campos do tweet | |
query = 'BBB25 -filter:retweets lang:pt -is:reply' | |
tweet_fields = ['text', 'created_at', 'lang', 'public_metrics'] | |
try: | |
# Busca tweets com os campos especificados | |
tweets = client.search_recent_tweets( | |
query=query, | |
max_results=100, | |
tweet_fields=tweet_fields | |
) | |
if not tweets.data: | |
print("Nenhum tweet encontrado") | |
st.error("Nenhum tweet encontrado para análise") | |
st.stop() | |
# Análise de sentimentos | |
sentiment_pipeline = pipeline( | |
'sentiment-analysis', | |
model='cardiffnlp/twitter-xlm-roberta-base-sentiment' | |
) | |
sentiments = [] | |
for tweet in tweets.data: | |
# Verificação adicional para garantir que temos tweets em português | |
if hasattr(tweet, 'lang') and tweet.lang == 'pt': | |
result = sentiment_pipeline(tweet.text) | |
sentiments.append(result[0]['label']) | |
# Calcular taxas | |
if sentiments: | |
positive = sentiments.count('positive') | |
negative = sentiments.count('negative') | |
neutral = sentiments.count('neutral') | |
total = len(sentiments) | |
positive_ratio = positive / total | |
negative_ratio = negative / total | |
neutral_ratio = neutral / total | |
# Gerar mensagem com IA | |
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." | |
# Gerar texto | |
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] # Limitar a 280 caracteres | |
try: | |
# Postar no Twitter | |
api.update_status(status=generated_text) | |
print(f"Tweet postado com sucesso: {generated_text}") | |
# Interface Streamlit | |
st.title("Análise de Sentimentos - BBB25") | |
# 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 | |
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") | |
except Exception as e: | |
st.error(f"Erro ao postar tweet: {str(e)}") | |
print(f"Erro ao postar: {e}") | |
except tweepy.errors.BadRequest as e: | |
st.error(f"Erro na requisição ao Twitter: {str(e)}") | |
print(f"Erro na requisição: {str(e)}") | |
except tweepy.errors.TweepyException as e: | |
st.error(f"Erro do Tweepy: {str(e)}") | |
print(f"Erro do Tweepy: {str(e)}") | |
except Exception as e: | |
st.error(f"Erro inesperado: {str(e)}") | |
print(f"Erro inesperado: {str(e)}") | |
# Footer | |
st.markdown("---") | |
st.markdown( | |
""" | |
<div style='text-align: center'> | |
<small>Desenvolvido com ❤️ usando Streamlit e Transformers</small> | |
</div> | |
""", | |
unsafe_allow_html=True | |
) | |