How to Monitor Twitter Accounts for New Tweets in Real-Time

pip install requests

import requests
import time
from datetime import datetime, timedelta, timezone

# Configuration
API_KEY = "Your API Key"  # TODO Replace with your TwitterAPI.io API key. You can find it in https://twitterapi.io/dashboard.
TARGET_ACCOUNT = "elonmusk"  # The account you want to monitor
CHECK_INTERVAL = 300  # Check every 5 minutes (300 seconds)
LAST_CHECKED_TIME = datetime.now(timezone.utc) - timedelta(hours=1)  # Start by checking the last hour

def check_for_new_tweets():
    global LAST_CHECKED_TIME

    # Compute the time window. The backend expects since_time / until_time
    # as Unix second timestamps in the query params (since:/until: in the
    # query string is no longer supported).
    until_time = datetime.now(timezone.utc)
    since_time = LAST_CHECKED_TIME

    # Construct the query
    query = f"from:{TARGET_ACCOUNT} include:nativeretweets"
    # Please refer to this document for detailed Twitter advanced search syntax.
    # https://github.com/igorbrigadir/twitter-advanced-search

    # API endpoint
    url = "https://api.twitterapi.io/twitter/tweet/advanced_search"

    # Request parameters
    params = {
        "query": query,
        "queryType": "Latest",
        "since_time": int(since_time.timestamp()),   # Unix seconds
        "until_time": int(until_time.timestamp()),
    }

    # Headers with API key
    headers = {
        "X-API-Key": API_KEY
    }

    # Make the request and handle pagination
    all_tweets = []
    next_cursor = None

    while True:
        # Add cursor to params if we have one
        if next_cursor:
            params["cursor"] = next_cursor

        response = requests.get(url, headers=headers, params=params)

        # Parse the response
        if response.status_code == 200:
            data = response.json()
            tweets = data.get("tweets", [])

            if tweets:
                all_tweets.extend(tweets)

            # Check if there are more pages
            if data.get("has_next_page", False) and data.get("next_cursor", "") != "":
                next_cursor = data.get("next_cursor")
                continue
            else:
                break
        else:
            print(f"Error: {response.status_code} - {response.text}")
            break

    # Process all collected tweets
    if all_tweets:
        print(f"Found {len(all_tweets)} total tweets from {TARGET_ACCOUNT}!")
        for tweet in all_tweets:
            print(f"[{tweet['createdAt']}] {tweet['text']}")
            # Here you could send notifications, save to database, etc.
    else:
        print(f"No new tweets from {TARGET_ACCOUNT} since last check.")

    # Update the last checked time
    LAST_CHECKED_TIME = until_time

# Main monitoring loop
def main():
    print(f"Starting to monitor tweets from @{TARGET_ACCOUNT}")
    print(f"Checking every {CHECK_INTERVAL} seconds")

    try:
        while True:
            check_for_new_tweets()
            time.sleep(CHECK_INTERVAL)
    except KeyboardInterrupt:
        print("Monitoring stopped.")

if __name__ == "__main__":
    main()