Twitter API и Python: как создать скрипты автоматизации соцсети

import os
from dotenv import load_dotenv

# Загружаем переменные окружения из файла .env
load_dotenv()

# Получаем ключи безопасным способом
api_key = os.environ.get("TWITTER_API_KEY")
api_secret = os.environ.get("TWITTER_API_SECRET")
access_token = os.environ.get("TWITTER_ACCESS_TOKEN")
access_secret = os.environ.get("TWITTER_ACCESS_SECRET")

TWITTER_API_KEY=ваш_ключ_api
TWITTER_API_SECRET=ваш_секретный_ключ
TWITTER_ACCESS_TOKEN=ваш_токен_доступа
TWITTER_ACCESS_SECRET=ваш_секретный_токен

pip install tweepy

import tweepy

# Аутентификация с использованием OAuth 1.0a (для некоторых endpoint'ов)
auth = tweepy.OAuth1UserHandler(
consumer_key="ваш_api_key",
consumer_secret="ваш_api_secret",
access_token="ваш_access_token",
access_token_secret="ваш_access_token_secret"
)
api = tweepy.API(auth)

# Аутентификация с использованием OAuth 2.0 Bearer Token (для Twitter API v2)
client = tweepy.Client(
bearer_token="ваш_bearer_token",
consumer_key="ваш_api_key",
consumer_secret="ваш_api_secret",
access_token="ваш_access_token",
access_token_secret="ваш_access_token_secret"
)

try:
# Для API v1.1
user = api.verify_credentials()
print(f"Аутентификация успешна! Имя пользователя: {user.screen_name}")

# Для API v2
me = client.get_me()
print(f"Аутентификация в API v2 успешна! ID: {me.data.id}")
except Exception as e:
print(f"Ошибка аутентификации: {e}")

# Включаем автоматическую задержку при достижении лимита запросов
api = tweepy.API(auth, wait_on_rate_limit=True)

try:
# Выполнение API-запроса
tweets = api.home_timeline()
for tweet in tweets:
print(f"{tweet.user.name}: {tweet.text}")
except tweepy.TweepyException as e:
print(f"Ошибка Tweepy: {e}")
except Exception as e:
print(f"Неожиданная ошибка: {e}")

# Публикация простого твита
response = client.create_tweet(text="Привет, Twitter! Это твит, отправленный с помощью Python и Tweepy.")
print(f"Твит опубликован! ID: {response.data['id']}")

# Публикация твита с медиа (требует API v1.1)
# Сначала загружаем медиафайл
media = api.media_upload("path/to/image.jpg")
# Затем публикуем твит с прикрепленным медиа
api.update_status("Твит с изображением!", media_ids=[media.media_id])

# Поиск последних твитов по хештегу (API v2)
query = "#Python lang:ru" # Поиск твитов на русском языке с хештегом Python
tweets = client.search_recent_tweets(query=query, max_results=10)

if tweets.data:
for tweet in tweets.data:
print(f"ID: {tweet.id}, Текст: {tweet.text}")
else:
print("Твитов не найдено")

# Расширенный поиск с фильтрами (API v1.1)
search_results = api.search_tweets(
q="#Python filter:images -filter:retweets", 
lang="ru",
count=10,
result_type="recent"
)

for tweet in search_results:
print(f"{tweet.user.screen_name}: {tweet.text}")

# Получение последних упоминаний (API v1.1)
mentions = api.mentions_timeline(count=5)
for mention in mentions:
print(f"От: @{mention.user.screen_name}, Текст: {mention.text}")

# Отвечаем на упоминание
api.update_status(
f"@{mention.user.screen_name} Спасибо за обращение! Мы рассмотрим ваш запрос.",
in_reply_to_status_id=mention.id
)

import tweepy
import time
import pandas as pd
from textblob import TextBlob # Для анализа тональности

# Настройка аутентификации (опущена для краткости)
api = tweepy.API(auth, wait_on_rate_limit=True)

# Список ключевых слов и хештегов бренда
brand_terms = ["#BrandName", "@BrandOfficial", "Brand Name"]

