curl --request POST \\
  --url https://api.twitterxapi.com/twitter/community/search-tweets \\
  --header 'Authorization: Bearer <token>' \\
  --header 'Content-Type: application/json' \\
  --data '{
    "community_id": "1234567890123456789",
    "target_count": 100,
    "query": "AI"
  }'

1{
2  "code": 200,
3  "msg": "success",
4  "data": [
5    {
6      "tweet_id": "1803006263529541838",
7      "text": "Excited about the latest AI advancements in computer vision! The progress is incredible. #AI #MachineLearning",
8      "created_at": "Mon Jun 17 03:51:48 +0000 2024",
9      "favorite_count": 123,
10      "retweet_count": 45,
11      "reply_count": 12,
12      "user": {
13        "name": "AI Researcher",
14        "screen_name": "ai_researcher_2024",
15        "followers_count": 10000,
16        "verified": true
17      },
18      "hashtags": ["AI", "MachineLearning"],
19      "mentions": [],
20      "media": []
21    }
22    // More tweets...
23  ]
24}

1import requests
2import json
3from datetime import datetime
4
5class CommunityTweetSearcher:
6    def __init__(self, bearer_token):
7        self.bearer_token = bearer_token
8        self.base_url = "https://api.twitterxapi.com/twitter/community/search-tweets"
9        self.headers = {
10            "Authorization": f"Bearer {bearer_token}",
11            "Content-Type": "application/json"
12        }
13    
14    def search_community_tweets(self, community_id, query, target_count=50):
15        """
16        Search for tweets within a specific community
17        """
18        payload = {
19            "community_id": community_id,
20            "target_count": target_count,
21            "query": query
22        }
23        
24        try:
25            response = requests.post(
26                self.base_url,
27                headers=self.headers,
28                data=json.dumps(payload),
29                timeout=30
30            )
31            
32            if response.status_code == 200:
33                data = response.json()
34                tweets = data.get("data", [])
35                print(f"✅ Search successful! Found {len(tweets)} tweets.")
36                return tweets
37            else:
38                print(f"❌ Error: {response.status_code}")
39                print(f"Response: {response.text}")
40                return None
41                
42        except requests.exceptions.RequestException as e:
43            print(f"❌ Request failed: {e}")
44            return None
45    
46    def analyze_tweets(self, tweets):
47        """
48        Analyze the retrieved tweets for insights
49        """
50        if not tweets:
51            print("No tweets to analyze")
52            return None
53        
54        total_tweets = len(tweets)
55        total_likes = sum(tweet.get('favorite_count', 0) for tweet in tweets)
56        total_retweets = sum(tweet.get('retweet_count', 0) for tweet in tweets)
57        total_replies = sum(tweet.get('reply_count', 0) for tweet in tweets)
58        
59        # Extract hashtags
60        all_hashtags = []
61        for tweet in tweets:
62            all_hashtags.extend(tweet.get('hashtags', []))
63        
64        hashtag_counts = {}
65        for hashtag in all_hashtags:
66            hashtag_counts[hashtag] = hashtag_counts.get(hashtag, 0) + 1
67        
68        # Top users by followers
69        users = [(tweet['user']['screen_name'], tweet['user']['followers_count']) 
70                for tweet in tweets if 'user' in tweet]
71        top_users = sorted(users, key=lambda x: x[1], reverse=True)[:5]
72        
73        analysis = {
74            'total_tweets': total_tweets,
75            'total_engagement': total_likes + total_retweets + total_replies,
76            'avg_likes': total_likes / total_tweets if total_tweets > 0 else 0,
77            'avg_retweets': total_retweets / total_tweets if total_tweets > 0 else 0,
78            'avg_replies': total_replies / total_tweets if total_tweets > 0 else 0,
79            'top_hashtags': sorted(hashtag_counts.items(), key=lambda x: x[1], reverse=True)[:10],
80            'top_users': top_users
81        }
82        
83        return analysis
84    
85    def print_analysis(self, analysis):
86        """
87        Print analysis results in a formatted way
88        """
89        if not analysis:
90            return
91        
92        print("\\n" + "="*50)
93        print("📊 COMMUNITY TWEET ANALYSIS")
94        print("="*50)
95        print(f"Total tweets: {analysis['total_tweets']:,}")
96        print(f"Total engagement: {analysis['total_engagement']:,}")
97        print(f"Average likes per tweet: {analysis['avg_likes']:.1f}")
98        print(f"Average retweets per tweet: {analysis['avg_retweets']:.1f}")
99        print(f"Average replies per tweet: {analysis['avg_replies']:.1f}")
100        
101        print("\\n🏷️ Top Hashtags:")
102        for hashtag, count in analysis['top_hashtags'][:5]:
103            print(f"  #{hashtag}: {count} mentions")
104        
105        print("\\n👥 Top Users (by followers):")
106        for username, followers in analysis['top_users'][:3]:
107            print(f"  @{username}: {followers:,} followers")
108
109# Example usage
110searcher = CommunityTweetSearcher("<your_bearer_token_here>")
111
112# Search for Python-related tweets in a tech community
113community_id = "1234567890123456789"  # Replace with actual community ID
114query = "Python"
115target_count = 100
116
117tweets = searcher.search_community_tweets(community_id, query, target_count)
118
119if tweets:
120    # Print first few tweets
121    print("\\n📝 Sample tweets:")
122    for i, tweet in enumerate(tweets[:3], 1):
123        print(f"\\n{i}. @{tweet['user']['screen_name']}:")
124        print(f"   {tweet['text'][:100]}...")
125        print(f"   💖 {tweet['favorite_count']} 🔄 {tweet['retweet_count']} 💬 {tweet['reply_count']}")
126    
127    # Analyze all tweets
128    analysis = searcher.analyze_tweets(tweets)
129    searcher.print_analysis(analysis)

