Tweet¶

from xeepy.models import Tweet

@dataclass
class Tweet:
    id: str
    text: str
    author: User
    created_at: datetime
    like_count: int = 0
    retweet_count: int = 0
    reply_count: int = 0
    quote_count: int = 0
    view_count: int = 0
    bookmark_count: int = 0
    url: str = ""
    conversation_id: Optional[str] = None
    in_reply_to_id: Optional[str] = None
    in_reply_to_user: Optional[str] = None
    is_retweet: bool = False
    is_quote: bool = False
    is_reply: bool = False
    media: List[Media] = field(default_factory=list)
    hashtags: List[str] = field(default_factory=list)
    mentions: List[str] = field(default_factory=list)
    urls: List[str] = field(default_factory=list)
    language: Optional[str] = None
    source: Optional[str] = None
    is_sensitive: bool = False
    poll: Optional[Poll] = None

@property
def engagement_rate(self) -> float:
    """Calculate engagement rate."""

@property
def total_engagement(self) -> int:
    """Sum of all engagement metrics."""

@property
def has_media(self) -> bool:
    """Check if tweet has media."""

@property
def is_thread(self) -> bool:
    """Check if part of a thread."""

from xeepy import Xeepy

async def main():
    async with Xeepy() as x:
        tweets = await x.scrape.tweets("username", limit=10)

        for tweet in tweets.items:
            print(f"ID: {tweet.id}")
            print(f"Text: {tweet.text}")
            print(f"Author: @{tweet.author.username}")
            print(f"Posted: {tweet.created_at}")
            print(f"Likes: {tweet.like_count:,}")
            print(f"Retweets: {tweet.retweet_count:,}")
            print(f"Views: {tweet.view_count:,}")
            print(f"URL: {tweet.url}")
            print("---")

asyncio.run(main())

from xeepy import Xeepy

async def categorize_tweets(username: str):
    async with Xeepy() as x:
        tweets = await x.scrape.tweets(username, limit=100)

        original = [t for t in tweets.items if not t.is_retweet and not t.is_reply]
        replies = [t for t in tweets.items if t.is_reply]
        retweets = [t for t in tweets.items if t.is_retweet]
        quotes = [t for t in tweets.items if t.is_quote]

        print(f"Original: {len(original)}")
        print(f"Replies: {len(replies)}")
        print(f"Retweets: {len(retweets)}")
        print(f"Quotes: {len(quotes)}")

asyncio.run(categorize_tweets("username"))

from xeepy import Xeepy

async def get_media_tweets(username: str):
    async with Xeepy() as x:
        tweets = await x.scrape.tweets(username, limit=50)

        for tweet in tweets.items:
            if tweet.has_media:
                print(f"Tweet: {tweet.text[:50]}...")
                for media in tweet.media:
                    print(f"  - {media.type}: {media.url}")

asyncio.run(get_media_tweets("username"))

from xeepy import Xeepy

async def top_engagement(username: str):
    async with Xeepy() as x:
        tweets = await x.scrape.tweets(username, limit=100)

        # Sort by engagement
        sorted_tweets = sorted(
            tweets.items,
            key=lambda t: t.total_engagement,
            reverse=True
        )

        print("Top 10 by engagement:")
        for tweet in sorted_tweets[:10]:
            print(f"{tweet.total_engagement:,} - {tweet.text[:40]}...")
            print(f"  Rate: {tweet.engagement_rate:.2f}%")

asyncio.run(top_engagement("username"))

from xeepy import Xeepy
from collections import Counter

async def popular_hashtags(username: str):
    async with Xeepy() as x:
        tweets = await x.scrape.tweets(username, limit=200)

        all_hashtags = []
        for tweet in tweets.items:
            all_hashtags.extend(tweet.hashtags)

        counter = Counter(all_hashtags)

        print("Most used hashtags:")
        for tag, count in counter.most_common(10):
            print(f"  #{tag}: {count}")

asyncio.run(popular_hashtags("username"))

from xeepy import Xeepy

async def filter_tweets(username: str):
    async with Xeepy() as x:
        tweets = await x.scrape.tweets(username, limit=100)

        # Tweets with links
        with_links = [t for t in tweets.items if t.urls]
        print(f"With links: {len(with_links)}")

        # Tweets with mentions
        with_mentions = [t for t in tweets.items if t.mentions]
        print(f"With mentions: {len(with_mentions)}")

        # Long tweets
        long_tweets = [t for t in tweets.items if len(t.text) > 200]
        print(f"Long tweets: {len(long_tweets)}")

asyncio.run(filter_tweets("username"))

from xeepy import Xeepy
import json

async def export_tweet(tweet_url: str):
    async with Xeepy() as x:
        # Get single tweet
        tweet = await x.scrape.tweet(tweet_url)

        # Convert to dict
        data = tweet.to_dict()

        # Save as JSON
        with open("tweet.json", "w") as f:
            json.dump(data, f, indent=2, default=str)

asyncio.run(export_tweet("https://x.com/user/status/123"))

from xeepy import Xeepy

async def get_poll_tweet(tweet_url: str):
    async with Xeepy() as x:
        tweet = await x.scrape.tweet(tweet_url)

        if tweet.poll:
            print(f"Poll question: {tweet.text}")
            print(f"Total votes: {tweet.poll.total_votes:,}")
            print(f"Status: {tweet.poll.status}")

            for option in tweet.poll.options:
                bar = "█" * int(option.percentage / 5)
                print(f"  {option.label}: {bar} {option.percentage:.1f}%")

asyncio.run(get_poll_tweet("https://x.com/user/status/123"))

from xeepy import Xeepy

async def get_threads(username: str):
    async with Xeepy() as x:
        tweets = await x.scrape.tweets(username, limit=100)

        # Group by conversation
        conversations = {}
        for tweet in tweets.items:
            if tweet.conversation_id:
                if tweet.conversation_id not in conversations:
                    conversations[tweet.conversation_id] = []
                conversations[tweet.conversation_id].append(tweet)

        # Find threads (2+ tweets in conversation by same author)
        threads = [
            tweets for tweets in conversations.values()
            if len(tweets) >= 2
        ]

        print(f"Found {len(threads)} threads")

asyncio.run(get_threads("username"))