Threads APIthreads.net ↗

"""
Threads Search API Client
Practical example for searching posts and users on Threads (threads.net)

Get your API key from: https://parse.bot/settings
"""

import os
import requests
from typing import Optional
from datetime import datetime


class ParseClient:
    """Client for interacting with the Threads Search API via Parse."""

    def __init__(self, api_key: Optional[str] = None):
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "48546944-bfeb-482c-9503-061e6b392624"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")

        if not self.api_key:
            raise ValueError(
                "API key not provided. Set PARSE_API_KEY environment variable or pass it to the constructor."
            )

    def _call(
        self, endpoint: str, method: str = "POST", **params
    ) -> dict:
        """
        Make a request to the Parse API.

        Args:
            endpoint: The endpoint name (e.g., 'search_posts')
            method: HTTP method - 'GET' or 'POST'
            **params: Query/payload parameters

        Returns:
            Response JSON as dictionary
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }

        if method == "GET":
            response = requests.get(url, headers=headers, params=params)
        elif method == "POST":
            response = requests.post(url, headers=headers, json=params)
        else:
            raise ValueError(f"Unsupported HTTP method: {method}")

        response.raise_for_status()
        return response.json()

    def search_posts(
        self,
        query: str,
        search_surface: str = "default"
    ) -> dict:
        """
        Search for posts/threads on Threads by keyword.

        Args:
            query: Search keyword or phrase to find posts about
            search_surface: Search result ordering - 'default', 'top', or 'recent'

        Returns:
            Dictionary containing posts array and pagination info
        """
        return self._call(
            "search_posts",
            method="GET",
            query=query,
            search_surface=search_surface
        )

    def search_users(
        self,
        query: str,
        limit: int = 10
    ) -> dict:
        """
        Search for user accounts on Threads by keyword.

        Args:
            query: Search keyword or username to find user accounts
            limit: Maximum number of user results to return (1-50)

        Returns:
            Dictionary containing users array
        """
        return self._call(
            "search_users",
            method="GET",
            query=query,
            limit=limit
        )


def main():
    """Practical workflow example: Find and analyze posts from popular tech accounts."""

    client = ParseClient()

    print("=" * 60)
    print("THREADS SEARCH API - PRACTICAL WORKFLOW EXAMPLE")
    print("=" * 60)

    search_topic = "artificial intelligence"
    print(f"\n📱 Step 1: Searching for users interested in '{search_topic}'...")

    users_response = client.search_users(query=search_topic, limit=5)
    tech_users = users_response.get("users", [])

    if not tech_users:
        print("No users found for this search.")
        return

    print(f"✅ Found {len(tech_users)} tech users:")
    for i, user in enumerate(tech_users, 1):
        status = "✓" if user.get("is_active_on_threads") else "✗"
        verified = "✓" if user.get("is_verified") else " "
        print(
            f"  {i}. @{user['username']} ({user['full_name']}) "
            f"[Active: {status}] [Verified: {verified}]"
        )

    print(f"\n🔍 Step 2: Searching for recent posts about '{search_topic}'...")
    posts_response = client.search_posts(query=search_topic, search_surface="recent")
    posts = posts_response.get("posts", [])

    if not posts:
        print("No posts found for this search.")
        return

    print(f"✅ Found {len(posts)} posts about '{search_topic}':")
    print()

    total_engagement = 0
    engagement_by_user = {}

    for i, post in enumerate(posts[:5], 1):
        user = post.get("user", {})
        username = user.get("username", "unknown")
        engagement = (
            post.get("like_count", 0)
            + post.get("reply_count", 0)
            + post.get("repost_count", 0)
            + post.get("quote_count", 0)
        )
        total_engagement += engagement

        if username not in engagement_by_user:
            engagement_by_user[username] = {
                "posts": 0,
                "total_engagement": 0,
                "user_info": user
            }

        engagement_by_user[username]["posts"] += 1
        engagement_by_user[username]["total_engagement"] += engagement

        timestamp = datetime.fromtimestamp(post.get("taken_at", 0)).strftime(
            "%Y-%m-%d %H:%M:%S"
        )

        text_preview = post.get("text", "")[:100]
        if len(post.get("text", "")) > 100:
            text_preview += "..."

        print(f"Post {i} by @{username}")
        print(f"  📝 {text_preview}")
        print(f"  📊 Engagement: ❤️ {post.get('like_count', 0)} | "
              f"💬 {post.get('reply_count', 0)} | "
              f"🔄 {post.get('repost_count', 0)} | "
              f"💭 {post.get('quote_count', 0)}")
        print(f"  ⏱️ Posted: {timestamp}")
        print()

    print("=" * 60)
    print("📊 ENGAGEMENT ANALYSIS")
    print("=" * 60)
    print(f"Total posts analyzed: {len(posts[:5])}")
    print(f"Total engagement: {total_engagement}")
    print(f"Average engagement per post: {total_engagement / len(posts[:5]):.1f}")
    print()

    print("Top engaging users:")
    sorted_users = sorted(
        engagement_by_user.items(),
        key=lambda x: x[1]["total_engagement"],
        reverse=True
    )

    for rank, (username, stats) in enumerate(sorted_users[:3], 1):
        user_info = stats["user_info"]
        verified = "✓" if user_info.get("is_verified") else " "
        print(
            f"  {rank}. @{username} - "
            f"Posts: {stats['posts']}, "
            f"Total Engagement: {stats['total_engagement']} "
            f"[Verified: {verified}]"
        )

    print("\n✨ Workflow complete!")


if __name__ == "__main__":
    main()