1import requests
2import json
3import time
4import pandas as pd
5from collections import defaultdict
6from datetime import datetime, timedelta
7
8class MultiCommunityAnalyzer:
9    def __init__(self, bearer_token):
10        self.searcher = CommunityTweetSearcher(bearer_token)
11        self.communities = {}
12        self.all_results = []
13    
14    def add_community(self, community_id, name, description=""):
15        """Add a community to track"""
16        self.communities[community_id] = {
17            "name": name,
18            "description": description,
19            "tweets": []
20        }
21    
22    def search_all_communities(self, query, target_count=50, delay=2):
23        """Search the same query across all registered communities"""
24        print(f"🔍 Searching for '{query}' across {len(self.communities)} communities...")
25        
26        results = {}
27        
28        for community_id, community_info in self.communities.items():
29            print(f"\\n📍 Searching in {community_info['name']}...")
30            
31            tweets = self.searcher.search_community_tweets(
32                community_id, query, target_count
33            )
34            
35            if tweets:
36                self.communities[community_id]["tweets"] = tweets
37                results[community_id] = tweets
38                print(f"   Found {len(tweets)} tweets")
39            else:
40                print(f"   No tweets found or error occurred")
41            
42            # Rate limiting
43            if delay > 0:
44                time.sleep(delay)
45        
46        return results
47    
48    def compare_communities(self, query_results):
49        """Compare engagement and activity across communities"""
50        comparison = {}
51        
52        for community_id, tweets in query_results.items():
53            if not tweets:
54                continue
55                
56            community_name = self.communities[community_id]["name"]
57            
58            total_engagement = sum(
59                tweet.get('favorite_count', 0) + 
60                tweet.get('retweet_count', 0) + 
61                tweet.get('reply_count', 0)
62                for tweet in tweets
63            )
64            
65            avg_engagement = total_engagement / len(tweets) if tweets else 0
66            
67            # Count verified users
68            verified_users = sum(1 for tweet in tweets 
69                               if tweet.get('user', {}).get('verified', False))
70            
71            comparison[community_id] = {
72                "name": community_name,
73                "tweet_count": len(tweets),
74                "total_engagement": total_engagement,
75                "avg_engagement": avg_engagement,
76                "verified_users": verified_users,
77                "verified_percentage": (verified_users / len(tweets) * 100) if tweets else 0
78            }
79        
80        return comparison
81    
82    def generate_report(self, query, comparison):
83        """Generate a comprehensive comparison report"""
84        print("\\n" + "="*70)
85        print(f"📊 MULTI-COMMUNITY ANALYSIS REPORT: '{query}'")
86        print("="*70)
87        
88        # Sort communities by total engagement
89        sorted_communities = sorted(
90            comparison.items(), 
91            key=lambda x: x[1]['total_engagement'], 
92            reverse=True
93        )
94        
95        print(f"{'Community':<25} {'Tweets':<8} {'Total Eng.':<12} {'Avg Eng.':<10} {'Verified %':<10}")
96        print("-" * 70)
97        
98        for community_id, data in sorted_communities:
99            print(f"{data['name']:<25} {data['tweet_count']:<8} "
100                  f"{data['total_engagement']:<12,} {data['avg_engagement']:<10.1f} "
101                  f"{data['verified_percentage']:<10.1f}%")
102    
103    def export_results(self, query, filename=None):
104        """Export all results to a JSON file"""
105        if not filename:
106            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
107            filename = f"community_search_{query}_{timestamp}.json"
108        
109        export_data = {
110            "query": query,
111            "timestamp": datetime.now().isoformat(),
112            "communities": self.communities
113        }
114        
115        with open(filename, 'w', encoding='utf-8') as f:
116            json.dump(export_data, f, indent=2, ensure_ascii=False)
117        
118        print(f"\\n💾 Results exported to {filename}")
119
120# Example usage: Compare AI discussions across tech communities
121analyzer = MultiCommunityAnalyzer("<your_bearer_token_here>")
122
123# Add communities to analyze (replace with real community IDs)
124tech_communities = {
125    "1234567890123456789": {"name": "AI & Machine Learning", "description": "ML researchers and practitioners"},
126    "2345678901234567890": {"name": "Python Developers", "description": "Python programming community"},
127    "3456789012345678901": {"name": "Data Science", "description": "Data scientists and analysts"},
128    "4567890123456789012": {"name": "Tech Startups", "description": "Startup founders and entrepreneurs"}
129}
130
131for community_id, info in tech_communities.items():
132    analyzer.add_community(community_id, info["name"], info["description"])
133
134# Search for AI-related discussions
135query = "artificial intelligence"
136results = analyzer.search_all_communities(query, target_count=75, delay=3)
137
138# Compare and generate report
139comparison = analyzer.compare_communities(results)
140analyzer.generate_report(query, comparison)
141
142# Export results
143analyzer.export_results(query)

