Modash APImodash.io ↗

"""
Modash Influencer Discovery API Client

This module provides a client for searching and analyzing influencer data
across Instagram, TikTok, and YouTube using the Parse API.

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

import os
import requests
from typing import Optional, Dict, Any


class ParseClient:
    """Client for interacting with the Modash Influencer Discovery API."""

    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the Parse API client.

        Args:
            api_key: Modash API key (defaults to PARSE_API_KEY env var)
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "3ef2b839-94b0-460c-aebe-37e4f435e153"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")

        if not self.api_key:
            raise ValueError("API key not provided and PARSE_API_KEY env var not set")

    def _call(
        self, endpoint: str, method: str = "POST", **params
    ) -> Dict[str, Any]:
        """
        Make an API call to the Parse bot.

        Args:
            endpoint: The endpoint name
            method: HTTP method (GET or POST)
            **params: Query/body 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_instagram_influencers(
        self,
        followers_min: Optional[int] = None,
        followers_max: Optional[int] = None,
        engagement_min: Optional[float] = None,
        location: Optional[str] = None,
        audience_location: Optional[str] = None,
        audience_location_weight: Optional[float] = None,
        is_verified: Optional[bool] = None,
        has_email: Optional[bool] = None,
        limit: int = 15,
        cursor: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Search for Instagram influencers with various filters.

        Args:
            followers_min: Minimum follower count
            followers_max: Maximum follower count
            engagement_min: Minimum engagement rate (e.g., 0.02 for 2%)
            location: Comma-separated ISO country codes
            audience_location: Comma-separated ISO country codes for audience location
            audience_location_weight: Minimum percentage of audience from location (0-1)
            is_verified: Filter for verified accounts
            has_email: Filter for influencers with public email
            limit: Number of results per page (max 15)
            cursor: Pagination cursor for next page

        Returns:
            Dictionary with lookalikes, total count, and cursor
        """
        params = {
            "api_key": self.api_key,
            "limit": limit,
        }

        if followers_min is not None:
            params["followers_min"] = followers_min
        if followers_max is not None:
            params["followers_max"] = followers_max
        if engagement_min is not None:
            params["engagement_min"] = engagement_min
        if location is not None:
            params["location"] = location
        if audience_location is not None:
            params["audience_location"] = audience_location
        if audience_location_weight is not None:
            params["audience_location_weight"] = audience_location_weight
        if is_verified is not None:
            params["is_verified"] = is_verified
        if has_email is not None:
            params["has_email"] = has_email
        if cursor is not None:
            params["cursor"] = cursor

        return self._call("search_instagram_influencers", method="POST", **params)

    def get_instagram_influencer_report(self, user_id: str) -> Dict[str, Any]:
        """
        Retrieve a detailed report for an Instagram influencer.

        Args:
            user_id: The Instagram user ID (numeric string)

        Returns:
            Dictionary with influencer details including contact info and demographics
        """
        return self._call(
            "get_instagram_influencer_report",
            method="GET",
            user_id=user_id,
            api_key=self.api_key,
        )

    def search_tiktok_influencers(
        self,
        followers_min: Optional[int] = None,
        location: Optional[str] = None,
        limit: int = 15,
        cursor: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Search for TikTok influencers with filters.

        Args:
            followers_min: Minimum follower count
            location: Comma-separated ISO country codes
            limit: Number of results per page
            cursor: Pagination cursor for next page

        Returns:
            Dictionary with TikTok influencer results
        """
        params = {"api_key": self.api_key, "limit": limit}

        if followers_min is not None:
            params["followers_min"] = followers_min
        if location is not None:
            params["location"] = location
        if cursor is not None:
            params["cursor"] = cursor

        return self._call("search_tiktok_influencers", method="POST", **params)

    def search_youtube_influencers(
        self,
        location: Optional[str] = None,
        limit: int = 15,
        cursor: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Search for YouTube influencers with filters.

        Args:
            location: Comma-separated ISO country codes
            limit: Number of results per page
            cursor: Pagination cursor for next page

        Returns:
            Dictionary with YouTube influencer results
        """
        params = {"api_key": self.api_key, "limit": limit}

        if location is not None:
            params["location"] = location
        if cursor is not None:
            params["cursor"] = cursor

        return self._call("search_youtube_influencers", method="POST", **params)

    def search_metadata(
        self,
        metadata_type: str = "locations",
        platform: str = "instagram",
        query: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Search for metadata like locations, topics, languages, brands, etc.

        Args:
            metadata_type: Type of metadata (locations, topics, languages, brands, interests, users, hashtags)
            platform: Social platform (instagram, tiktok, youtube)
            query: Search keyword (e.g., 'New York' for locations)

        Returns:
            Dictionary with metadata results
        """
        params = {
            "api_key": self.api_key,
            "metadata_type": metadata_type,
            "platform": platform,
        }

        if query is not None:
            params["query"] = query

        return self._call("search_metadata", method="GET", **params)

    def search_influencers_by_email(self, email: str) -> Dict[str, Any]:
        """
        Lookup influencer accounts associated with an email address.

        Args:
            email: Email address to look up

        Returns:
            Dictionary with influencer results across platforms
        """
        return self._call(
            "search_influencers_by_email",
            method="POST",
            api_key=self.api_key,
            email=email,
        )

    def ai_search_influencers(
        self, query: str, platform: str = "instagram"
    ) -> Dict[str, Any]:
        """
        AI-powered natural language search for influencers.

        Args:
            query: Natural language query (e.g., 'fitness influencers interested in yoga')
            platform: Social platform (instagram, tiktok, youtube)

        Returns:
            Dictionary with AI-matched influencer results
        """
        return self._call(
            "ai_search_influencers",
            method="POST",
            api_key=self.api_key,
            query=query,
            platform=platform,
        )


def main():
    """
    Practical workflow: Search for high-engagement Instagram influencers,
    retrieve detailed reports with contact info, and compile a list of top prospects.
    """
    # Initialize client
    client = ParseClient()

    print("=" * 80)
    print("MODASH INFLUENCER DISCOVERY - CAMPAIGN PROSPECT RESEARCH")
    print("=" * 80)

    # Step 1: Use AI search for natural language discovery
    print("\n[Step 1] Using AI to search for relevant influencers...")
    print("Query: 'sustainable fashion influencers with engaged audience'\n")

    ai_results = client.ai_search_influencers(
        query="sustainable fashion influencers with engaged audience",
        platform="instagram",
    )

    total_ai_found = ai_results.get("total", 0)
    ai_influencers = ai_results.get("lookalikes", [])[:3]  # Get first 3

    print(f"AI Search found {total_ai_found} potential matches")
    print(f"Showing top {len(ai_influencers)} results from AI search:")
    print("-" * 80)

    ai_user_ids = []
    for idx, influencer in enumerate(ai_influencers, 1):
        print(
            f"\n{idx}. @{influencer.get('username')} - {influencer.get('followers', 0):,} followers"
        )
        print(f"   User ID: {influencer.get('userId')}")
        ai_user_ids.append(influencer.get("userId"))

    # Step 2: Search with specific filters for additional prospects
    print("\n" + "=" * 80)
    print("[Step 2] Searching Instagram with specific filters...")
    print("Criteria: 10k-100k followers, 3%+ engagement, Verified accounts\n")

    search_results = client.search_instagram_influencers(
        followers_min=10000,
        followers_max=100000,
        engagement_min=0.03,
        is_verified=True,
        has_email=True,
        limit=5,
    )

    total_found = search_results.get("total", 0)
    influencers = search_results.get("lookalikes", [])

    print(f"Found {total_found} total influencers matching criteria")
    print(f"Retrieved {len(influencers)} influencers in this batch:")
    print("-" * 80)

    prospects = []

    # Step 3: Get detailed reports for all discovered influencers
    all_user_ids = ai_user_ids + [inf.get("userId") for inf in influencers]

    for user_id in all_user_ids:
        if not user_id:
            continue

        try:
            print(f"\nFetching detailed report for user {user_id}...")
            report = client.get_instagram_influencer_report(user_id)

            prospect = {
                "userId": user_id,
                "username": report.get("username", "N/A"),
                "fullname": report.get("profile", {}).get("fullname", "N/A"),
                "followers": report.get("profile", {}).get("followers", 0),
                "engagement": report.get("stats", {}).get("engagementRate", 0),
                "avg_likes": report.get("stats", {}).get("avgLikes", 0),
                "audience_countries": report.get("audience", {}).get("countries", [])[:3],
                "contact_email": report.get("contact", {}).get("email", "Not available"),
                "biography": report.get("profile", {}).get("biography", ""),
            }

            prospects.append(prospect)

            print(f"  ✓ Username: @{prospect['username']}")
            print(f"  ✓ Followers: {prospect['followers']:,}")
            print(f"  ✓ Avg Likes per Post: {prospect['avg_likes']}")
            print(f"  ✓ Contact: {prospect['contact_email']}")

        except requests.exceptions.RequestException as e:
            print(f"  ✗ Error fetching report: {str(e)}")
        except Exception as e:
            print(f"  ✗ Unexpected error: {str(e)}")

    # Step 4: Display final summary and recommendations
    print("\n" + "=" * 80)
    print("PROSPECT SUMMARY & RECOMMENDATIONS")
    print("=" * 80)

    if prospects:
        print(f"\nTotal prospects identified: {len(prospects)}")
        print("\nTop prospects by engagement (sorted by avg likes):")
        print("-" * 80)

        sorted_prospects = sorted(prospects, key=lambda x: x["avg_likes"], reverse=True)

        for idx, prospect in enumerate(sorted_prospects[:5], 1):
            print(
                f"\n{idx}. {prospect['fullname']} (@{prospect['username']})"
            )
            print(f"   Followers: {prospect['followers']:,}")
            print(f"   Avg Likes: {prospect['avg_likes']}")
            print(f"   Engagement Rate: {prospect['engagement']:.2%}")
            print(f"   Email: {prospect['contact_email']}")

            if prospect["audience_countries"]:
                countries = ", ".join(prospect["audience_countries"])
                print(f"   Top Audience Countries: {countries}")

            if prospect["biography"]:
                bio_preview = prospect["biography"][:60] + (
                    "..." if len(prospect["biography"]) > 60 else ""
                )
                print(f"   Bio: {bio_preview}")

        print("\n" + "-" * 80)
        print("NEXT STEPS:")
        print("1. Review top prospects above")
        print("2. Check their recent posts for content alignment")
        print("3. Prepare outreach emails to prospects with contact info")
        print("4. Consider collaboration opportunities with highest engagement performers")

    else:
        print("\nNo detailed prospects retrieved. Please check your API key and try again.")

    print("\n" + "=" * 80)


if __name__ == "__main__":
    main()