Twitter(X) データ収集と分析
Twitter API v2でツイートデータを取得し、トレンド分析や感情分析を実装します。
この記事で学べること
- Twitter API v2の認証と基本操作
- ツイート検索とユーザー情報取得
- データの可視化とトレンド分析
- 感情分析の実装
1. 準備：必要なライブラリ
# インストール
pip install tweepy
pip install pandas
pip install matplotlib
pip install seaborn
pip install textblob
pip install wordcloud
pip install python-dotenvrequirements.txt
tweepy==4.14.0
pandas==2.1.0
matplotlib==3.8.0
seaborn==0.13.0
textblob==0.17.1
wordcloud==1.9.2
python-dotenv==1.0.02. Twitter API認証情報の取得
Twitter Developer Portal での設定
1. Twitter Developer Portal にアクセス
https://developer.twitter.com/
2. アカウント作成とプロジェクト作成
「Projects & Apps」→「Create Project」
3. アプリケーションを作成
- App name: 任意の名前
- Use case: Research/Academic など
4. API Keysを取得
- API Key (Consumer Key)
- API Secret Key (Consumer Secret)
- Bearer Token
- Access Token
- Access Token Secret
5. APIアクセスレベル確認
- Essential (無料): 500,000ツイート/月
- Elevated: 申請が必要.envファイルの設定
# .env
TWITTER_BEARER_TOKEN=YOUR_BEARER_TOKEN
TWITTER_API_KEY=YOUR_API_KEY
TWITTER_API_SECRET=YOUR_API_SECRET
TWITTER_ACCESS_TOKEN=YOUR_ACCESS_TOKEN
TWITTER_ACCESS_SECRET=YOUR_ACCESS_SECRET3. 基本的なツイート取得
twitter_scraper.py
#!/usr/bin/env python3
"""Twitter(X) スクレイパー"""
import os
import tweepy
from dotenv import load_dotenv
import pandas as pd
from datetime import datetime, timedelta
# 環境変数読み込み
load_dotenv()
class TwitterScraper:
def __init__(self):
"""Twitter API クライアント初期化"""
self.bearer_token = os.getenv('TWITTER_BEARER_TOKEN')
# API v2 クライアント
self.client = tweepy.Client(
bearer_token=self.bearer_token,
consumer_key=os.getenv('TWITTER_API_KEY'),
consumer_secret=os.getenv('TWITTER_API_SECRET'),
access_token=os.getenv('TWITTER_ACCESS_TOKEN'),
access_token_secret=os.getenv('TWITTER_ACCESS_SECRET'),
wait_on_rate_limit=True
)
def search_tweets(self, query, max_results=100):
"""キーワードでツイート検索"""
try:
tweets = self.client.search_recent_tweets(
query=query,
max_results=max_results,
tweet_fields=['created_at', 'public_metrics', 'lang'],
user_fields=['username', 'verified', 'public_metrics'],
expansions=['author_id']
)
if not tweets.data:
print("ツイートが見つかりませんでした")
return None
# ユーザー情報をマッピング
users = {user.id: user for user in tweets.includes['users']}
# データフレーム作成
tweet_list = []
for tweet in tweets.data:
user = users.get(tweet.author_id)
tweet_data = {
'id': tweet.id,
'text': tweet.text,
'created_at': tweet.created_at,
'language': tweet.lang,
'retweet_count': tweet.public_metrics['retweet_count'],
'reply_count': tweet.public_metrics['reply_count'],
'like_count': tweet.public_metrics['like_count'],
'quote_count': tweet.public_metrics['quote_count'],
'username': user.username if user else None,
'verified': user.verified if user else False,
'followers': user.public_metrics['followers_count'] if user else 0
}
tweet_list.append(tweet_data)
df = pd.DataFrame(tweet_list)
return df
except Exception as e:
print(f"エラー: {e}")
return None
def get_user_tweets(self, username, max_results=100):
"""特定ユーザーのツイート取得"""
try:
# ユーザーID取得
user = self.client.get_user(username=username)
user_id = user.data.id
# ツイート取得
tweets = self.client.get_users_tweets(
id=user_id,
max_results=max_results,
tweet_fields=['created_at', 'public_metrics'],
exclude=['retweets', 'replies']
)
if not tweets.data:
return None
tweet_list = []
for tweet in tweets.data:
tweet_data = {
'text': tweet.text,
'created_at': tweet.created_at,
'likes': tweet.public_metrics['like_count'],
'retweets': tweet.public_metrics['retweet_count']
}
tweet_list.append(tweet_data)
return pd.DataFrame(tweet_list)
except Exception as e:
print(f"エラー: {e}")
return None
# 使用例
if __name__ == '__main__':
scraper = TwitterScraper()
# Python関連のツイート検索
df = scraper.search_tweets('Python programming', max_results=100)
if df is not None:
print(f"取得したツイート数: {len(df)}")
print(f"\n最も人気のツイート:")
top_tweet = df.loc[df['like_count'].idxmax()]
print(f"テキスト: {top_tweet['text'][:100]}...")
print(f"いいね数: {top_tweet['like_count']:,}")実行結果
取得したツイート数: 100
最も人気のツイート:
テキスト: Just released Python 3.12! Check out the new features including improved error messages...
いいね数: 15,2344. ツイートデータの可視化
visualizer.py
#!/usr/bin/env python3
"""Twitter データ可視化"""
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
from wordcloud import WordCloud
import re
# スタイル設定
sns.set_style('whitegrid')
plt.rcParams['font.sans-serif'] = ['MS Gothic', 'Yu Gothic', 'Arial']
plt.rcParams['axes.unicode_minus'] = False
def plot_engagement_metrics(df):
"""エンゲージメント指標のグラフ"""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# いいね数の分布
axes[0, 0].hist(df['like_count'], bins=30, color='#1DA1F2', alpha=0.7)
axes[0, 0].set_xlabel('いいね数', fontsize=11)
axes[0, 0].set_ylabel('ツイート数', fontsize=11)
axes[0, 0].set_title('いいね数の分布', fontsize=13, fontweight='bold')
# リツイート数の分布
axes[0, 1].hist(df['retweet_count'], bins=30, color='#17BF63', alpha=0.7)
axes[0, 1].set_xlabel('リツイート数', fontsize=11)
axes[0, 1].set_ylabel('ツイート数', fontsize=11)
axes[0, 1].set_title('リツイート数の分布', fontsize=13, fontweight='bold')
# 時間別ツイート数
df['hour'] = pd.to_datetime(df['created_at']).dt.hour
hourly_counts = df['hour'].value_counts().sort_index()
axes[1, 0].bar(hourly_counts.index, hourly_counts.values, color='#F91880')
axes[1, 0].set_xlabel('時刻', fontsize=11)
axes[1, 0].set_ylabel('ツイート数', fontsize=11)
axes[1, 0].set_title('時間帯別ツイート数', fontsize=13, fontweight='bold')
axes[1, 0].set_xticks(range(0, 24, 2))
# エンゲージメント率（上位20ツイート）
df['engagement'] = df['like_count'] + df['retweet_count'] + df['reply_count']
top_tweets = df.nlargest(20, 'engagement')
axes[1, 1].barh(
range(len(top_tweets)),
top_tweets['engagement'],
color='#FFAD1F'
)
axes[1, 1].set_xlabel('総エンゲージメント', fontsize=11)
axes[1, 1].set_ylabel('ツイート', fontsize=11)
axes[1, 1].set_title('エンゲージメント上位20', fontsize=13, fontweight='bold')
axes[1, 1].set_yticks([])
plt.tight_layout()
plt.savefig('twitter_engagement.png', dpi=300, bbox_inches='tight')
plt.show()
def create_wordcloud(df):
"""ワードクラウド作成"""
# テキストを結合
all_text = ' '.join(df['text'].astype(str))
# URLとメンションを除去
all_text = re.sub(r'http\S+', '', all_text)
all_text = re.sub(r'@\w+', '', all_text)
all_text = re.sub(r'#\w+', '', all_text)
# ワードクラウド生成
wordcloud = WordCloud(
width=1200,
height=600,
background_color='white',
colormap='viridis',
max_words=100,
relative_scaling=0.5
).generate(all_text)
# 表示
plt.figure(figsize=(15, 8))
plt.imshow(wordcloud, interpolation='bilinear')
plt.axis('off')
plt.title('ツイート頻出ワード', fontsize=16, fontweight='bold', pad=20)
plt.tight_layout()
plt.savefig('twitter_wordcloud.png', dpi=300, bbox_inches='tight')
plt.show()
def plot_time_series(df):
"""時系列グラフ"""
df['date'] = pd.to_datetime(df['created_at']).dt.date
daily_stats = df.groupby('date').agg({
'like_count': 'sum',
'retweet_count': 'sum',
'text': 'count'
}).reset_index()
daily_stats.columns = ['date', 'total_likes', 'total_retweets', 'tweet_count']
fig, axes = plt.subplots(2, 1, figsize=(12, 8))
# ツイート数の推移
axes[0].plot(
daily_stats['date'],
daily_stats['tweet_count'],
marker='o',
linewidth=2,
color='#1DA1F2'
)
axes[0].set_ylabel('ツイート数', fontsize=11)
axes[0].set_title('日別ツイート数の推移', fontsize=13, fontweight='bold')
axes[0].grid(True, alpha=0.3)
# エンゲージメントの推移
axes[1].plot(
daily_stats['date'],
daily_stats['total_likes'],
marker='o',
linewidth=2,
label='いいね',
color='#F91880'
)
axes[1].plot(
daily_stats['date'],
daily_stats['total_retweets'],
marker='s',
linewidth=2,
label='リツイート',
color='#17BF63'
)
axes[1].set_xlabel('日付', fontsize=11)
axes[1].set_ylabel('エンゲージメント数', fontsize=11)
axes[1].set_title('日別エンゲージメントの推移', fontsize=13, fontweight='bold')
axes[1].legend()
axes[1].grid(True, alpha=0.3)
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig('twitter_timeseries.png', dpi=300, bbox_inches='tight')
plt.show()5. 感情分析
sentiment_analyzer.py
#!/usr/bin/env python3
"""ツイート感情分析"""
from textblob import TextBlob
import pandas as pd
import matplotlib.pyplot as plt
def analyze_sentiment(text):
"""テキストの感情分析"""
try:
analysis = TextBlob(text)
polarity = analysis.sentiment.polarity
# 極性を分類
if polarity > 0.1:
return 'ポジティブ', polarity
elif polarity < -0.1:
return 'ネガティブ', polarity
else:
return '中立', polarity
except:
return '不明', 0.0
def analyze_tweets_sentiment(df):
"""ツイートの感情分析を実行"""
sentiments = []
polarities = []
for text in df['text']:
sentiment, polarity = analyze_sentiment(text)
sentiments.append(sentiment)
polarities.append(polarity)
df['sentiment'] = sentiments
df['polarity'] = polarities
return df
def plot_sentiment_distribution(df):
"""感情分布のグラフ"""
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# 感情カテゴリの分布
sentiment_counts = df['sentiment'].value_counts()
colors = {
'ポジティブ': '#17BF63',
'中立': '#FFD700',
'ネガティブ': '#F91880'
}
bars = axes[0].bar(
sentiment_counts.index,
sentiment_counts.values,
color=[colors.get(s, '#1DA1F2') for s in sentiment_counts.index]
)
axes[0].set_xlabel('感情', fontsize=12)
axes[0].set_ylabel('ツイート数', fontsize=12)
axes[0].set_title('感情分布', fontsize=14, fontweight='bold')
# 値をバーの上に表示
for bar in bars:
height = bar.get_height()
axes[0].text(
bar.get_x() + bar.get_width()/2.,
height,
f'{int(height)}',
ha='center',
va='bottom',
fontsize=11
)
# 極性スコアの分布
axes[1].hist(df['polarity'], bins=30, color='#1DA1F2', alpha=0.7, edgecolor='black')
axes[1].axvline(x=0, color='red', linestyle='--', linewidth=2, label='中立')
axes[1].set_xlabel('極性スコア', fontsize=12)
axes[1].set_ylabel('ツイート数', fontsize=12)
axes[1].set_title('極性スコア分布', fontsize=14, fontweight='bold')
axes[1].legend()
plt.tight_layout()
plt.savefig('twitter_sentiment.png', dpi=300, bbox_inches='tight')
plt.show()
# 統計情報
print("\n=== 感情分析結果 ===")
print(f"ポジティブ: {(df['sentiment']=='ポジティブ').sum()} ({(df['sentiment']=='ポジティブ').sum()/len(df)*100:.1f}%)")
print(f"中立: {(df['sentiment']=='中立').sum()} ({(df['sentiment']=='中立').sum()/len(df)*100:.1f}%)")
print(f"ネガティブ: {(df['sentiment']=='ネガティブ').sum()} ({(df['sentiment']=='ネガティブ').sum()/len(df)*100:.1f}%)")
print(f"平均極性: {df['polarity'].mean():.3f}")実行例
from twitter_scraper import TwitterScraper
from sentiment_analyzer import analyze_tweets_sentiment, plot_sentiment_distribution
# データ取得
scraper = TwitterScraper()
df = scraper.search_tweets('Python programming', max_results=500)
# 感情分析
df = analyze_tweets_sentiment(df)
plot_sentiment_distribution(df)
# 結果
=== 感情分析結果 ===
ポジティブ: 234 (46.8%)
中立: 198 (39.6%)
ネガティブ: 68 (13.6%)
平均極性: 0.1426. ハッシュタグトレンド分析
import re
from collections import Counter
def extract_hashtags(df):
"""ハッシュタグを抽出"""
all_hashtags = []
for text in df['text']:
hashtags = re.findall(r'#(\w+)', text)
all_hashtags.extend([tag.lower() for tag in hashtags])
return Counter(all_hashtags)
def plot_top_hashtags(df, top_n=20):
"""トップハッシュタグのグラフ"""
hashtag_counts = extract_hashtags(df)
top_hashtags = hashtag_counts.most_common(top_n)
if not top_hashtags:
print("ハッシュタグが見つかりませんでした")
return
tags, counts = zip(*top_hashtags)
plt.figure(figsize=(12, 6))
bars = plt.barh(range(len(tags)), counts, color='#1DA1F2')
plt.yticks(range(len(tags)), [f'#{tag}' for tag in tags])
plt.xlabel('出現回数', fontsize=12)
plt.title(f'トップ{top_n}ハッシュタグ', fontsize=14, fontweight='bold')
plt.gca().invert_yaxis()
# 値を表示
for i, (bar, count) in enumerate(zip(bars, counts)):
plt.text(
count,
i,
f' {count}',
va='center',
fontsize=10
)
plt.tight_layout()
plt.savefig('twitter_hashtags.png', dpi=300, bbox_inches='tight')
plt.show()
# 使用例
plot_top_hashtags(df, top_n=15)7. ユーザー分析
def analyze_users(df):
"""ツイートユーザーの分析"""
user_stats = df.groupby('username').agg({
'like_count': 'sum',
'retweet_count': 'sum',
'followers': 'first',
'verified': 'first',
'text': 'count'
}).reset_index()
user_stats.columns = [
'username', 'total_likes', 'total_retweets',
'followers', 'verified', 'tweet_count'
]
# エンゲージメント率
user_stats['engagement_rate'] = (
(user_stats['total_likes'] + user_stats['total_retweets']) /
user_stats['tweet_count']
)
return user_stats.sort_values('engagement_rate', ascending=False)
def plot_user_analysis(user_stats, top_n=15):
"""ユーザー分析グラフ"""
top_users = user_stats.head(top_n)
fig, axes = plt.subplots(1, 2, figsize=(14, 6))
# エンゲージメント率
colors = ['#FFD700' if v else '#1DA1F2' for v in top_users['verified']]
axes[0].barh(range(len(top_users)), top_users['engagement_rate'], color=colors)
axes[0].set_yticks(range(len(top_users)))
axes[0].set_yticklabels(top_users['username'])
axes[0].set_xlabel('平均エンゲージメント', fontsize=11)
axes[0].set_title('ユーザー別エンゲージメント率', fontsize=13, fontweight='bold')
axes[0].invert_yaxis()
# フォロワー数 vs エンゲージメント
axes[1].scatter(
user_stats['followers'],
user_stats['engagement_rate'],
alpha=0.6,
s=100,
c='#1DA1F2'
)
axes[1].set_xlabel('フォロワー数', fontsize=11)
axes[1].set_ylabel('エンゲージメント率', fontsize=11)
axes[1].set_title('フォロワー数 vs エンゲージメント', fontsize=13, fontweight='bold')
axes[1].set_xscale('log')
axes[1].grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('twitter_users.png', dpi=300, bbox_inches='tight')
plt.show()8. 完全な分析パイプライン
#!/usr/bin/env python3
"""Twitter完全分析パイプライン"""
from twitter_scraper import TwitterScraper
from visualizer import *
from sentiment_analyzer import *
import pandas as pd
def full_twitter_analysis(query, max_results=1000):
"""完全なTwitter分析を実行"""
print(f"=== Twitter分析開始: '{query}' ===")
print(f"取得ツイート数: {max_results}\n")
# 1. データ収集
print("[1/6] データ収集中...")
scraper = TwitterScraper()
df = scraper.search_tweets(query, max_results=max_results)
if df is None or len(df) == 0:
print("❌ データ取得失敗")
return None
print(f"✅ {len(df)}件のツイートを取得\n")
# 2. 基本統計
print("[2/6] 基本統計...")
print(f"期間: {df['created_at'].min()} ～ {df['created_at'].max()}")
print(f"平均いいね数: {df['like_count'].mean():.1f}")
print(f"平均リツイート数: {df['retweet_count'].mean():.1f}")
print(f"ユニークユーザー数: {df['username'].nunique()}\n")
# 3. エンゲージメント分析
print("[3/6] エンゲージメント分析...")
plot_engagement_metrics(df)
print("✅ グラフ保存: twitter_engagement.png\n")
# 4. 感情分析
print("[4/6] 感情分析...")
df = analyze_tweets_sentiment(df)
plot_sentiment_distribution(df)
print("✅ グラフ保存: twitter_sentiment.png\n")
# 5. ハッシュタグ分析
print("[5/6] ハッシュタグ分析...")
plot_top_hashtags(df, top_n=20)
print("✅ グラフ保存: twitter_hashtags.png\n")
# 6. ワードクラウド
print("[6/6] ワードクラウド作成...")
create_wordcloud(df)
print("✅ グラフ保存: twitter_wordcloud.png\n")
# CSV保存
output_file = f'twitter_{query.replace(" ", "_")}_analysis.csv'
df.to_csv(output_file, index=False, encoding='utf-8-sig')
print(f"\n📊 結果をCSV保存: {output_file}")
print("\n=== 分析完了 ===")
return df
if __name__ == '__main__':
# 分析実行
df = full_twitter_analysis('Python programming', max_results=500)9. レート制限対策
import time
from datetime import datetime, timedelta
class RateLimitHandler:
"""APIレート制限管理"""
def __init__(self):
self.request_times = []
self.max_requests_per_15min = 180 # Essential tier
def wait_if_needed(self):
"""必要に応じて待機"""
now = datetime.now()
# 15分以内のリクエストを抽出
recent = [t for t in self.request_times
if now - t < timedelta(minutes=15)]
if len(recent) >= self.max_requests_per_15min:
# 最古のリクエストから15分経過まで待機
wait_until = recent[0] + timedelta(minutes=15)
wait_seconds = (wait_until - now).total_seconds()
if wait_seconds > 0:
print(f"⏳ レート制限: {wait_seconds:.0f}秒待機中...")
time.sleep(wait_seconds)
# 現在時刻を記録
self.request_times.append(now)
# 古い記録を削除
self.request_times = [t for t in self.request_times
if now - t < timedelta(minutes=15)]10. エラーハンドリング
from tweepy.errors import TweepyException, Unauthorized, Forbidden
def safe_api_call(func):
"""安全なAPI呼び出し"""
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Unauthorized:
print("❌ 認証エラー: API キーを確認してください")
except Forbidden:
print("❌ アクセス拒否: APIアクセスレベルを確認")
except TweepyException as e:
print(f"❌ Twitter APIエラー: {e}")
except Exception as e:
print(f"❌ 予期しないエラー: {e}")
return None
return wrapperまとめ
Twitter APIでSNSデータを収集・分析できます！
重要ポイント
- ✅ Twitter Developer Portalでアプリ登録必須
- ✅ API v2を使用（tweepyライブラリ）
- ✅ レート制限に注意（Essential: 500K/月）
- ✅ 感情分析でトレンド把握
- ✅ 複数の可視化手法を活用
応用例
- ブランドモニタリング
- 競合分析
- インフルエンサー発掘
- 危機管理（炎上検知）
- マーケティングキャンペーン分析
注意事項
- ⚠️ Twitter利用規約を遵守
- ⚠️ 個人情報の取り扱いに注意
- ⚠️ APIクォータ管理
- ⚠️ ボット判定に注意