1import schedule
2import time
3from datetime import datetime
4import json
5
6class CommunityMonitor:
7    def __init__(self, bearer_token):
8        self.searcher = CommunityTweetSearcher(bearer_token)
9        self.monitoring_configs = []
10        self.alerts = []
11        
12    def add_monitoring_config(self, community_id, queries, alert_threshold=10):
13        """Add a community and queries to monitor"""
14        config = {
15            "community_id": community_id,
16            "queries": queries if isinstance(queries, list) else [queries],
17            "alert_threshold": alert_threshold,
18            "last_check": None,
19            "historical_data": []
20        }
21        self.monitoring_configs.append(config)
22        
23    def check_community_activity(self, config):
24        """Check activity for a specific community configuration"""
25        community_id = config["community_id"]
26        results = {}
27        
28        for query in config["queries"]:
29            tweets = self.searcher.search_community_tweets(
30                community_id, query, target_count=20
31            )
32            
33            if tweets:
34                # Calculate engagement metrics
35                total_engagement = sum(
36                    tweet.get('favorite_count', 0) + 
37                    tweet.get('retweet_count', 0)
38                    for tweet in tweets
39                )
40                
41                results[query] = {
42                    "tweet_count": len(tweets),
43                    "total_engagement": total_engagement,
44                    "avg_engagement": total_engagement / len(tweets),
45                    "top_tweet": max(tweets, key=lambda x: x.get('favorite_count', 0)),
46                    "recent_tweets": tweets[:3]  # Store top 3 for alerts
47                }
48                
49                # Check for alerts
50                if len(tweets) >= config["alert_threshold"]:
51                    self.create_alert(community_id, query, results[query])
52        
53        # Store historical data
54        config["historical_data"].append({
55            "timestamp": datetime.now().isoformat(),
56            "results": results
57        })
58        config["last_check"] = datetime.now()
59        
60        return results
61    
62    def create_alert(self, community_id, query, data):
63        """Create an alert for high activity"""
64        alert = {
65            "timestamp": datetime.now().isoformat(),
66            "community_id": community_id,
67            "query": query,
68            "tweet_count": data["tweet_count"],
69            "total_engagement": data["total_engagement"],
70            "top_tweet": data["top_tweet"],
71            "alert_type": "high_activity"
72        }
73        
74        self.alerts.append(alert)
75        self.send_alert_notification(alert)
76    
77    def send_alert_notification(self, alert):
78        """Send alert notification (placeholder for actual notification system)"""
79        print(f"\\n🚨 ALERT: High activity detected!")
80        print(f"Community: {alert['community_id']}")
81        print(f"Query: '{alert['query']}'")
82        print(f"Tweets found: {alert['tweet_count']}")
83        print(f"Total engagement: {alert['total_engagement']:,}")
84        print(f"Top tweet: {alert['top_tweet']['text'][:100]}...")
85        print(f"Link: https://twitter.com/i/status/{alert['top_tweet']['tweet_id']}")
86    
87    def run_monitoring_cycle(self):
88        """Run one complete monitoring cycle"""
89        print(f"\\n⏰ Running monitoring cycle at {datetime.now()}")
90        
91        for i, config in enumerate(self.monitoring_configs):
92            print(f"\\n📍 Checking community {i+1}/{len(self.monitoring_configs)}")
93            
94            try:
95                results = self.check_community_activity(config)
96                print(f"   ✅ Monitoring completed")
97                
98                # Brief summary
99                for query, data in results.items():
100                    print(f"   '{query}': {data['tweet_count']} tweets, "
101                          f"{data['avg_engagement']:.1f} avg engagement")
102                          
103            except Exception as e:
104                print(f"   ❌ Error monitoring community: {e}")
105            
106            # Rate limiting between communities
107            time.sleep(2)
108    
109    def start_monitoring(self, interval_minutes=30):
110        """Start continuous monitoring"""
111        print(f"🚀 Starting community monitoring (every {interval_minutes} minutes)")
112        print(f"Monitoring {len(self.monitoring_configs)} community configurations")
113        
114        # Schedule monitoring
115        schedule.every(interval_minutes).minutes.do(self.run_monitoring_cycle)
116        
117        # Run initial check
118        self.run_monitoring_cycle()
119        
120        # Keep running
121        try:
122            while True:
123                schedule.run_pending()
124                time.sleep(60)  # Check every minute for scheduled tasks
125        except KeyboardInterrupt:
126            print("\\n🛑 Monitoring stopped by user")
127    
128    def get_monitoring_summary(self):
129        """Get a summary of monitoring activity"""
130        print("\\n" + "="*50)
131        print("📊 MONITORING SUMMARY")
132        print("="*50)
133        print(f"Active monitors: {len(self.monitoring_configs)}")
134        print(f"Total alerts: {len(self.alerts)}")
135        
136        if self.alerts:
137            recent_alerts = sorted(self.alerts, key=lambda x: x['timestamp'], reverse=True)[:5]
138            print("\\n🚨 Recent alerts:")
139            for alert in recent_alerts:
140                print(f"   {alert['timestamp']}: {alert['query']} ({alert['tweet_count']} tweets)")
141
142# Example usage: Monitor crypto communities for trending topics
143monitor = CommunityMonitor("<your_bearer_token_here>")
144
145# Add monitoring configurations
146crypto_community_id = "1234567890123456789"
147monitor.add_monitoring_config(
148    crypto_community_id,
149    ["bitcoin", "ethereum", "NFT", "DeFi"],
150    alert_threshold=15
151)
152
153ai_community_id = "2345678901234567890"
154monitor.add_monitoring_config(
155    ai_community_id,
156    ["ChatGPT", "machine learning", "deep learning"],
157    alert_threshold=10
158)
159
160# Get initial summary
161monitor.get_monitoring_summary()
162
163# Start monitoring (uncomment to run)
164# monitor.start_monitoring(interval_minutes=30)

