pip install tweepy

import tweepy
 
client = tweepy.Client(bearer_token="YOUR_BEARER_TOKEN")
 
query = "artificial intelligence -is:retweet lang:en"
 
response = client.search_recent_tweets(
    query=query,
    max_results=100,
    tweet_fields=["created_at", "public_metrics", "author_id", "lang"],
    expansions=["author_id"],
    user_fields=["username", "public_metrics"]
)
 
users = {u.id: u for u in response.includes.get("users", [])}
 
tweets = []
for tweet in response.data:
    author = users.get(tweet.author_id)
    tweets.append({
        "id": tweet.id,
        "text": tweet.text,
        "author": author.username if author else None,
        "created_at": tweet.created_at.isoformat(),
        "likes": tweet.public_metrics["like_count"],
        "retweets": tweet.public_metrics["retweet_count"],
        "replies": tweet.public_metrics["reply_count"],
        "impressions": tweet.public_metrics.get("impression_count", 0),
    })
 
for t in tweets[:5]:
    print(f"@{t['author']}: {t['text'][:80]}... | {t['likes']} likes")

pip install scrapegraph-py

from scrapegraph_py import Client
 
client = Client(api_key="your-sgai-key")
 
response = client.smartscraper(
    website_url="https://x.com/OpenAI",
    user_prompt="Extract the 10 most recent tweets with full text, timestamp, like count, retweet count, reply count, and view count"
)
 
tweets = response["result"]
for tweet in tweets:
    print(f"{tweet['timestamp']} | {tweet['text'][:80]}... | {tweet['likes']} likes")
 
client.close()

from scrapegraph_py import Client
 
client = Client(api_key="your-sgai-key")
 
response = client.smartscraper(
    website_url="https://x.com/search?q=web+scraping+python&src=typed_query&f=live",
    user_prompt="""Extract all visible tweets. For each tweet return:
    - author_handle
    - display_name
    - full_text
    - timestamp
    - likes
    - retweets
    - replies
    - views
    - has_media (true/false)
    - media_urls (list of image/video URLs if present)"""
)
 
tweets = response["result"]
print(f"Extracted {len(tweets)} tweets")
 
client.close()

from scrapegraph_py import Client
 
client = Client(api_key="your-sgai-key")
 
response = client.smartscraper(
    website_url="https://x.com/ylecun",
    user_prompt="""Extract:
    1. Profile info: display name, handle, bio, follower count, following count, verified status
    2. The 5 most recent tweets with text, timestamp, and all engagement metrics
    3. Any pinned tweet if present"""
)
 
data = response["result"]
print(f"@{data.get('handle')} | {data.get('follower_count')} followers")
print(f"Bio: {data.get('bio')}")
 
client.close()

from scrapegraph_py import Client
 
client = Client(api_key="your-sgai-key")
 
response = client.smartscraper(
    website_url="https://x.com/username/status/1234567890",
    user_prompt="""This is a tweet thread. Extract:
    - The original tweet text and metrics
    - All replies from the original author (thread continuation)
    - Any quoted tweet text and author
    - Total engagement across the thread (sum of likes, retweets, replies)"""
)
 
thread = response["result"]
print(f"Thread length: {len(thread.get('thread_tweets', []))} posts")
print(f"Total likes across thread: {thread.get('total_likes', 'N/A')}")
 
client.close()

{
  "profile": {
    "display_name": "Yann LeCun",
    "handle": "ylecun",
    "bio": "VP & Chief AI Scientist at Meta. Professor at NYU. ACM Turing Award Laureate.",
    "followers": "695.4K",
    "following": "2,891",
    "verified": true
  },
  "pinned_tweet": null,
  "recent_tweets": [
    {
      "text": "New paper on V-JEPA 2.0. We trained a video prediction model that learns rich visual representations without any labels.",
      "timestamp": "2026-03-23T16:45:00Z",
      "likes": 8420,
      "retweets": 2150,
      "replies": 342,
      "views": "1.8M",
      "has_media": true,
      "is_thread": false
    }
  ]
}

from scrapegraph_py import Client
import json
import time
from datetime import datetime
 
