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}

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']}")

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")

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)

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)