1class CommunityTrendDetector:
2    def __init__(self, bearer_token):
3        self.searcher = CommunityTweetSearcher(bearer_token)
4        self.trend_keywords = [
5            "trending", "viral", "breaking", "announcement", 
6            "launch", "release", "update", "news"
7        ]
8    
9    def detect_emerging_trends(self, community_id, hours_back=24):
10        """Detect emerging trends in a community"""
11        trends = {}
12        
13        for keyword in self.trend_keywords:
14            tweets = self.searcher.search_community_tweets(
15                community_id, keyword, target_count=100
16            )
17            
18            if tweets:
19                # Filter recent tweets
20                recent_tweets = self.filter_recent_tweets(tweets, hours_back)
21                
22                if len(recent_tweets) >= 5:  # Minimum threshold for trend
23                    trend_score = self.calculate_trend_score(recent_tweets)
24                    trends[keyword] = {
25                        "tweet_count": len(recent_tweets),
26                        "trend_score": trend_score,
27                        "sample_tweets": recent_tweets[:3]
28                    }
29        
30        return sorted(trends.items(), key=lambda x: x[1]["trend_score"], reverse=True)
31    
32    def filter_recent_tweets(self, tweets, hours_back):
33        """Filter tweets from the last N hours"""
34        cutoff_time = datetime.now() - timedelta(hours=hours_back)
35        recent_tweets = []
36        
37        for tweet in tweets:
38            tweet_time = datetime.strptime(
39                tweet['created_at'], "%a %b %d %H:%M:%S +0000 %Y"
40            )
41            if tweet_time >= cutoff_time:
42                recent_tweets.append(tweet)
43        
44        return recent_tweets
45    
46    def calculate_trend_score(self, tweets):
47        """Calculate a trend score based on engagement and recency"""
48        if not tweets:
49            return 0
50        
51        total_engagement = sum(
52            tweet.get('favorite_count', 0) + 
53            tweet.get('retweet_count', 0) * 2 +  # Weight retweets higher
54            tweet.get('reply_count', 0)
55            for tweet in tweets
56        )
57        
58        # Boost score for verified users
59        verified_boost = sum(
60            50 for tweet in tweets 
61            if tweet.get('user', {}).get('verified', False)
62        )
63        
64        return total_engagement + verified_boost
65
66# Example usage
67trend_detector = CommunityTrendDetector("<your_bearer_token_here>")
68
69tech_community = "1234567890123456789"
70trends = trend_detector.detect_emerging_trends(tech_community, hours_back=12)
71
72print("🔥 Emerging trends in the last 12 hours:")
73for keyword, data in trends[:5]:
74    print(f"   {keyword}: {data['tweet_count']} tweets, score: {data['trend_score']}")
75    print(f"      Sample: {data['sample_tweets'][0]['text'][:80]}...")