def collect_tweets(keywords, max_per_keyword=100):
    client = Client(api_key="your-sgai-key")
    all_tweets = []
 
    for keyword in keywords:
        search_url = f"https://x.com/search?q={keyword.replace(' ', '+')}&src=typed_query&f=live"
 
        response = client.smartscraper(
            website_url=search_url,
            user_prompt=f"Extract all visible tweets about '{keyword}'. Include author_handle, full_text, timestamp, likes, retweets, replies, views."
        )
 
        tweets = response.get("result", [])
        for tweet in tweets[:max_per_keyword]:
            tweet["keyword"] = keyword
            tweet["collected_at"] = datetime.utcnow().isoformat()
            all_tweets.append(tweet)
 
        time.sleep(2)
 
    client.close()
    return all_tweets
 
keywords = ["ScrapeGraphAI", "web scraping API", "AI data extraction"]
raw_tweets = collect_tweets(keywords)
 
with open("raw_tweets.json", "w") as f:
    json.dump(raw_tweets, f, indent=2)
 
print(f"Collected {len(raw_tweets)} tweets across {len(keywords)} keywords")

import json
import re
import sqlite3
from hashlib import sha256
 
def clean_tweet_text(text):
    text = re.sub(r"http\S+", "", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text
 
def deduplicate_tweets(tweets):
    seen = set()
    unique = []
    for tweet in tweets:
        content_hash = sha256(
            f"{tweet.get('author_handle', '')}:{tweet.get('full_text', '')[:100]}".encode()
        ).hexdigest()
        if content_hash not in seen:
            seen.add(content_hash)
            tweet["content_hash"] = content_hash
            tweet["clean_text"] = clean_tweet_text(tweet.get("full_text", ""))
            unique.append(tweet)
    return unique
 
def init_db(db_path="tweets.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS tweets (
            content_hash TEXT PRIMARY KEY,
            author_handle TEXT,
            full_text TEXT,
            clean_text TEXT,
            timestamp TEXT,
            likes INTEGER DEFAULT 0,
            retweets INTEGER DEFAULT 0,
            replies INTEGER DEFAULT 0,
            views TEXT,
            keyword TEXT,
            collected_at TEXT
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_keyword ON tweets(keyword)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON tweets(timestamp)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_author ON tweets(author_handle)")
    conn.commit()
    return conn
 
def insert_tweets(conn, tweets):
    inserted = 0
    for tweet in tweets:
        try:
            conn.execute("""
                INSERT OR IGNORE INTO tweets
                (content_hash, author_handle, full_text, clean_text, timestamp,
                 likes, retweets, replies, views, keyword, collected_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                tweet.get("content_hash"),
                tweet.get("author_handle"),
                tweet.get("full_text"),
                tweet.get("clean_text"),
                tweet.get("timestamp"),
                tweet.get("likes", 0),
                tweet.get("retweets", 0),
                tweet.get("replies", 0),
                tweet.get("views"),
                tweet.get("keyword"),
                tweet.get("collected_at"),
            ))
            inserted += 1
        except Exception:
            continue
    conn.commit()
    return inserted
 
with open("raw_tweets.json") as f:
    raw = json.load(f)
 
cleaned = deduplicate_tweets(raw)
conn = init_db()
count = insert_tweets(conn, cleaned)
print(f"Deduped: {len(raw)} → {len(cleaned)} | Inserted {count} into database")
conn.close()

#!/usr/bin/env python3
import sqlite3
import time
from datetime import datetime
from hashlib import sha256
from scrapegraph_py import Client
 
KEYWORDS = ["ScrapeGraphAI", "web scraping", "AI scraping"]
DB_PATH = "/var/data/tweets.db"
API_KEY = "your-sgai-key"
 
def run_collection():
    client = Client(api_key=API_KEY)
    conn = sqlite3.connect(DB_PATH)
 
    conn.execute("""
        CREATE TABLE IF NOT EXISTS tweets (
            content_hash TEXT PRIMARY KEY,
            author_handle TEXT,
            full_text TEXT,
            timestamp TEXT,
            likes INTEGER DEFAULT 0,
            retweets INTEGER DEFAULT 0,
            keyword TEXT,
            collected_at TEXT
        )
    """)
 
    total = 0
    for keyword in KEYWORDS:
        url = f"https://x.com/search?q={keyword.replace(' ', '+')}&f=live"
        try:
            response = client.smartscraper(
                website_url=url,
                user_prompt="Extract all visible tweets with author_handle, full_text, timestamp, likes, retweets"
            )
            for tweet in response.get("result", []):
                h = sha256(f"{tweet.get('author_handle')}:{tweet.get('full_text', '')[:100]}".encode()).hexdigest()
                conn.execute(
                    "INSERT OR IGNORE INTO tweets VALUES (?,?,?,?,?,?,?,?)",
                    (h, tweet.get("author_handle"), tweet.get("full_text"),
                     tweet.get("timestamp"), tweet.get("likes", 0),
                     tweet.get("retweets", 0), keyword, datetime.utcnow().isoformat())
                )
                total += 1
            time.sleep(3)
        except Exception as e:
            print(f"Error collecting '{keyword}': {e}")
 
    conn.commit()
    conn.close()
    client.close()
    print(f"[{datetime.utcnow().isoformat()}] Collected {total} tweets")
 
if __name__ == "__main__":
    run_collection()

# Run every 6 hours
0 */6 * * * /usr/bin/python3 /opt/scripts/collect_tweets.py >> /var/log/tweet_collector.log 2>&1

with open("tweets.json", "w", encoding="utf-8") as f:
    json.dump(tweets, f, ensure_ascii=False, indent=2)

from scrapegraph_py import Client
from textblob import TextBlob
import sqlite3
from datetime import datetime
from hashlib import sha256
 
TARGETS = {
    "your_product": ["YourProduct", "@yourhandle"],
    "competitor_a": ["CompetitorA", "@compa"],
    "competitor_b": ["CompetitorB", "@compb"],
}
 
def collect_and_analyze():
    client = Client(api_key="your-sgai-key")
    conn = sqlite3.connect("competitive_intel.db")
 
    conn.execute("""
        CREATE TABLE IF NOT EXISTS mentions (
            hash TEXT PRIMARY KEY,
            target TEXT,
            author TEXT,
            text TEXT,
            timestamp TEXT,
            likes INTEGER,
            sentiment REAL,
            sentiment_label TEXT,
            collected_at TEXT
        )
    """)
 
    for target_name, keywords in TARGETS.items():
        query = " OR ".join(keywords)
        url = f"https://x.com/search?q={query}&f=live"
 
        response = client.smartscraper(
            website_url=url,
            user_prompt="Extract all tweets with author_handle, full_text, timestamp, likes"
        )
 
        for tweet in response.get("result", []):
            text = tweet.get("full_text", "")
            score = TextBlob(text).sentiment.polarity
            label = "positive" if score > 0.1 else ("negative" if score < -0.1 else "neutral")
            h = sha256(f"{tweet.get('author_handle')}:{text[:100]}".encode()).hexdigest()
 
            conn.execute(
                "INSERT OR IGNORE INTO mentions VALUES (?,?,?,?,?,?,?,?,?)",
                (h, target_name, tweet.get("author_handle"), text,
                 tweet.get("timestamp"), tweet.get("likes", 0),
                 score, label, datetime.utcnow().isoformat())
            )
 
    conn.commit()
 
    for target in TARGETS:
        row = conn.execute("""
            SELECT COUNT(*), ROUND(AVG(sentiment), 3),
                   SUM(CASE WHEN sentiment_label='negative' THEN 1 ELSE 0 END)
            FROM mentions WHERE target = ?
        """, (target,)).fetchone()
        neg_pct = (row[2] / row[0] * 100) if row[0] > 0 else 0
        alert = " !! SPIKE" if neg_pct > 30 else ""
        print(f"  {target}: {row[0]} mentions, sentiment={row[1]}, {neg_pct:.0f}% negative{alert}")
 
    conn.close()
    client.close()
 
collect_and_analyze()