social.py

# async-blockers: no-import-scope
# async-blockers: no-changed-range-scope  # pre-existing patterns surfaced by type-annotation import changes
import asyncio
import os
from datetime import datetime, timezone
from typing import Dict, Any, Callable, List, Awaitable, cast

from pydantic import BaseModel
from ulid import ULID

from database.apps import (
    update_app_in_db,
    upsert_app_to_db,
    get_persona_by_id_db,
    get_persona_by_username_twitter_handle_db,
)
from database.redis_db import delete_generic_cache, save_username
import httpx

from utils.llm.persona import generate_twitter_persona_prompt
from utils.conversations.memories import process_twitter_memories
from utils.executors import db_executor, llm_executor, postprocess_executor, run_blocking
import logging

logger = logging.getLogger(__name__)

rapid_api_host = os.getenv('RAPID_API_HOST')
rapid_api_key = os.getenv('RAPID_API_KEY')

defaultTimeoutSec = 15


class TwitterTweet(BaseModel):
    text: str
    created_at: str
    id: str


class TwitterTimeline(BaseModel):
    timeline: List[TwitterTweet]


class TwitterProfile(BaseModel):
    name: str
    profile: str  # Twitter handle
    rest_id: str
    avatar: str
    desc: str  # Bio description
    friends: int  # Following count
    sub_count: int  # Followers count
    id: str
    status: str = "error"  # Default status for successful profile fetch

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "TwitterProfile":
        """Create a TwitterProfile instance from API response dictionary"""
        return cls(
            name=data.get("name") or "",
            profile=data.get("profile") or "",
            rest_id=data.get("rest_id") or "",
            avatar=(data.get("avatar") or "").replace("_normal", ""),  # Get full-size avatar
            desc=data.get("desc") or "",
            friends=data.get("friends") or 0,
            sub_count=data.get("sub_count") or 0,
            id=data.get("id") or "",
            status=data.get("status", "error"),
        )


async def async_with_retry(operation_name: str, func: Callable[[], Awaitable[Any]]) -> Any:
    max_retries = 5
    base_delay = 1

    for attempt in range(max_retries):
        try:
            return await func()
        except Exception as e:
            delay = base_delay * (2**attempt)
            if attempt == max_retries - 1:
                raise
            logger.error(f"Error in {operation_name} (attempt {attempt + 1}/{max_retries}): {str(e)}")
            logger.warning(f"Retrying in {delay} seconds...")
            await asyncio.sleep(delay)
    raise Exception("Maximum retries exceeded")


async def get_twitter_profile(handle: str) -> TwitterProfile:
    """Fetch Twitter profile for a user and return structured data"""
    url = f"https://{rapid_api_host}/screenname.php?screenname={handle}"

    headers = cast(Dict[str, str], {"X-RapidAPI-Key": rapid_api_key, "X-RapidAPI-Host": rapid_api_host})

    async def fetch_profile():
        async with httpx.AsyncClient(timeout=httpx.Timeout(15.0, connect=2.0)) as client:
            response = await client.get(url, headers=headers)
            if response.status_code == 200:
                data = response.json()
                if data.get('status') == 'error':
                    raise Exception(f"API returned error status: {data.get('message', 'Unknown error')}")

                # Ensure avatar URL is properly formatted (full size)
                if 'avatar' in data and data['avatar'] and '_normal' in data['avatar']:
                    data['avatar'] = data['avatar'].replace('_normal', '')

                return TwitterProfile.from_dict(data)
            # else
            response.raise_for_status()

    return await async_with_retry(f"fetching Twitter profile for {handle}", fetch_profile)


def create_memories_from_twitter_tweets(uid: str, persona_id: str, tweets: List[TwitterTweet]) -> None:
    """Create individual memories from tweets for more detailed persona information"""
    # Combine tweets into a single text for memory extraction
    combined_text = "\n".join([f"{tweet.text} (Posted: {tweet.created_at})" for tweet in tweets])

    # Process tweets and extract memories using the dedicated function
    process_twitter_memories(uid, combined_text, persona_id)


async def get_twitter_timeline(handle: str) -> TwitterTimeline:
    """Fetch Twitter timeline for a user and return structured data"""
    logger.info(f"Fetching Twitter timeline for {handle}...")
    url = f"https://{rapid_api_host}/timeline.php?screenname={handle}"

    headers = cast(Dict[str, str], {"X-RapidAPI-Key": rapid_api_key, "X-RapidAPI-Host": rapid_api_host})

    async def fetch_timeline():
        async with httpx.AsyncClient(timeout=httpx.Timeout(15.0, connect=2.0)) as client:
            response = await client.get(url, headers=headers)
            if response.status_code == 200:
                data = response.json()
                if data.get('status') == 'error':
                    raise Exception(f"API returned error status: {data.get('message', 'Unknown error')}")

                # Convert raw timeline to structured model
                timeline_data = data.get('timeline', [])
                tweets = [
                    TwitterTweet(text=tweet['text'], created_at=tweet['created_at'], id=tweet['tweet_id'])
                    for tweet in timeline_data
                ]

                return TwitterTimeline(timeline=tweets)
            # else
            response.raise_for_status()

    return await async_with_retry(f"fetching Twitter timeline for {handle}", fetch_timeline)