1class CommunityResearchHelper:
2    def __init__(self):
3        self.research_templates = {
4            "brand_monitoring": [
5                "brand name", "product name", "company name",
6                "@brand_handle", "customer service", "review"
7            ],
8            "trend_analysis": [
9                "trending", "viral", "popular", "hot topic",
10                "everyone's talking about", "must see"
11            ],
12            "sentiment_analysis": [
13                "love", "hate", "amazing", "terrible", "excited",
14                "disappointed", "satisfied", "frustrated"
15            ],
16            "competitive_intelligence": [
17                "competitor name", "vs competitor", "alternative to",
18                "better than", "compared to", "switch from"
19            ]
20        }
21    
22    def suggest_queries(self, research_type, custom_terms=None):
23        """Suggest effective search queries for different research types"""
24        base_queries = self.research_templates.get(research_type, [])
25        
26        if custom_terms:
27            # Combine base queries with custom terms
28            combined_queries = []
29            for base in base_queries[:3]:  # Use top 3 base queries
30                for term in custom_terms:
31                    combined_queries.append(f"{term} {base}")
32                    combined_queries.append(f"{base} {term}")
33            
34            return combined_queries[:10]  # Return top 10 combinations
35        
36        return base_queries
37    
38    def optimize_search_parameters(self, community_size="medium"):
39        """Suggest optimal search parameters based on community size"""
40        configs = {
41            "small": {"target_count": 25, "delay": 1},
42            "medium": {"target_count": 50, "delay": 2},
43            "large": {"target_count": 100, "delay": 3},
44            "very_large": {"target_count": 200, "delay": 5}
45        }
46        
47        return configs.get(community_size, configs["medium"])
48    
49    def validate_community_id(self, community_id):
50        """Basic validation for community ID format"""
51        if not isinstance(community_id, str):
52            return False, "Community ID must be a string"
53        
54        if len(community_id) < 10:
55            return False, "Community ID appears too short"
56        
57        if not community_id.isdigit():
58            return False, "Community ID should contain only digits"
59        
60        return True, "Valid format"
61
62# Usage examples
63helper = CommunityResearchHelper()
64
65# Get suggested queries for brand monitoring
66brand_queries = helper.suggest_queries("brand_monitoring", ["OpenAI", "ChatGPT"])
67print("Brand monitoring queries:", brand_queries[:5])
68
69# Get optimal parameters for a large community
70params = helper.optimize_search_parameters("large")
71print("Recommended parameters:", params)
72
73# Validate community ID
74is_valid, message = helper.validate_community_id("1234567890123456789")
75print(f"Community ID validation: {message}")

