Spaces:
Sleeping
Sleeping
File size: 5,407 Bytes
dd9cc55 aeb2715 3251884 fb0b40b dd9cc55 81a43fe dd9cc55 81a43fe fcb0322 81a43fe dd9cc55 3251884 dd9cc55 3251884 dd9cc55 3251884 aeb2715 fcb0322 3251884 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
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
)
|