async def verify_latest_tweet(username: str, handle: str) -> Dict[str, Any]:
    """Verify if the latest tweet contains verification text"""
    logger.info(f"Fetching latest tweet for {handle}, username {username}...")

    # Get timeline
    timeline = await get_twitter_timeline(handle)

    # Check if there are any tweets
    if not timeline.timeline:
        return {"tweet": "", "verified": False}

    # Get the latest tweet (first in the timeline)
    latest_tweet = timeline.timeline[0]

    # Check if it contains verification text
    if f'Verifying my clone({username})' in latest_tweet.text:
        return {"tweet": latest_tweet.text, "verified": True}

    return {"tweet": latest_tweet.text, "verified": False}


async def upsert_persona_from_twitter_profile(username: str, handle: str, uid: str) -> Dict[str, Any]:
    """Create or update a persona based on Twitter profile and generate memories"""
    # Get Twitter profile data
    profile = await get_twitter_profile(handle)

    # Get tweets
    timeline = await get_twitter_timeline(handle)

    # Create or update persona
    persona = await run_blocking(db_executor, _create_or_update_persona, profile, username, uid, handle)

    # Generate persona prompt from tweets
    formatted_tweets = [{'tweet': tweet.text, 'posted_at': tweet.created_at} for tweet in timeline.timeline]
    persona_prompt = await run_blocking(
        llm_executor, generate_twitter_persona_prompt, formatted_tweets, persona["name"]
    )
    persona['persona_prompt'] = persona_prompt

    # Save persona to database
    await run_blocking(db_executor, upsert_app_to_db, persona)
    await run_blocking(db_executor, save_username, username, uid)
    await run_blocking(db_executor, delete_generic_cache, 'get_public_approved_apps_data')

    # Create memories from persona prompt and tweets
    await run_blocking(postprocess_executor, create_memories_from_twitter_tweets, uid, persona['id'], timeline.timeline)

    return persona


def _create_or_update_persona(profile: TwitterProfile, username: str, uid: str, handle: str) -> Dict[str, Any]:
    """Create a new persona or update an existing one"""
    persona = get_persona_by_username_twitter_handle_db(username, handle)

    # Create new persona if it doesn't exist
    if not persona:
        persona = cast(
            Dict[str, Any],
            {
                "name": profile.name,
                "author": profile.name,
                "uid": uid,
                "id": str(ULID()),
                "status": "approved",
                "capabilities": ["persona"],
                "username": username,
                "connected_accounts": ["twitter"],
                "description": profile.desc,
                "image": profile.avatar,
                "category": "personality-emulation",
                "approved": True,
                "private": False,
                "created_at": datetime.now(timezone.utc),
            },
        )

    # Update persona with Twitter data
    persona["twitter"] = {
        "username": profile.profile,
        "avatar": profile.avatar,
        "connected_at": datetime.now(timezone.utc),
    }

    # Ensure persona is published
    persona["status"] = "approved"
    persona["approved"] = True
    persona["private"] = False

    return persona


async def add_twitter_to_persona(handle: str, persona_id: str) -> Dict[str, Any]:
    """Add Twitter account to an existing persona"""
    persona = await run_blocking(db_executor, get_persona_by_id_db, persona_id)
    if persona is None:
        raise ValueError(f"Persona not found: {persona_id}")
    profile = await get_twitter_profile(handle)

    if 'twitter' not in persona['connected_accounts']:
        persona['connected_accounts'].append('twitter')

    persona['twitter'] = {
        "username": profile.profile,
        "avatar": profile.avatar,
        "connected_at": datetime.now(timezone.utc),
    }

    await run_blocking(db_executor, update_app_in_db, persona)
    await run_blocking(db_executor, delete_generic_cache, 'get_public_approved_apps_data')

    # Get tweets from the Twitter timeline
    timeline = await get_twitter_timeline(handle)

    # Create memories from the tweets
    if timeline and timeline.timeline:
        await run_blocking(
            postprocess_executor, create_memories_from_twitter_tweets, persona['uid'], persona_id, timeline.timeline
        )

    return persona