1import time
2import random
3from functools import wraps
4
5def retry_with_backoff(max_retries=3, base_delay=1):
6    """Decorator for implementing retry logic with exponential backoff"""
7    def decorator(func):
8        @wraps(func)
9        def wrapper(*args, **kwargs):
10            for attempt in range(max_retries):
11                try:
12                    return func(*args, **kwargs)
13                except requests.exceptions.RequestException as e:
14                    if attempt == max_retries - 1:
15                        print(f"❌ Final attempt failed: {e}")
16                        raise
17                    
18                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
19                    print(f"⚠️ Attempt {attempt + 1} failed, retrying in {delay:.1f}s...")
20                    time.sleep(delay)
21            
22            return None
23        return wrapper
24    return decorator
25
26class RobustCommunitySearcher(CommunityTweetSearcher):
27    def __init__(self, bearer_token):
28        super().__init__(bearer_token)
29        self.request_count = 0
30        self.last_request_time = 0
31        self.min_request_interval = 1  # Minimum seconds between requests
32    
33    def rate_limit(self):
34        """Implement rate limiting"""
35        current_time = time.time()
36        time_since_last = current_time - self.last_request_time
37        
38        if time_since_last < self.min_request_interval:
39            sleep_time = self.min_request_interval - time_since_last
40            print(f"⏱️ Rate limiting: waiting {sleep_time:.1f}s")
41            time.sleep(sleep_time)
42        
43        self.last_request_time = time.time()
44        self.request_count += 1
45    
46    @retry_with_backoff(max_retries=3, base_delay=2)
47    def search_community_tweets_robust(self, community_id, query, target_count=50):
48        """Robust version with rate limiting and error handling"""
49        # Validate inputs
50        if not community_id or not query:
51            raise ValueError("Community ID and query are required")
52        
53        if target_count > 1000:
54            print("⚠️ Warning: Large target_count may hit rate limits")
55        
56        # Apply rate limiting
57        self.rate_limit()
58        
59        payload = {
60            "community_id": community_id,
61            "target_count": target_count,
62            "query": query
63        }
64        
65        print(f"🔍 Searching '{query}' in community {community_id[:10]}...")
66        
67        response = requests.post(
68            self.base_url,
69            headers=self.headers,
70            data=json.dumps(payload),
71            timeout=30
72        )
73        
74        if response.status_code == 200:
75            data = response.json()
76            tweets = data.get("data", [])
77            print(f"✅ Found {len(tweets)} tweets")
78            return tweets
79        elif response.status_code == 429:
80            print("🚫 Rate limit exceeded")
81            raise requests.exceptions.RequestException("Rate limit exceeded")
82        elif response.status_code == 422:
83            print(f"❌ Validation error: {response.text}")
84            return None
85        else:
86            print(f"❌ API error: {response.status_code}")
87            raise requests.exceptions.RequestException(f"API error: {response.status_code}")
88
89# Example usage with robust error handling
90robust_searcher = RobustCommunitySearcher("<your_bearer_token_here>")
91
92try:
93    tweets = robust_searcher.search_community_tweets_robust(
94        "1234567890123456789",
95        "machine learning",
96        target_count=75
97    )
98    
99    if tweets:
100        print(f"Successfully retrieved {len(tweets)} tweets")
101    else:
102        print("No tweets found or validation error occurred")
103        
104except Exception as e:
105    print(f"Search failed after all retries: {e}")