How to Automate X (Twitter) Analytics Reports using the API

1#!/usr/bin/env python3
2import os
3import logging
4import requests
5from datetime import datetime, timedelta, timezone
6
7logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
8
9ACCESS_TOKEN = os.environ['TWITTER_ACCESS_TOKEN']
10REFRESH_TOKEN = os.environ['TWITTER_REFRESH_TOKEN']
11CLIENT_ID = os.environ['TWITTER_CLIENT_ID']
12CLIENT_SECRET = os.environ['TWITTER_CLIENT_SECRET']
13SLACK_WEBHOOK = os.environ['SLACK_WEBHOOK_URL']
14
15def refresh_token():
16    global ACCESS_TOKEN, REFRESH_TOKEN
17    resp = requests.post(
18        'https://api.twitter.com/2/oauth2/token',
19        auth=(CLIENT_ID, CLIENT_SECRET),
20        data={'grant_type': 'refresh_token', 'refresh_token': REFRESH_TOKEN, 'client_id': CLIENT_ID}
21    )
22    resp.raise_for_status()
23    data = resp.json()
24    ACCESS_TOKEN = data['access_token']
25    if 'refresh_token' in data:
26        REFRESH_TOKEN = data['refresh_token']  # Store new refresh token!
27    logging.info('Token refreshed successfully')
28
29def twitter_get(path, params=None):
30    resp = requests.get(
31        f'https://api.twitter.com/2{path}',
32        params=params,
33        headers={'Authorization': f'Bearer {ACCESS_TOKEN}'}
34    )
35    if resp.status_code == 401:
36        refresh_token()
37        resp = requests.get(
38            f'https://api.twitter.com/2{path}',
39            params=params,
40            headers={'Authorization': f'Bearer {ACCESS_TOKEN}'}
41        )
42    resp.raise_for_status()
43    return resp.json()
44
45def get_user_id():
46    data = twitter_get('/users/me', {'user.fields': 'public_metrics'})
47    return data['data']['id']
48
49def get_recent_tweets(user_id):
50    start_time = (datetime.now(timezone.utc) - timedelta(days=7)).strftime('%Y-%m-%dT%H:%M:%SZ')
51    tweets = []
52    pagination_token = None
53    while True:
54        params = {
55            'max_results': 100,
56            'tweet.fields': 'public_metrics,organic_metrics,created_at',
57            'start_time': start_time
58        }
59        if pagination_token:
60            params['pagination_token'] = pagination_token
61        data = twitter_get(f'/users/{user_id}/tweets', params)
62        tweets.extend(data.get('data', []))
63        pagination_token = data.get('meta', {}).get('next_token')
64        if not pagination_token:
65            break
66    logging.info(f'Fetched {len(tweets)} tweets (cost: ~${len(tweets) * 0.001:.4f} on pay-per-use)')
67    return tweets
68
69def analyze_tweets(tweets):
70    results = []
71    for t in tweets:
72        m = t.get('public_metrics', {})
73        impressions = m.get('impression_count', 0)
74        total_eng = sum(m.get(k, 0) for k in ['like_count', 'retweet_count', 'reply_count', 'quote_count'])
75        results.append({
76            'id': t['id'], 'text': t['text'][:100],
77            'impressions': impressions, 'total_engagement': total_eng,
78            'engagement_rate': round(total_eng / impressions * 100, 2) if impressions > 0 else 0,
79            'created_at': t.get('created_at', '')
80        })
81    return sorted(results, key=lambda x: x['total_engagement'], reverse=True)
82
83def send_slack_report(analyzed):
84    total_imp = sum(t['impressions'] for t in analyzed)
85    total_eng = sum(t['total_engagement'] for t in analyzed)
86    avg_rate = total_eng / total_imp * 100 if total_imp > 0 else 0
87    top = analyzed[0] if analyzed else {}
88    payload = {
89        'text': '*X (Twitter) Weekly Analytics Report*',
90        'attachments': [{
91            'color': '#1DA1F2',
92            'fields': [
93                {'title': 'Impressions (7d)', 'value': f"{total_imp:,}", 'short': True},
94                {'title': 'Engagements (7d)', 'value': f"{total_eng:,}", 'short': True},
95                {'title': 'Avg Eng Rate', 'value': f"{avg_rate:.2f}%", 'short': True},
96                {'title': 'Tweets Published', 'value': str(len(analyzed)), 'short': True},
97                {'title': 'Top Tweet', 'value': top.get('text', 'N/A'), 'short': False}
98            ]
99        }]
100    }
101    requests.post(SLACK_WEBHOOK, json=payload)
102    logging.info('Slack report sent')
103
104if __name__ == '__main__':
105    user_id = get_user_id()
106    tweets = get_recent_tweets(user_id)
107    if tweets:
108        analyzed = analyze_tweets(tweets)
109        send_slack_report(analyzed)
110    else:
111        logging.info('No tweets in the last 7 days')