cloudseeding / tools / bluesky / search_bluesky.py

import os
from typing import Optional, List, Dict, Any
from atproto import Client, models


def _normalize_datetime(date_str: Optional[str]) -> Optional[str]:
    """
    Convert date string to ISO 8601 datetime format.
    Accepts: "2024-01-15" or "2024-01-15T10:30:00Z"
    Returns: "2024-01-15T00:00:00Z" or the original if already full datetime
    """
    if not date_str:
        return None

    # If it's already a full datetime (contains 'T'), return as-is
    if 'T' in date_str:
        return date_str

    # If it's just a date (YYYY-MM-DD), add time component
    if len(date_str) == 10 and date_str.count('-') == 2:
        return f"{date_str}T00:00:00Z"

    # Return as-is if we can't parse it (let the API handle the error)
    return date_str


def search_bluesky(
    query: str,
    search_type: str = "posts",
    author: Optional[str] = None,
    mentions: Optional[str] = None,
    hashtags: Optional[List[str]] = None,
    domain: Optional[str] = None,
    language: Optional[str] = None,
    sort: str = "latest",
    since: Optional[str] = None,
    until: Optional[str] = None,
    limit: int = 25
) -> Dict[str, Any]:
    """
    Search Bluesky for posts or users with comprehensive filtering and sorting options.

    This tool allows you to search across Bluesky's content using keywords, phrases, and
    multiple filter criteria. You can search for posts with specific hashtags, by particular
    authors, mentioning certain users, linking to domains, or within date ranges. You can
    also search for user accounts. The tool supports Lucene query syntax for advanced
    searches with AND, OR, and NOT operators.

    Use this tool when you need to find specific content on Bluesky, discover posts about
    topics, find posts by or mentioning users, or search for user accounts.

    Args:
        query (str): The search keywords or phrases to find. This parameter is required and
            cannot be empty.

            - For simple searches: Use plain keywords like "artificial intelligence"
            - For exact phrases: Use quotes like "climate change policy"
            - For advanced searches: Use Lucene syntax with AND, OR, NOT operators
              Example: "python AND (tutorial OR guide) NOT beginner"

            The query is matched against post text content or user names/handles depending
            on the search_type parameter.

        search_type (str, optional): The type of content to search for. Defaults to "posts".

            - "posts": Search through all Bluesky posts (status updates, replies, etc.)
            - "users": Search for user accounts by handle or display name

            This parameter determines which filters are available. Most filters only work
            with "posts" search type.

        author (Optional[str]): Filter posts to only those created by a specific user.
            Only applies when search_type="posts". Defaults to None (search all users).

            Provide the user's handle (e.g., "user.bsky.social") or their DID
            (e.g., "did:plc:..."). This is useful for searching within a specific user's
            posts.

        mentions (Optional[str]): Filter posts to only those mentioning a specific user.
            Only applies when search_type="posts". Defaults to None (no mention filter).

            Provide the mentioned user's handle or DID. This finds posts that @mention
            the specified user.

        hashtags (Optional[List[str]]): Filter posts to only those containing specific hashtags.
            Only applies when search_type="posts". Defaults to None (no hashtag filter).

            IMPORTANT: Provide hashtag text WITHOUT the # symbol.
            - Correct: ["python", "programming"]
            - Incorrect: ["#python", "#programming"]

            When multiple hashtags are provided, posts must contain ALL of them (AND logic).

        domain (Optional[str]): Filter posts to only those containing links to a specific domain.
            Only applies when search_type="posts". Defaults to None (no domain filter).

            Provide just the domain name without protocol or path.
            Examples: "nytimes.com", "github.com", "example.org"

        language (Optional[str]): Filter posts to only those in a specific language.
            Only applies when search_type="posts". Defaults to None (all languages).

            Use ISO 639-1 language codes:
            - "en" - English
            - "es" - Spanish
            - "ja" - Japanese
            - "fr" - French
            - "de" - German
            - "pt" - Portuguese

        sort (str, optional): Sort order for post results. Only applies when search_type="posts".
            Defaults to "latest".

            - "latest": Newest posts first (chronological order, most recent first)
            - "top": Most popular posts first (by engagement metrics)

            User searches are always sorted by relevance and ignore this parameter.

        since (Optional[str]): Filter posts to only those created after this date/time.
            Only applies when search_type="posts". Defaults to None (no start date).

            Accepts two formats:
            - Simple date: "2024-01-15" (automatically converts to midnight UTC)
            - Full datetime: "2024-01-15T10:30:00Z" (ISO 8601 format)

        until (Optional[str]): Filter posts to only those created before this date/time.
            Only applies when search_type="posts". Defaults to None (no end date).

            Accepts two formats:
            - Simple date: "2024-12-31" (automatically converts to midnight UTC)
            - Full datetime: "2024-12-31T23:59:59Z" (ISO 8601 format)

        limit (int, optional): Maximum number of results to return. Defaults to 25.

            Must be between 1 and 100. Higher limits take longer but provide more results.
            Be mindful that very high limits may take significant time to process.

    Returns:
        Dict[str, Any]: A dictionary containing the search results with the following keys:

            Common keys (all searches):
            - status (str): Either "success" or "error"
            - search_type (str): Either "posts" or "users"
            - query (str): The search query that was executed
            - result_count (int): Number of results found

            For successful post searches:
            - posts (List[Dict]): List of post objects, each containing:
                - author (str): The handle of the post author
                - authorDID (str): The DID of the post author
                - uri (str): The AT Protocol URI of the post
                - message (str): The text content of the post
                - posted-datetime (str): When the post was created
                - replies (int): Number of replies
                - reposts (int): Number of reposts
                - likes (int): Number of likes
                - quotes (int): Number of quote posts

            For successful user searches:
            - users (List[Dict]): List of user objects, each containing:
                - handle (str): The user's handle
                - did (str): The user's DID
                - display_name (str): The user's display name
                - bio (str): The user's bio/description
                - followers (int): Follower count
                - following (int): Following count
                - posts (int): Post count
                - avatar_url (str): URL to avatar image (if present)
                - profile_url (str): Link to view profile on bsky.app

            For errors:
            - message (str): Human-readable error description with guidance

    Examples:
        # Simple search for posts about AI
        search_bluesky(query="artificial intelligence")

        # Search for recent posts by a specific user
        search_bluesky(
            query="announcement",
            author="bsky.app"
        )

        # Find posts with specific hashtags (multiple tags = AND logic)
        search_bluesky(
            query="tutorial",
            hashtags=["python", "programming"]
        )

        # Search for popular posts mentioning someone
        search_bluesky(
            query="great work",
            mentions="alice.bsky.social",
            sort="top"
        )

        # Find posts with links to a specific domain
        search_bluesky(
            query="article",
            domain="nytimes.com"
        )

        # Search posts in a specific language
        search_bluesky(
            query="news",
            language="es"
        )

        # Search posts within a date range
        search_bluesky(
            query="conference",
            since="2024-01-01",
            until="2024-12-31"
        )

        # Search for user accounts
        search_bluesky(
            query="software engineer",
            search_type="users"
        )

        # Get top posts with more results
        search_bluesky(
            query="bluesky",
            sort="top",
            since="2024-10-28",
            limit=50
        )

        # Advanced search with Lucene syntax
        search_bluesky(
            query="python AND (tutorial OR guide) NOT beginner"
        )
    """
    try:
        # Validate query
        if not query or len(query.strip()) == 0:
            return {
                "status": "error",
                "message": "Error: The search query parameter is empty. To resolve this, provide keywords or "
                           "phrases to search for, such as 'artificial intelligence' or 'climate change'. "
                           "This is a common mistake and can be fixed by calling the tool again with a valid search query."
            }

        # Validate search_type
        valid_search_types = ["posts", "users"]
        if search_type not in valid_search_types:
            return {
                "status": "error",
                "message": f"Error: The search_type '{search_type}' is not valid. To resolve this, use either 'posts' "
                           f"to search for posts or 'users' to search for user accounts. Feed search is not supported. "
                           f"This is a common parameter error that can be fixed by using one of the supported search types."
            }

        # Validate limit
        if limit < 1 or limit > 100:
            return {
                "status": "error",
                "message": f"Error: The limit parameter {limit} is out of range. To resolve this, choose a value between "
                           f"1 and 100. Lower limits return results faster, while higher limits provide more comprehensive "
                           f"results. This is a validation error that can be fixed by adjusting the limit parameter."
            }

        # Validate sort for posts
        if search_type == "posts" and sort not in ["latest", "top"]:
            return {
                "status": "error",
                "message": f"Error: The sort parameter '{sort}' is not valid for post searches. To resolve this, use "
                           f"'latest' for newest posts first (chronological) or 'top' for most popular posts first "
                           f"(by engagement). This is a parameter validation error that can be fixed by using a supported "
                           f"sort option."
            }

        # Validate hashtags format
        if hashtags:
            for tag in hashtags:
                if tag.startswith('#'):
                    corrected = [t.lstrip('#') for t in hashtags]
                    return {
                        "status": "error",
                        "message": f"Error: The hashtag '{tag}' includes the # symbol, but hashtags should be provided "
                                   f"without the # prefix. To resolve this, use {corrected} instead. This is a common "
                                   f"formatting mistake that can be fixed by removing the # symbol from hashtag values."
                    }

        # Get credentials
        username = os.environ.get("BSKY_USERNAME")
        password = os.environ.get("BSKY_APP_PASSWORD")
        if not username or not password:
            return {
                "status": "error",
                "message": "Error: Missing Bluesky authentication credentials. The BSKY_USERNAME and BSKY_APP_PASSWORD "
                           "environment variables are not set. To resolve this, ask the user to configure these environment "
                           "variables with valid Bluesky credentials. The user can generate an app password in Bluesky "
                           "Settings → App Passwords. This is a configuration issue that the user needs to address before "
                           "you can search Bluesky."
            }

        # Login to Bluesky
        client = Client()
        client.login(username, password)

        if search_type == "posts":
            # Normalize date formats to full ISO 8601 datetime
            normalized_since = _normalize_datetime(since)
            normalized_until = _normalize_datetime(until)

            # Build search parameters
            params = models.AppBskyFeedSearchPosts.Params(
                q=query,
                limit=min(limit, 100)
            )

            # Add optional filters
            if author:
                params.author = author
            if mentions:
                params.mentions = mentions
            if hashtags:
                params.tag = hashtags
            if domain:
                params.domain = domain
            if language:
                params.lang = language
            if sort:
                params.sort = sort
            if normalized_since:
                params.since = normalized_since
            if normalized_until:
                params.until = normalized_until

            # Execute search
            try:
                response = client.app.bsky.feed.search_posts(params)
            except Exception as e:
                return {
                    "status": "error",
                    "message": f"Error: Failed to search posts. The Bluesky API returned: {str(e)}. To resolve this, "
                               f"verify your search parameters are correct and try again. This type of error can occur due "
                               f"to invalid filter combinations, malformed date strings, temporary API issues, or network "
                               f"problems, and usually succeeds on retry."
                }

            # Transform results to compact format
            posts = []
            for item in response.posts:
                record = item.record
                posts.append({
                    "author": item.author.handle,
                    "authorDID": item.author.did,
                    "uri": item.uri,
                    "message": getattr(record, 'text', ''),
                    "posted-datetime": getattr(record, 'created_at', ''),
                    "replies": getattr(item, 'reply_count', 0),
                    "reposts": getattr(item, 'repost_count', 0),
                    "likes": getattr(item, 'like_count', 0),
                    "quotes": getattr(item, 'quote_count', 0)
                })

            return {
                "status": "success",
                "search_type": "posts",
                "query": query,
                "result_count": len(posts),
                "posts": posts
            }

        else:  # search_type == "users"
            # Build search parameters for users
            params = models.AppBskyActorSearchActors.Params(
                q=query,
                limit=min(limit, 100)
            )

            # Execute search
            try:
                response = client.app.bsky.actor.search_actors(params)
            except Exception as e:
                return {
                    "status": "error",
                    "message": f"Error: Failed to search users. The Bluesky API returned: {str(e)}. To resolve this, "
                               f"verify your search query is correct and try again. This type of error can occur due to "
                               f"temporary API issues or network problems, and usually succeeds on retry."
                }

            # Transform results to compact format
            users = []
            for actor in response.actors:
                users.append({
                    "handle": actor.handle,
                    "did": actor.did,
                    "display_name": getattr(actor, 'display_name', '') or actor.handle,
                    "bio": getattr(actor, 'description', '') or '',
                    "followers": getattr(actor, 'followers_count', 0),
                    "following": getattr(actor, 'follows_count', 0),
                    "posts": getattr(actor, 'posts_count', 0),
                    "avatar_url": getattr(actor, 'avatar', None),
                    "profile_url": f"https://bsky.app/profile/{actor.handle}"
                })

            return {
                "status": "success",
                "search_type": "users",
                "query": query,
                "result_count": len(users),
                "users": users
            }

    except ImportError:
        return {
            "status": "error",
            "message": "Error: The atproto Python package is not installed in the execution environment. "
                       "To resolve this, the system administrator needs to install it using 'pip install atproto'. "
                       "This is a dependency issue that prevents the tool from connecting to Bluesky. Once the "
                       "package is installed, this tool will work normally."
        }
    except Exception as e:
        return {
            "status": "error",
            "message": f"Error: An unexpected issue occurred while searching Bluesky: {str(e)}. To resolve this, "
                       f"verify your credentials are correct, check your network connection, and ensure your search "
                       f"parameters are valid. This type of error is uncommon but can usually be resolved by retrying "
                       f"or adjusting the search parameters."
        }