# Функция анализа тональности (настроения) твита
def analyze_sentiment(text):
analysis = TextBlob(text)
if analysis.sentiment.polarity > 0:
return "позитивный"
elif analysis.sentiment.polarity < 0:
return "негативный"
else:
return "нейтральный"

# Основной цикл мониторинга
last_id = None
tweet_data = []

while True:
try:
for term in brand_terms:
tweets = api.search_tweets(q=term, count=50, since_id=last_id)

for tweet in tweets:
# Пропускаем ретвиты
if hasattr(tweet, 'retweeted_status'):
continue

sentiment = analyze_sentiment(tweet.text)

# Сохраняем данные для анализа
tweet_data.append({
'id': tweet.id,
'user': tweet.user.screen_name,
'text': tweet.text,
'sentiment': sentiment,
'created_at': tweet.created_at
})

# Отвечаем на твит в зависимости от его тональности
if sentiment == "позитивный":
reply_text = f"@{tweet.user.screen_name} Спасибо за ваш положительный отзыв! Мы ценим вашу поддержку."
elif sentiment == "негативный":
reply_text = f"@{tweet.user.screen_name} Нам жаль, что у вас возникли проблемы. Пожалуйста, напишите детали в личные сообщения, мы постараемся помочь."
else:
reply_text = f"@{tweet.user.screen_name} Благодарим за упоминание нашего бренда. Если у вас есть вопросы, мы всегда готовы помочь!"

# Публикуем ответ
api.update_status(reply_text, in_reply_to_status_id=tweet.id)

# Обновляем ID последнего обработанного твита
if last_id is None or tweet.id > last_id:
last_id = tweet.id

# Каждые 30 минут сохраняем данные в CSV
df = pd.DataFrame(tweet_data)
df.to_csv('brand_mentions_analysis.csv', index=False)

# Пауза для соблюдения лимитов API
time.sleep(60)

except Exception as e:
print(f"Произошла ошибка: {e}")
time.sleep(15)

# Создаем класс для обработки потока (API v1.1)
class MyStreamListener(tweepy.StreamListener):
def on_status(self, status):
print(f"{status.user.screen_name}: {status.text}")

def on_error(self, status_code):
if status_code == 420:
# Возвращаем False в случае ошибки превышения лимита
return False
print(f"Ошибка: {status_code}")

# Инициализируем поток
stream_listener = MyStreamListener()
stream = tweepy.Stream(auth=api.auth, listener=stream_listener)

# Запускаем отслеживание ключевых слов
stream.filter(track=["Python", "программирование", "разработка"], languages=["ru"])

import pandas as pd
import time

# Функция для сбора твитов по запросу
def collect_tweets(query, count=100):
tweets_data = []

# Пагинация результатов поиска с помощью Cursor
for tweet in tweepy.Cursor(api.search_tweets, 
q=query, 
lang="ru", 
tweet_mode="extended").items(count):
try:
data = {
'created_at': tweet.created_at,
'id': tweet.id,
'full_text': tweet.full_text,
'user_name': tweet.user.screen_name,
'user_followers': tweet.user.followers_count,
'retweet_count': tweet.retweet_count,
'favorite_count': tweet.favorite_count,
'hashtags': [hashtag['text'] for hashtag in tweet.entities['hashtags']]
}
tweets_data.append(data)
except Exception as e:
print(f"Ошибка при обработке твита: {e}")

# Создаем DataFrame
df = pd.DataFrame(tweets_data)
return df

# Собираем данные
query = "#Python OR #DataScience -filter:retweets"
df_tweets = collect_tweets(query, count=1000)

# Сохраняем в CSV
df_tweets.to_csv('python_tweets.csv', index=False)
print(f"Собрано и сохранено {len(df_tweets)} твитов")

import matplotlib.pyplot as plt
import seaborn as sns
from collections import Counter
import re
from textblob import TextBlob

