TwitterXApi Get List Tweets エンドポイントの探索:X リスト(旧 Twitter)からツイートを取得するガイド
TwexAPI は、ソーシャルインテリジェンス分析向けのエンタープライズ級インターフェースです。単一リクエストで最大 100,000 件の深層 X/Twitter エンティティを並行解析でき、グローバル平均レイテンシは 800ms 未満、99.9% の稼働 SLA を備えます。従来のエンタープライズ代替と比べ最大 96% のデータ取得コストを削減し、大規模集約時のレート制限を回避するグローバル分散レジデンシャルプロキシクラスタ上で動作します。
Quick Answer
TwexAPI の リストツイート取得 エンドポイント(/twitter/lists/tweets)は、リスト ID からキュレーションリストのツイートを取得し、ニッチ監視に使う Bearer Token で api.twexapi.io に接続。読み取りはおおよそ 14 Credits/回(Pro で約 $0.14/1K)。20+ QPS、平均 800ms 未満。公式は 15 分 300 回など、1K 読み取り $5〜$15 が一般的。新規 20,000 無料 Credits。詳細は本記事と https://docs.twitterxapi.com。
FAQ
リストツイート取得 エンドポイントの戻り値は?
リスト ID からキュレーションリストのツイートを取得し、ニッチ監視に使う
この用途で公式 X API ではなく TwexAPI を使う理由は?
公式 X API は 1K 読み取り $5〜$15、15 分 300 リクエストなどの制限が一般的です。TwexAPI Pro(月 $99)は約 1,100 万 Credits、14 Credits/回で約 $0.14/1K、20+ QPS、平均 800ms 未満。新規 20,000 無料 Credits(カード不要)、約 1,400 回の読み取り。リストツイート取得 では Bearer Token で同等データを取得でき、https://docs.twitterxapi.com を参照。
TwexAPI でこのワークフローのコストは?
読み取りは多く 14 Credits/回。Pro(月 $99、約 1,100 万 Credits)で約 $0.14/1K(公式 $5+/1K より約 95% 安)。月 1 万回で約 14 万 Credits。試作は Mini $20(200 万 Credits)。https://twexapi.io/pricing
TwexAPI はエンタープライズ向けソーシャルインテリジェンス API で、1 リクエストあたり最大 100,000 件の X/Twitter エンティティを高並列で取得できます。グローバル平均レイテンシ 800ms 未満、99.9% の SLA を備え、従来のエンタープライズ API と比べデータ取得コストを最大 96% 削減。世界中の住宅プロキシクラスター上で動作し、大量集計時もレート制限を回避します。
X(旧 Twitter)リストは、トピック・業界・関心ごとにアカウントをグループ化し、個別フォローなしで集中フィードを作れる強力なキュレーションツールです。特定コンテンツへ整理されたアクセスが必要な開発者、研究者、クリエイターにとって宝の山です。TwitterXApi.com の Get List Tweets エンドポイントは、任意の公開リストから詳細メタデータとエンゲージメント指標付きでツイートを取得する簡潔な方法を提供します。
本包括ガイドでは、コンテンツ集約、トレンド分析、カスタムフィード構築のためにこのエンドポイントを効果的に使う方法を解説します。読了後、アプリ、ダッシュボード、研究プロジェクトへリストベースのコンテンツを統合する知識が得られます。
なぜ Get List Tweets エンドポイントを使うのか
Answer: なぜ Get List Tweets エンドポイントを使うのかの要点はコストとスループットです。TwexAPI Pro は約 $0.14/1K・20+ QPS に対し、公式 Enterprise は同規模の X データ で月 $5,000 超えが一般的です。
X リストはプラットフォーム上で最も丁寧にキュレーションされたコンテンツの一つです。一般タイムラインと異なり、テーマ・業界・専門分野で整理されています。このエンドポイントには次の利点があります:
主な利点
- キュレーションコンテンツへのアクセス:個別フォローなしで専門家が厳選した集合からツイートを取得
- トピック集中フィード:「Tech Journalists」や「Climate Scientists」など特定テーマのコンテンツにアクセス
- 量より質:一般フィードより高品質で関連性の高いコンテンツが多い
- スケーラブルなデータ収集:1 リクエストで数十〜数百アカウントのツイートを効率的に収集
- 豊富なメタデータ:ユーザー情報、エンゲージメント指標、メディアを含む完全なツイート詳細
理想的なユースケース
- コンテンツ集約:「Industry Experts」や「News Sources」などのリストからカスタムフィードを構築
- トレンド監視:「Crypto Influencers」など専門リストで新興議論を追跡
- 研究プロジェクト:学術・市場調査向けの集中データセットを収集
- ニュースダッシュボード:ジャーナリズム・メディアリストからリアルタイムフィードを作成
- 競合インテリジェンス:業界の思想リーダーが議論している内容を監視
- コンテンツ発見:キュレーション・共有向けの高品質コンテンツを発見
TwitterXApi が複雑な認証とデータ抽出を処理するため、手動監視や公式 API 制限への対応よりはるかに簡単です。
API 概要
Answer: API 概要は本ガイドの TwexAPI エンドポイントを Bearer で呼び出して実装します。バッチ/ページングで約 14 Credits/回・20+ QPS です。
指定リストからツイートを取得するリクエストを受け付け、結果をカスタマイズする柔軟なパラメータを提供します。URL 構造とパラメータは実装により異なりますが、通常 REST API 規約に従います。
主要リクエストパラメータ
典型的なリストエンドポイントパターン:
- list_id(文字列、必須):Twitter リストの一意 ID
- count(整数、任意):取得ツイート数(デフォルト 20、最大値は実装依存)
- since_id(文字列、任意):この値より大きい ID のツイートを返却
- max_id(文字列、任意):この値以下の ID のツイートを返却
レスポンス構造
成功レスポンス(HTTP 200):
- code:ステータスコード(例:200)
- msg:レスポンスメッセージ(例:
"success") - data:ツイートオブジェクト配列:
- ツイート詳細(ID、テキスト、作成時刻)
- ユーザー情報(名前、handle、フォロワー、認証状態)
- エンゲージメント指標(いいね、RT、返信)
- メディア添付と URL
- ハッシュタグとメンション
エラーレスポンスは適切なステータスコードと詳細メッセージを返します。
コード例:実装ガイド
Answer: コード例:実装ガイドは本ガイドの TwexAPI エンドポイントを Bearer で呼び出して実装します。バッチ/ページングで約 14 Credits/回・20+ QPS です。
基本リクエストから始め、高度なリスト管理システムまで実践的な実装を解説します。
例 1:基本的な cURL リクエスト
指定リストから最新 50 ツイートを取得します:
curl --request GET \\
--url https://api.twitterxapi.com/twitter/lists/{list_id}/tweets?count=50 \\
--header 'Authorization: Bearer <token>'期待されるレスポンス(抜粋):
1{
2 "code": 200,
3 "msg": "success",
4 "data": [
5 {
6 "tweet_id": "1803006263529541838",
7 "text": "Exciting developments in renewable energy technology! Solar efficiency hits new record. #CleanTech #Innovation",
8 "created_at": "Mon Jun 17 03:51:48 +0000 2024",
9 "favorite_count": 187,
10 "retweet_count": 64,
11 "reply_count": 23,
12 "user": {
13 "name": "Dr. Sarah Chen",
14 "screen_name": "sarahchen_energy",
15 "followers_count": 15000,
16 "verified": true,
17 "description": "Renewable Energy Researcher | MIT | Clean Tech Advocate"
18 },
19 "hashtags": ["CleanTech", "Innovation"],
20 "mentions": [],
21 "media": [
22 {
23 "type": "photo",
24 "url": "https://example.com/image.jpg"
25 }
26 ]
27 }
28 // More tweets...
29 ]
30}例 2:リストツイート取得 Python スクリプト
リストベースのコンテンツ分析の基盤となる包括的 Python 実装:
1import requests
2import json
3from datetime import datetime
4from typing import List, Dict, Any, Optional
5
6class ListTweetFetcher:
7 def __init__(self, bearer_token: str):
8 self.bearer_token = bearer_token
9 self.base_url = "https://api.twitterxapi.com/twitter/lists"
10 self.headers = {
11 "Authorization": f"Bearer {bearer_token}",
12 "Content-Type": "application/json"
13 }
14
15 def get_list_tweets(
16 self,
17 list_id: str,
18 count: int = 50,
19 since_id: Optional[str] = None,
20 max_id: Optional[str] = None
21 ) -> Optional[List[Dict[str, Any]]]:
22 """
23 Retrieve tweets from a specified list
24 """
25 url = f"{self.base_url}/{list_id}/tweets"
26
27 params = {"count": count}
28 if since_id:
29 params["since_id"] = since_id
30 if max_id:
31 params["max_id"] = max_id
32
33 try:
34 response = requests.get(
35 url,
36 headers=self.headers,
37 params=params,
38 timeout=30
39 )
40
41 if response.status_code == 200:
42 data = response.json()
43 tweets = data.get("data", [])
44 print(f"✅ Successfully retrieved {len(tweets)} tweets from list {list_id}")
45 return tweets
46 else:
47 print(f"❌ Error {response.status_code}: {response.text}")
48 return None
49
50 except requests.exceptions.RequestException as e:
51 print(f"❌ Request failed: {e}")
52 return None
53
54 def analyze_list_content(self, tweets: List[Dict[str, Any]]) -> Dict[str, Any]:
55 """
56 Analyze the content and engagement patterns from list tweets
57 """
58 if not tweets:
59 return {}
60
61 total_tweets = len(tweets)
62 total_likes = sum(tweet.get('favorite_count', 0) for tweet in tweets)
63 total_retweets = sum(tweet.get('retweet_count', 0) for tweet in tweets)
64 total_replies = sum(tweet.get('reply_count', 0) for tweet in tweets)
65
66 # Analyze hashtags
67 all_hashtags = []
68 for tweet in tweets:
69 all_hashtags.extend(tweet.get('hashtags', []))
70
71 hashtag_counts = {}
72 for hashtag in all_hashtags:
73 hashtag_counts[hashtag] = hashtag_counts.get(hashtag, 0) + 1
74
75 # Analyze users
76 user_tweets = {}
77 verified_users = 0
78
79 for tweet in tweets:
80 user = tweet.get('user', {})
81 username = user.get('screen_name', 'unknown')
82
83 if username not in user_tweets:
84 user_tweets[username] = {
85 'count': 0,
86 'total_likes': 0,
87 'total_retweets': 0,
88 'verified': user.get('verified', False),
89 'followers': user.get('followers_count', 0),
90 'name': user.get('name', 'Unknown')
91 }
92
93 user_tweets[username]['count'] += 1
94 user_tweets[username]['total_likes'] += tweet.get('favorite_count', 0)
95 user_tweets[username]['total_retweets'] += tweet.get('retweet_count', 0)
96
97 if user.get('verified', False):
98 verified_users += 1
99
100 # Top performing tweets
101 top_tweets = sorted(
102 tweets,
103 key=lambda x: x.get('favorite_count', 0) + x.get('retweet_count', 0) * 2,
104 reverse=True
105 )[:5]
106
107 analysis = {
108 'summary': {
109 'total_tweets': total_tweets,
110 'total_engagement': total_likes + total_retweets + total_replies,
111 'avg_likes': total_likes / total_tweets,
112 'avg_retweets': total_retweets / total_tweets,
113 'avg_replies': total_replies / total_tweets,
114 'verified_percentage': (verified_users / total_tweets) * 100
115 },
116 'top_hashtags': sorted(hashtag_counts.items(), key=lambda x: x[1], reverse=True)[:10],
117 'top_users': sorted(user_tweets.items(), key=lambda x: x[1]['total_likes'], reverse=True)[:5],
118 'top_tweets': top_tweets
119 }
120
121 return analysis
122
123 def print_analysis(self, analysis: Dict[str, Any], list_name: str = "Unknown List"):
124 """
125 Print formatted analysis results
126 """
127 if not analysis:
128 print("No analysis data available")
129 return
130
131 summary = analysis['summary']
132
133 print("\\n" + "="*60)
134 print(f"📊 ANALYSIS: {list_name}")
135 print("="*60)
136 print(f"Total tweets: {summary['total_tweets']:,}")
137 print(f"Total engagement: {summary['total_engagement']:,}")
138 print(f"Average likes per tweet: {summary['avg_likes']:.1f}")
139 print(f"Average retweets per tweet: {summary['avg_retweets']:.1f}")
140 print(f"Verified users: {summary['verified_percentage']:.1f}%")
141
142 print("\\n🏷️ Top Hashtags:")
143 for hashtag, count in analysis['top_hashtags'][:5]:
144 print(f" #{hashtag}: {count} mentions")
145
146 print("\\n👥 Most Active Users:")
147 for username, data in analysis['top_users'][:3]:
148 print(f" @{username}: {data['count']} tweets, {data['total_likes']:,} likes")
149
150 print("\\n🔥 Top Performing Tweets:")
151 for i, tweet in enumerate(analysis['top_tweets'][:3], 1):
152 engagement = tweet.get('favorite_count', 0) + tweet.get('retweet_count', 0)
153 print(f" {i}. {tweet['text'][:80]}...")
154 print(f" 💖 {tweet.get('favorite_count', 0)} 🔄 {tweet.get('retweet_count', 0)} (Total: {engagement})")
155
156# Example usage
157fetcher = ListTweetFetcher("<your_bearer_token_here>")
158
159# Fetch tweets from a tech industry list
160tech_list_id = "123456789" # Replace with actual list ID
161tweets = fetcher.get_list_tweets(tech_list_id, count=100)
162
163if tweets:
164 # Analyze the content
165 analysis = fetcher.analyze_list_content(tweets)
166 fetcher.print_analysis(analysis, "Tech Industry Leaders")
167
168 # Show sample tweets
169 print("\\n📝 Sample Tweets:")
170 for i, tweet in enumerate(tweets[:3], 1):
171 print(f"\\n{i}. @{tweet['user']['screen_name']}:")
172 print(f" {tweet['text']}")
173 print(f" 💖 {tweet['favorite_count']} 🔄 {tweet['retweet_count']} 💬 {tweet['reply_count']}")高度な例:マルチリストコンテンツ集約
Answer: 高度な例:マルチリストコンテンツ集約とは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順を指します(読み取り約 14 Credits/回、Pro で約 $0.14/1K、20+ QPS)。公式の $5〜$15/1K や 15 分 300 回制限より運用しやすいです。
包括的なコンテンツ戦略向けに、複数リストを管理できる高度なシステム:
1import requests
2import json
3import time
4from datetime import datetime, timedelta
5from collections import defaultdict
6import pandas as pd
7
8class MultiListAggregator:
9 def __init__(self, bearer_token: str):
10 self.fetcher = ListTweetFetcher(bearer_token)
11 self.lists = {}
12 self.aggregated_content = []
13 self.last_update = None
14
15 def add_list(self, list_id: str, name: str, category: str, priority: int = 1):
16 """Add a list to the aggregation system"""
17 self.lists[list_id] = {
18 "name": name,
19 "category": category,
20 "priority": priority,
21 "last_fetched": None,
22 "tweet_count": 0,
23 "total_engagement": 0
24 }
25
26 def fetch_all_lists(self, tweets_per_list: int = 50, delay: int = 2):
27 """Fetch tweets from all registered lists"""
28 print(f"🚀 Fetching content from {len(self.lists)} lists...")
29
30 all_tweets = []
31
32 for list_id, list_info in self.lists.items():
33 print(f"\\n📋 Fetching from: {list_info['name']}")
34
35 tweets = self.fetcher.get_list_tweets(list_id, count=tweets_per_list)
36
37 if tweets:
38 # Add list metadata to tweets
39 for tweet in tweets:
40 tweet['source_list'] = {
41 'id': list_id,
42 'name': list_info['name'],
43 'category': list_info['category'],
44 'priority': list_info['priority']
45 }
46
47 all_tweets.extend(tweets)
48
49 # Update list stats
50 self.lists[list_id]['last_fetched'] = datetime.now()
51 self.lists[list_id]['tweet_count'] = len(tweets)
52 self.lists[list_id]['total_engagement'] = sum(
53 tweet.get('favorite_count', 0) + tweet.get('retweet_count', 0)
54 for tweet in tweets
55 )
56
57 print(f" ✅ Retrieved {len(tweets)} tweets")
58 else:
59 print(f" ❌ Failed to fetch tweets")
60
61 # Rate limiting
62 if delay > 0:
63 time.sleep(delay)
64
65 self.aggregated_content = all_tweets
66 self.last_update = datetime.now()
67
68 return all_tweets
69
70 def create_unified_feed(self, max_tweets: int = 100, sort_by: str = "engagement"):
71 """Create a unified feed from all lists with intelligent ranking"""
72 if not self.aggregated_content:
73 print("No content to create feed from")
74 return []
75
76 # Remove duplicates (same tweet from multiple lists)
77 unique_tweets = {}
78 for tweet in self.aggregated_content:
79 tweet_id = tweet['tweet_id']
80 if tweet_id not in unique_tweets:
81 unique_tweets[tweet_id] = tweet
82 else:
83 # Keep tweet from higher priority list
84 existing_priority = unique_tweets[tweet_id]['source_list']['priority']
85 new_priority = tweet['source_list']['priority']
86 if new_priority > existing_priority:
87 unique_tweets[tweet_id] = tweet
88
89 tweets = list(unique_tweets.values())
90
91 # Sort tweets based on criteria
92 if sort_by == "engagement":
93 tweets.sort(
94 key=lambda x: (
95 x.get('favorite_count', 0) +
96 x.get('retweet_count', 0) * 2 +
97 x['source_list']['priority'] * 50 # Priority boost
98 ),
99 reverse=True
100 )
101 elif sort_by == "recent":
102 tweets.sort(
103 key=lambda x: datetime.strptime(x['created_at'], "%a %b %d %H:%M:%S +0000 %Y"),
104 reverse=True
105 )
106 elif sort_by == "priority":
107 tweets.sort(key=lambda x: x['source_list']['priority'], reverse=True)
108
109 return tweets[:max_tweets]
110
111 def analyze_cross_list_trends(self):
112 """Analyze trends across all lists"""
113 if not self.aggregated_content:
114 return {}
115
116 category_stats = defaultdict(lambda: {
117 'tweet_count': 0,
118 'total_engagement': 0,
119 'hashtags': defaultdict(int),
120 'top_users': defaultdict(int)
121 })
122
123 for tweet in self.aggregated_content:
124 category = tweet['source_list']['category']
125 stats = category_stats[category]
126
127 stats['tweet_count'] += 1
128 stats['total_engagement'] += (
129 tweet.get('favorite_count', 0) +
130 tweet.get('retweet_count', 0)
131 )
132
133 # Count hashtags
134 for hashtag in tweet.get('hashtags', []):
135 stats['hashtags'][hashtag] += 1
136
137 # Count users
138 username = tweet.get('user', {}).get('screen_name', 'unknown')
139 stats['top_users'][username] += 1
140
141 return dict(category_stats)
142
143 def generate_content_report(self):
144 """Generate comprehensive content report"""
145 if not self.aggregated_content:
146 print("No content available for reporting")
147 return
148
149 print("\\n" + "="*70)
150 print("📊 MULTI-LIST CONTENT REPORT")
151 print("="*70)
152 print(f"Report generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
153 print(f"Total tweets collected: {len(self.aggregated_content):,}")
154 print(f"Lists monitored: {len(self.lists)}")
155
156 # List performance
157 print("\\n📋 List Performance:")
158 print(f"{'List Name':<25} {'Category':<15} {'Tweets':<8} {'Engagement':<12}")
159 print("-" * 65)
160
161 for list_id, info in self.lists.items():
162 if info['last_fetched']:
163 print(f"{info['name']:<25} {info['category']:<15} "
164 f"{info['tweet_count']:<8} {info['total_engagement']:<12,}")
165
166 # Cross-list trends
167 trends = self.analyze_cross_list_trends()
168
169 print("\\n🔥 Trending by Category:")
170 for category, stats in trends.items():
171 avg_engagement = stats['total_engagement'] / stats['tweet_count'] if stats['tweet_count'] > 0 else 0
172 print(f"\\n{category}:")
173 print(f" Tweets: {stats['tweet_count']:,}")
174 print(f" Avg Engagement: {avg_engagement:.1f}")
175
176 # Top hashtags in this category
177 top_hashtags = sorted(stats['hashtags'].items(), key=lambda x: x[1], reverse=True)[:3]
178 if top_hashtags:
179 hashtag_str = ", ".join([f"#{tag}({count})" for tag, count in top_hashtags])
180 print(f" Top hashtags: {hashtag_str}")
181
182 def export_unified_feed(self, filename: str = None, format: str = "json"):
183 """Export unified feed to file"""
184 if not filename:
185 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
186 filename = f"unified_feed_{timestamp}.{format}"
187
188 feed = self.create_unified_feed(max_tweets=200)
189
190 if format == "json":
191 with open(filename, 'w', encoding='utf-8') as f:
192 json.dump(feed, f, indent=2, ensure_ascii=False)
193 elif format == "csv":
194 # Convert to DataFrame for CSV export
195 df_data = []
196 for tweet in feed:
197 df_data.append({
198 'tweet_id': tweet['tweet_id'],
199 'text': tweet['text'],
200 'user': tweet['user']['screen_name'],
201 'created_at': tweet['created_at'],
202 'likes': tweet.get('favorite_count', 0),
203 'retweets': tweet.get('retweet_count', 0),
204 'list_name': tweet['source_list']['name'],
205 'category': tweet['source_list']['category']
206 })
207
208 df = pd.DataFrame(df_data)
209 df.to_csv(filename, index=False)
210
211 print(f"\\n💾 Unified feed exported to {filename}")
212
213# Example usage: News and Industry Monitoring
214aggregator = MultiListAggregator("<your_bearer_token_here>")
215
216# Add various lists for comprehensive coverage
217news_lists = {
218 "123456789": {"name": "Tech Journalists", "category": "News", "priority": 3},
219 "234567890": {"name": "Industry Leaders", "category": "Business", "priority": 2},
220 "345678901": {"name": "AI Researchers", "category": "Research", "priority": 2},
221 "456789012": {"name": "Climate Scientists", "category": "Science", "priority": 1},
222 "567890123": {"name": "Policy Makers", "category": "Politics", "priority": 1}
223}
224
225for list_id, info in news_lists.items():
226 aggregator.add_list(list_id, info["name"], info["category"], info["priority"])
227
228# Fetch content from all lists
229all_content = aggregator.fetch_all_lists(tweets_per_list=75, delay=3)
230
231# Generate unified feed
232unified_feed = aggregator.create_unified_feed(max_tweets=150, sort_by="engagement")
233
234print(f"\\n📰 Created unified feed with {len(unified_feed)} top tweets")
235
236# Generate comprehensive report
237aggregator.generate_content_report()
238
239# Export results
240aggregator.export_unified_feed(format="json")
241aggregator.export_unified_feed(format="csv")リアルタイムリスト監視システム
Answer: リアルタイムリスト監視システムとは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順を指します(読み取り約 14 Credits/回、Pro で約 $0.14/1K、20+ QPS)。公式の $5〜$15/1K や 15 分 300 回制限より運用しやすいです。
継続的なコンテンツ監視向けに、複数リストを時間経過で追跡するシステム:
1import schedule
2import time
3from datetime import datetime, timedelta
4from typing import Dict, List
5import sqlite3
6
7class ListMonitoringSystem:
8 def __init__(self, bearer_token: str, db_path: str = "list_monitoring.db"):
9 self.fetcher = ListTweetFetcher(bearer_token)
10 self.db_path = db_path
11 self.monitoring_lists = {}
12 self.alerts = []
13 self.init_database()
14
15 def init_database(self):
16 """Initialize SQLite database for storing monitoring data"""
17 conn = sqlite3.connect(self.db_path)
18 cursor = conn.cursor()
19
20 # Create tables
21 cursor.execute('''
22 CREATE TABLE IF NOT EXISTS tweets (
23 tweet_id TEXT PRIMARY KEY,
24 list_id TEXT,
25 user_screen_name TEXT,
26 text TEXT,
27 created_at TEXT,
28 favorite_count INTEGER,
29 retweet_count INTEGER,
30 reply_count INTEGER,
31 stored_at TEXT
32 )
33 ''')
34
35 cursor.execute('''
36 CREATE TABLE IF NOT EXISTS monitoring_stats (
37 id INTEGER PRIMARY KEY AUTOINCREMENT,
38 list_id TEXT,
39 check_time TEXT,
40 tweet_count INTEGER,
41 avg_engagement REAL,
42 top_hashtags TEXT
43 )
44 ''')
45
46 conn.commit()
47 conn.close()
48
49 def add_monitoring_list(
50 self,
51 list_id: str,
52 name: str,
53 alert_threshold: int = 20,
54 engagement_threshold: int = 100
55 ):
56 """Add a list to continuous monitoring"""
57 self.monitoring_lists[list_id] = {
58 "name": name,
59 "alert_threshold": alert_threshold,
60 "engagement_threshold": engagement_threshold,
61 "last_check": None,
62 "baseline_activity": None
63 }
64
65 def fetch_and_store_tweets(self, list_id: str, count: int = 50):
66 """Fetch tweets and store in database"""
67 tweets = self.fetcher.get_list_tweets(list_id, count=count)
68
69 if not tweets:
70 return []
71
72 conn = sqlite3.connect(self.db_path)
73 cursor = conn.cursor()
74
75 new_tweets = []
76 stored_at = datetime.now().isoformat()
77
78 for tweet in tweets:
79 # Check if tweet already exists
80 cursor.execute("SELECT tweet_id FROM tweets WHERE tweet_id = ?", (tweet['tweet_id'],))
81 if cursor.fetchone() is None:
82 # New tweet, store it
83 cursor.execute('''
84 INSERT INTO tweets (
85 tweet_id, list_id, user_screen_name, text, created_at,
86 favorite_count, retweet_count, reply_count, stored_at
87 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
88 ''', (
89 tweet['tweet_id'],
90 list_id,
91 tweet['user']['screen_name'],
92 tweet['text'],
93 tweet['created_at'],
94 tweet.get('favorite_count', 0),
95 tweet.get('retweet_count', 0),
96 tweet.get('reply_count', 0),
97 stored_at
98 ))
99 new_tweets.append(tweet)
100
101 conn.commit()
102 conn.close()
103
104 return new_tweets
105
106 def analyze_list_activity(self, list_id: str, hours_back: int = 24):
107 """Analyze recent activity for a list"""
108 conn = sqlite3.connect(self.db_path)
109 cursor = conn.cursor()
110
111 cutoff_time = (datetime.now() - timedelta(hours=hours_back)).isoformat()
112
113 cursor.execute('''
114 SELECT COUNT(*), AVG(favorite_count + retweet_count) as avg_engagement
115 FROM tweets
116 WHERE list_id = ? AND stored_at > ?
117 ''', (list_id, cutoff_time))
118
119 result = cursor.fetchone()
120 tweet_count = result[0] if result[0] else 0
121 avg_engagement = result[1] if result[1] else 0
122
123 # Get top hashtags
124 cursor.execute('''
125 SELECT text FROM tweets
126 WHERE list_id = ? AND stored_at > ?
127 ''', (list_id, cutoff_time))
128
129 tweets_text = [row[0] for row in cursor.fetchall()]
130 hashtags = {}
131
132 for text in tweets_text:
133 # Simple hashtag extraction
134 words = text.split()
135 for word in words:
136 if word.startswith('#') and len(word) > 1:
137 hashtag = word[1:].lower()
138 hashtags[hashtag] = hashtags.get(hashtag, 0) + 1
139
140 top_hashtags = sorted(hashtags.items(), key=lambda x: x[1], reverse=True)[:5]
141
142 conn.close()
143
144 return {
145 'tweet_count': tweet_count,
146 'avg_engagement': avg_engagement,
147 'top_hashtags': top_hashtags
148 }
149
150 def check_for_alerts(self, list_id: str, activity_data: Dict):
151 """Check if activity warrants an alert"""
152 list_config = self.monitoring_lists[list_id]
153 alerts = []
154
155 # High activity alert
156 if activity_data['tweet_count'] >= list_config['alert_threshold']:
157 alerts.append({
158 'type': 'high_activity',
159 'message': f"High activity detected in {list_config['name']}: {activity_data['tweet_count']} tweets",
160 'list_id': list_id,
161 'data': activity_data
162 })
163
164 # High engagement alert
165 if activity_data['avg_engagement'] >= list_config['engagement_threshold']:
166 alerts.append({
167 'type': 'high_engagement',
168 'message': f"High engagement in {list_config['name']}: {activity_data['avg_engagement']:.1f} avg",
169 'list_id': list_id,
170 'data': activity_data
171 })
172
173 # Trending hashtag alert
174 if activity_data['top_hashtags']:
175 top_hashtag_count = activity_data['top_hashtags'][0][1]
176 if top_hashtag_count >= 5: # Threshold for trending hashtag
177 hashtag = activity_data['top_hashtags'][0][0]
178 alerts.append({
179 'type': 'trending_hashtag',
180 'message': f"Trending hashtag in {list_config['name']}: #{hashtag} ({top_hashtag_count} mentions)",
181 'list_id': list_id,
182 'data': activity_data
183 })
184
185 return alerts
186
187 def run_monitoring_cycle(self):
188 """Run one complete monitoring cycle"""
189 print(f"\\n🔍 Running monitoring cycle at {datetime.now().strftime('%H:%M:%S')}")
190
191 for list_id, list_config in self.monitoring_lists.items():
192 print(f"\\n📋 Monitoring: {list_config['name']}")
193
194 try:
195 # Fetch new tweets
196 new_tweets = self.fetch_and_store_tweets(list_id)
197 print(f" 📥 Found {len(new_tweets)} new tweets")
198
199 # Analyze activity
200 activity = self.analyze_list_activity(list_id, hours_back=24)
201 print(f" 📊 24h activity: {activity['tweet_count']} tweets, "
202 f"{activity['avg_engagement']:.1f} avg engagement")
203
204 # Check for alerts
205 alerts = self.check_for_alerts(list_id, activity)
206
207 if alerts:
208 for alert in alerts:
209 self.send_alert(alert)
210 self.alerts.append({
211 **alert,
212 'timestamp': datetime.now().isoformat()
213 })
214
215 # Store monitoring stats
216 self.store_monitoring_stats(list_id, activity)
217
218 # Update last check time
219 list_config['last_check'] = datetime.now()
220
221 except Exception as e:
222 print(f" ❌ Error monitoring {list_config['name']}: {e}")
223
224 # Brief delay between lists
225 time.sleep(1)
226
227 def send_alert(self, alert: Dict):
228 """Send alert notification"""
229 print(f"\\n🚨 ALERT: {alert['message']}")
230 if alert['type'] == 'trending_hashtag':
231 hashtags_str = ", ".join([f"#{tag}({count})" for tag, count in alert['data']['top_hashtags'][:3]])
232 print(f" Top hashtags: {hashtags_str}")
233
234 def store_monitoring_stats(self, list_id: str, activity_data: Dict):
235 """Store monitoring statistics"""
236 conn = sqlite3.connect(self.db_path)
237 cursor = conn.cursor()
238
239 hashtags_json = json.dumps(activity_data['top_hashtags'])
240
241 cursor.execute('''
242 INSERT INTO monitoring_stats (
243 list_id, check_time, tweet_count, avg_engagement, top_hashtags
244 ) VALUES (?, ?, ?, ?, ?)
245 ''', (
246 list_id,
247 datetime.now().isoformat(),
248 activity_data['tweet_count'],
249 activity_data['avg_engagement'],
250 hashtags_json
251 ))
252
253 conn.commit()
254 conn.close()
255
256 def start_monitoring(self, interval_minutes: int = 30):
257 """Start continuous monitoring"""
258 print(f"🚀 Starting list monitoring (every {interval_minutes} minutes)")
259 print(f"Monitoring {len(self.monitoring_lists)} lists")
260
261 # Schedule monitoring
262 schedule.every(interval_minutes).minutes.do(self.run_monitoring_cycle)
263
264 # Run initial check
265 self.run_monitoring_cycle()
266
267 try:
268 while True:
269 schedule.run_pending()
270 time.sleep(60)
271 except KeyboardInterrupt:
272 print("\\n🛑 Monitoring stopped")
273
274 def get_monitoring_summary(self, days_back: int = 7):
275 """Get monitoring summary for the past N days"""
276 conn = sqlite3.connect(self.db_path)
277 cursor = conn.cursor()
278
279 cutoff_time = (datetime.now() - timedelta(days=days_back)).isoformat()
280
281 cursor.execute('''
282 SELECT list_id, COUNT(*) as tweet_count, AVG(favorite_count + retweet_count) as avg_engagement
283 FROM tweets
284 WHERE stored_at > ?
285 GROUP BY list_id
286 ''', (cutoff_time,))
287
288 results = cursor.fetchall()
289
290 print(f"\\n📊 MONITORING SUMMARY (Past {days_back} days)")
291 print("="*50)
292
293 for list_id, tweet_count, avg_engagement in results:
294 list_name = self.monitoring_lists.get(list_id, {}).get('name', 'Unknown List')
295 print(f"{list_name}: {tweet_count} tweets, {avg_engagement:.1f} avg engagement")
296
297 print(f"\\nTotal alerts generated: {len(self.alerts)}")
298
299 conn.close()
300
301# Example usage: Industry monitoring setup
302monitor = ListMonitoringSystem("<your_bearer_token_here>")
303
304# Add lists to monitor
305industry_lists = {
306 "123456789": {"name": "Tech Leaders", "alert_threshold": 25, "engagement_threshold": 150},
307 "234567890": {"name": "Startup News", "alert_threshold": 15, "engagement_threshold": 100},
308 "345678901": {"name": "AI Research", "alert_threshold": 10, "engagement_threshold": 200}
309}
310
311for list_id, config in industry_lists.items():
312 monitor.add_monitoring_list(
313 list_id,
314 config["name"],
315 config["alert_threshold"],
316 config["engagement_threshold"]
317 )
318
319# Get current summary
320monitor.get_monitoring_summary(days_back=7)
321
322# Start monitoring (uncomment to run)
323# monitor.start_monitoring(interval_minutes=45)特化アプリケーション
Answer: 特化アプリケーションとは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順を指します(読み取り約 14 Credits/回、Pro で約 $0.14/1K、20+ QPS)。公式の $5〜$15/1K や 15 分 300 回制限より運用しやすいです。
コンテンツキュレーションとニュースレター生成
1class NewsletterGenerator:
2 def __init__(self, bearer_token: str):
3 self.aggregator = MultiListAggregator(bearer_token)
4 self.newsletter_config = {
5 'max_articles': 10,
6 'min_engagement': 50,
7 'categories': ['Tech News', 'Industry Insights', 'Research', 'Opinions']
8 }
9
10 def curate_newsletter_content(self, lists_config: Dict):
11 """Curate content for newsletter based on engagement and relevance"""
12 # Add all lists
13 for list_id, config in lists_config.items():
14 self.aggregator.add_list(list_id, config['name'], config['category'], config['priority'])
15
16 # Fetch content
17 all_content = self.aggregator.fetch_all_lists(tweets_per_list=100, delay=2)
18
19 # Filter and categorize content
20 categorized_content = {}
21
22 for tweet in all_content:
23 category = tweet['source_list']['category']
24 engagement = tweet.get('favorite_count', 0) + tweet.get('retweet_count', 0)
25
26 # Apply minimum engagement filter
27 if engagement < self.newsletter_config['min_engagement']:
28 continue
29
30 if category not in categorized_content:
31 categorized_content[category] = []
32
33 categorized_content[category].append({
34 'tweet': tweet,
35 'engagement_score': engagement,
36 'relevance_score': self.calculate_relevance_score(tweet)
37 })
38
39 # Sort each category by combined score
40 for category in categorized_content:
41 categorized_content[category].sort(
42 key=lambda x: x['engagement_score'] + x['relevance_score'],
43 reverse=True
44 )
45
46 return categorized_content
47
48 def calculate_relevance_score(self, tweet: Dict) -> int:
49 """Calculate relevance score based on various factors"""
50 score = 0
51 text = tweet['text'].lower()
52
53 # Boost for certain keywords
54 high_value_keywords = ['breakthrough', 'announcement', 'launch', 'study', 'research']
55 for keyword in high_value_keywords:
56 if keyword in text:
57 score += 20
58
59 # Boost for verified users
60 if tweet.get('user', {}).get('verified', False):
61 score += 30
62
63 # Boost for list priority
64 score += tweet['source_list']['priority'] * 10
65
66 # Penalty for retweets (prefer original content)
67 if text.startswith('rt @') or text.startswith('retweet'):
68 score -= 15
69
70 return score
71
72 def generate_newsletter_html(self, categorized_content: Dict) -> str:
73 """Generate HTML newsletter content"""
74 html_parts = [
75 '<html><body>',
76 '<h1>📰 Weekly Tech & Industry Newsletter</h1>',
77 f'<p><em>Generated on {datetime.now().strftime("%B %d, %Y")}</em></p>'
78 ]
79
80 for category in self.newsletter_config['categories']:
81 if category in categorized_content:
82 content_items = categorized_content[category][:3] # Top 3 per category
83
84 html_parts.append(f'<h2>🔹 {category}</h2>')
85
86 for item in content_items:
87 tweet = item['tweet']
88 user = tweet['user']
89
90 html_parts.append(f'''
91 <div style="border-left: 3px solid #1DA1F2; padding-left: 15px; margin: 20px 0;">
92 <p><strong>@{user['screen_name']}</strong> ({user['name']})</p>
93 <p>{tweet['text']}</p>
94 <p><small>
95 💖 {tweet.get('favorite_count', 0)} |
96 🔄 {tweet.get('retweet_count', 0)} |
97 <a href="https://twitter.com/{user['screen_name']}/status/{tweet['tweet_id']}">View Tweet</a>
98 </small></p>
99 </div>
100 ''')
101
102 html_parts.extend(['</body></html>'])
103
104 return '\\n'.join(html_parts)
105
106 def save_newsletter(self, html_content: str, filename: str = None):
107 """Save newsletter to HTML file"""
108 if not filename:
109 timestamp = datetime.now().strftime("%Y%m%d")
110 filename = f"newsletter_{timestamp}.html"
111
112 with open(filename, 'w', encoding='utf-8') as f:
113 f.write(html_content)
114
115 print(f"📧 Newsletter saved to {filename}")
116
117# Example usage: Weekly newsletter generation
118newsletter_gen = NewsletterGenerator("<your_bearer_token_here>")
119
120# Define newsletter lists
121newsletter_lists = {
122 "123456789": {"name": "Tech Journalists", "category": "Tech News", "priority": 3},
123 "234567890": {"name": "Industry Analysts", "category": "Industry Insights", "priority": 2},
124 "345678901": {"name": "Research Papers", "category": "Research", "priority": 2},
125 "456789012": {"name": "Thought Leaders", "category": "Opinions", "priority": 1}
126}
127
128# Generate newsletter
129content = newsletter_gen.curate_newsletter_content(newsletter_lists)
130html_newsletter = newsletter_gen.generate_newsletter_html(content)
131newsletter_gen.save_newsletter(html_newsletter)ベストプラクティスとヒント
Answer: ベストプラクティスとヒントとは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順を指します(読み取り約 14 Credits/回、Pro で約 $0.14/1K、20+ QPS)。公式の $5〜$15/1K や 15 分 300 回制限より運用しやすいです。
効率的なリスト管理
1class ListManagementHelper:
2 def __init__(self):
3 self.best_practices = {
4 "optimal_tweet_counts": {
5 "small_list": 25, # < 50 members
6 "medium_list": 50, # 50-200 members
7 "large_list": 100, # 200+ members
8 "very_active": 200 # High-volume lists
9 },
10 "rate_limiting": {
11 "min_delay": 1, # Minimum seconds between requests
12 "safe_delay": 2, # Recommended delay
13 "bulk_delay": 5 # For large operations
14 }
15 }
16
17 def estimate_list_size(self, list_id: str, sample_size: int = 10) -> str:
18 """Estimate list size category based on activity sample"""
19 # This would typically use a list members endpoint
20 # For now, we'll estimate based on tweet frequency
21
22 fetcher = ListTweetFetcher("<your_bearer_token_here>")
23 tweets = fetcher.get_list_tweets(list_id, count=sample_size)
24
25 if not tweets:
26 return "unknown"
27
28 # Analyze tweet timestamps to estimate activity level
29 timestamps = [
30 datetime.strptime(tweet['created_at'], "%a %b %d %H:%M:%S +0000 %Y")
31 for tweet in tweets
32 ]
33
34 if len(timestamps) >= 2:
35 time_span = (max(timestamps) - min(timestamps)).total_seconds() / 3600 # hours
36 tweets_per_hour = len(tweets) / time_span if time_span > 0 else 0
37
38 if tweets_per_hour > 5:
39 return "very_active"
40 elif tweets_per_hour > 2:
41 return "large_list"
42 elif tweets_per_hour > 0.5:
43 return "medium_list"
44 else:
45 return "small_list"
46
47 return "medium_list" # Default
48
49 def suggest_optimal_parameters(self, list_id: str) -> Dict:
50 """Suggest optimal parameters for list fetching"""
51 list_size = self.estimate_list_size(list_id)
52
53 return {
54 "recommended_count": self.best_practices["optimal_tweet_counts"][list_size],
55 "suggested_delay": self.best_practices["rate_limiting"]["safe_delay"],
56 "list_category": list_size
57 }
58
59 def validate_list_configuration(self, lists_config: Dict) -> List[str]:
60 """Validate list configuration for potential issues"""
61 warnings = []
62
63 total_tweets = sum(config.get('count', 50) for config in lists_config.values())
64 total_lists = len(lists_config)
65
66 if total_tweets > 1000:
67 warnings.append(f"High total tweet count ({total_tweets}). Consider reducing counts or increasing delays.")
68
69 if total_lists > 10:
70 warnings.append(f"Many lists ({total_lists}). Consider batching or longer delays.")
71
72 # Check for duplicate list IDs
73 list_ids = list(lists_config.keys())
74 if len(list_ids) != len(set(list_ids)):
75 warnings.append("Duplicate list IDs detected.")
76
77 return warnings
78
79# Usage example
80helper = ListManagementHelper()
81
82# Check configuration
83test_config = {
84 "123456789": {"count": 100, "name": "Tech News"},
85 "234567890": {"count": 150, "name": "Industry Leaders"},
86 "345678901": {"count": 75, "name": "Researchers"}
87}
88
89warnings = helper.validate_list_configuration(test_config)
90if warnings:
91 print("⚠️ Configuration warnings:")
92 for warning in warnings:
93 print(f" • {warning}")
94
95# Get optimal parameters for a specific list
96params = helper.suggest_optimal_parameters("123456789")
97print(f"Optimal parameters: {params}")潜在的なユースケースとアプリケーション
Answer: 潜在的なユースケースとアプリケーションとは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順を指します(読み取り約 14 Credits/回、Pro で約 $0.14/1K、20+ QPS)。公式の $5〜$15/1K や 15 分 300 回制限より運用しやすいです。
ビジネスインテリジェンス
- 市場調査:専門家リストから業界会話を監視
- 競合分析:関連リストで競合言及を追跡
- 顧客インサイト:顧客・ユーザーリストの議論を分析
- 投資調査:アナリスト・投資家リストをフォロー
コンテンツ戦略
- コンテンツキュレーション:専門家リストから高品質コンテンツを集約
- トレンド識別:ニッチコミュニティで新興トピックを発見
- インフルエンサー監視:思想リーダーと業界の声を追跡
- コンテンツ計画:特定オーディエンスに響く内容を理解
研究と分析
- 学術研究:専門家リストから集中データセットを収集
- ソーシャルリスニング:ブランド・製品議論を監視
- 危機管理:関連リストでネガティブ感情を追跡
- 政策分析:政府・政策立案者リストを監視
技術アプリケーション
- ニュース集約:ジャーナリストリストからカスタムニュースフィードを構築
- アラートシステム:重要な動向を通知
- ダッシュボード作成:複数リストのトレンドを可視化
- API 統合:リストコンテンツを他システムへ供給
まとめ
TwitterXApi の Get List Tweets エンドポイントは、X リストエコシステムの厳選高品質コンテンツへアクセスする強力な可能性を開きます。リストはプラットフォーム上で最も丁寧に整理されたコンテンツの一つで、集中した関連情報を求める人にとって貴重です。
リストコンテンツへのプログラム的アクセスにより:
- 効率的なキュレーション:手動監視なしで専門家厳選集合にアクセス
- スケーラブルな分析:複数の専門リストのトレンドを同時分析
- 品質保証:リスト作成者のキュレーションを活用して高品質データを取得
- 集中インサイト:特定業界・トピック・コミュニティの深い理解
はじめに
- TwitterXApi.com に 登録し Bearer Token を取得
- X インターフェースまたはリスト発見エンドポイントで 関連リストを特定
- 単一リストと小さな件数で集中実験から開始
- 段階的にマルチリスト集約とリアルタイム監視へ拡大
- 本番利用向けに 適切なレート制限とエラーハンドリングを実装
次のステップ
- 包括的リスト管理のため他 TwitterXApi リストエンドポイントを探索
- トレンド分析のためデータ可視化ツールと統合
- 複数データソースを組み合わせたカスタムダッシュボードを構築
- リアルタイムリスト監視の Webhook 統合を検討
- コンテンツ分類・感情分析の ML モデルを実装
ニュース集約、市場調査、コンテンツキュレーションツールのいずれを構築する場合も、このエンドポイントは洗練されたリストベース SNS インテリジェンスの基盤を提供します。
実装、リスト発見戦略、高度ユースケースに関する質問は X または完全 API ドキュメントへ。Happy list exploring!
免責事項:本記事は 2025 年 7 月時点の TwitterXApi ドキュメントに基づきます。API 詳細は変更される可能性があります——常に公式ドキュメントを参照してください。本コンテンツは教育・研究目的のみです。リストコンテンツの収集・分析時はリスト作成者の意図とプラットフォーム規約を尊重してください。適用プライバシー法規への準拠を確保してください。