# Анализ активности по времени
df_tweets['date'] = pd.to_datetime(df_tweets['created_at']).dt.date
tweets_by_date = df_tweets.groupby('date').size()

plt.figure(figsize=(12, 6))
tweets_by_date.plot(kind='line')
plt.title('Количество твитов по датам')
plt.xlabel('Дата')
plt.ylabel('Количество твитов')
plt.grid(True)
plt.savefig('tweets_by_date.png')

# Анализ популярных хештегов
all_hashtags = []
for hashtags_list in df_tweets['hashtags']:
all_hashtags.extend(hashtags_list)

hashtag_counts = Counter(all_hashtags)
top_hashtags = hashtag_counts.most_common(15)

plt.figure(figsize=(12, 8))
sns.barplot(x=[tag for tag, count in top_hashtags],
y=[count for tag, count in top_hashtags])
plt.title('15 самых популярных хештегов')
plt.xlabel('Хештег')
plt.ylabel('Частота')
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig('top_hashtags.png')

# Анализ тональности твитов (sentiment analysis)
def get_sentiment(text):
try:
return TextBlob(text).sentiment.polarity
except:
return 0

df_tweets['sentiment'] = df_tweets['full_text'].apply(get_sentiment)

# Классификация настроений
df_tweets['sentiment_category'] = pd.cut(
df_tweets['sentiment'],
bins=[-1, -0.25, 0.25, 1],
labels=['Негативный', 'Нейтральный', 'Позитивный']
)

sentiment_counts = df_tweets['sentiment_category'].value_counts()

plt.figure(figsize=(10, 6))
sentiment_counts.plot(kind='pie', autopct='%1.1f%%', colors=['red', 'gray', 'green'])
plt.title('Распределение тональности твитов')
plt.ylabel('')
plt.savefig('sentiment_distribution.png')

import nltk
from nltk.tokenize import word_tokenize
from nltk.corpus import stopwords
from wordcloud import WordCloud

# Загружаем необходимые ресурсы NLTK
nltk.download('punkt')
nltk.download('stopwords')

# Объединяем все тексты твитов
all_text = ' '.join(df_tweets['full_text'])

# Удаляем ссылки, упоминания и спецсимволы
clean_text = re.sub(r'http\S+|@\S+|#\S+|[^\w\s]', '', all_text)

# Токенизация и удаление стоп-слов
stop_words = set(stopwords.words('russian'))
tokens = word_tokenize(clean_text.lower())
filtered_tokens = [word for word in tokens if word not in stop_words and len(word) > 2]

# Создаем облако слов
wordcloud = WordCloud(width=800, height=400, background_color='white', max_words=200).generate(' '.join(filtered_tokens))

plt.figure(figsize=(16, 8))
plt.imshow(wordcloud, interpolation='bilinear')
plt.axis('off')
plt.title('Облако слов из твитов')
plt.savefig('wordcloud.png')

import schedule
import time
import json
import os

# Загружаем запланированные твиты из JSON
def load_scheduled_tweets():
if os.path.exists('scheduled_tweets.json'):
with open('scheduled_tweets.json', 'r', encoding='utf-8') as file:
return json.load(file)
return []

# Публикуем твит с поддержкой медиа
def post_tweet(tweet_data):
try:
media_ids = []

# Загружаем медиафайлы, если они есть
if 'media_paths' in tweet_data and tweet_data['media_paths']:
for media_path in tweet_data['media_paths']:
if os.path.exists(media_path):
media = api.media_upload(media_path)
media_ids.append(media.media_id)

# Публикуем твит
if media_ids:
api.update_status(status=tweet_data['text'], media_ids=media_ids)
else:
api.update_status(status=tweet_data['text'])

print(f"Твит опубликован: {tweet_data['text'][:30]}...")
return True
except Exception as e:
print(f"Ошибка при публикации твита: {e}")
return False

# Основная функция планировщика
def scheduler():
tweets = load_scheduled_tweets()

# Фильтруем твиты, которые нужно опубликовать сегодня
today = time.strftime('%Y-%m-%d')
today_tweets = [t for t in tweets if t['scheduled_date'] == today]

# Планируем публикации на конкретное время
for tweet in today_tweets:
schedule.every().day.at(tweet['scheduled_time']).do(
lambda t=tweet: post_tweet(t) and tweets.remove(t)
)

# Основной цикл планировщика
while True:
schedule.run_pending()
time.sleep(60)

# Запускаем планировщик
if __name__ == '__main__':
scheduler()

import threading
import queue

# Очередь для обработки задач
task_queue = queue.Queue()

# Обработчик упоминаний
def handle_mentions():
last_mention_id = None

while True:
try:
# Получаем новые упоминания
mentions = api.mentions_timeline(since_id=last_mention_id)

if mentions:
last_mention_id = mentions[0].id

for mention in mentions:
# Добавляем задачу в очередь
task_queue.put({
'type': 'mention',
'data': mention
})

# Избегаем частых запросов к API
time.sleep(60)
except Exception as e:
print(f"Ошибка при получении упоминаний: {e}")
time.sleep(120)

# Обработчик личных сообщений
def handle_direct_messages():
last_dm_id = None

while True:
try:
# Получаем новые личные сообщения (требует расширенного доступа)
dms = api.get_direct_messages(count=50)

if dms:
newest_dm_id = max(dm.id for dm in dms)

if last_dm_id is None:
last_dm_id = newest_dm_id
continue

new_dms = [dm for dm in dms if dm.id > last_dm_id]
last_dm_id = newest_dm_id

for dm in new_dms:
# Добавляем задачу в очередь
task_queue.put({
'type': 'direct_message',
'data': dm
})

# Интервал проверки новых сообщений
time.sleep(60)
except Exception as e:
print(f"Ошибка при получении сообщений: {e}")
time.sleep(120)

# Обработчик задач
def process_tasks():
while True:
try:
# Получаем задачу из очереди
task = task_queue.get()

if task['type'] == 'mention':
mention = task['data']
text = mention.text.lower()

# Логика обработки упоминаний
if 'помощь' in text or 'help' in text:
response = f"@{mention.user.screen_name} Вот список доступных команд: #помощь, #статус, #информация"
elif 'статус' in text:
response = f"@{mention.user.screen_name} Все системы работают нормально! 👍"
else:
response = f"@{mention.user.screen_name} Спасибо за обращение! Чем я могу помочь?"

# Отправляем ответ
api.update_status(response, in_reply_to_status_id=mention.id)

elif task['type'] == 'direct_message':
dm = task['data']

# Логика обработки личных сообщений
response = "Спасибо за ваше сообщение! Это автоматический ответ. Команда скоро свяжется с вами."
api.send_direct_message(recipient_id=dm.sender_id, text=response)

# Отмечаем задачу как выполненную
task_queue.task_done()

# Небольшая пауза между обработкой задач
time.sleep(2)

except Exception as e:
print(f"Ошибка при обработке задачи: {e}")
task_queue.task_done()

# Запускаем потоки для параллельной работы
mention_thread = threading.Thread(target=handle_mentions)
dm_thread = threading.Thread(target=handle_direct_messages)
task_thread = threading.Thread(target=process_tasks)

mention_thread.daemon = True
dm_thread.daemon = True
task_thread.daemon = True

mention_thread.start()
dm_thread.start()
task_thread.start()

# Держим основной поток активным
while True:
time.sleep(10)

import logging
from logging.handlers import RotatingFileHandler

# Настройка логирования
logger = logging.getLogger('twitter_bot')
logger.setLevel(logging.INFO)

# Ротация логов для контроля размера файлов
handler = RotatingFileHandler('twitter_bot.log', maxBytes=10485760, backupCount=5)
formatter = logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s')
handler.setFormatter(formatter)

logger.addHandler(handler)

# Использование в коде
try:
# Код, выполняющий операции с API
api.update_status("Тестовый твит")
logger.info("Твит успешно опубликован")
except Exception as e:
logger.error(f"Ошибка при публикации твита: {e}", exc_info=True)