openclaw-backups/skills/threads-reader/scripts/threads.py

#!/usr/bin/env python3
"""
Threads Reader - Fetch feed, trending posts, and user content
"""

import os
import sys
import json
import asyncio
import argparse
from pathlib import Path

try:
    from threads_api.src.threads_api import ThreadsAPI
except ImportError:
    print("Error: threads-api not installed. Run: pip3 install threads-api", file=sys.stderr)
    sys.exit(1)

CONFIG_FILE = Path.home() / ".config/threads/config.json"
TOKEN_CACHE = Path.home() / ".config/threads/.token"

def load_config():
    """Load credentials from env vars or config file"""
    username = os.getenv("INSTAGRAM_USERNAME")
    password = os.getenv("INSTAGRAM_PASSWORD")
    
    if not username and CONFIG_FILE.exists():
        with open(CONFIG_FILE) as f:
            config = json.load(f)
            username = config.get("username")
            password = config.get("password")
    
    if not username or not password:
        print("Error: Set INSTAGRAM_USERNAME and INSTAGRAM_PASSWORD", file=sys.stderr)
        print("Or create config at ~/.config/threads/config.json", file=sys.stderr)
        sys.exit(1)
    
    return username, password

async def get_feed(limit=10):
    """Fetch For You feed"""
    username, password = load_config()
    
    api = ThreadsAPI()
    
    # Create token cache directory
    TOKEN_CACHE.parent.mkdir(parents=True, exist_ok=True)
    
    try:
        # Login with cached token
        await api.login(username, password, cached_token_path=str(TOKEN_CACHE))
        
        # Get recommended feed
        feed = await api.get_timeline()
        
        posts = []
        if feed and 'data' in feed:
            threads = feed.get('data', {}).get('mediaData', {}).get('threads', [])
            
            for thread in threads[:limit]:
                thread_items = thread.get('thread_items', [])
                if thread_items:
                    post = thread_items[0].get('post', {})
                    user = post.get('user', {})
                    
                    posts.append({
                        'author': user.get('username', 'unknown'),
                        'text': post.get('caption', {}).get('text', ''),
                        'likes': post.get('like_count', 0),
                        'replies': post.get('text_post_app_info', {}).get('direct_reply_count', 0),
                        'created_at': post.get('taken_at', 0),
                        'url': f"https://www.threads.net/@{user.get('username')}/post/{post.get('code', '')}"
                    })
        
        await api.close_gracefully()
        return posts
        
    except Exception as e:
        print(f"Error fetching feed: {e}", file=sys.stderr)
        await api.close_gracefully()
        return []

async def get_trending(limit=5):
    """Fetch trending/popular posts"""
    # For now, use regular feed and filter by engagement
    posts = await get_feed(limit=50)
    
    # Sort by engagement (likes + replies)
    posts.sort(key=lambda p: p['likes'] + p['replies'], reverse=True)
    
    return posts[:limit]

async def get_user_posts(username, limit=5):
    """Fetch specific user's posts"""
    user, password = load_config()
    
    api = ThreadsAPI()
    TOKEN_CACHE.parent.mkdir(parents=True, exist_ok=True)
    
    try:
        await api.login(user, password, cached_token_path=str(TOKEN_CACHE))
        
        # Get user ID first
        user_data = await api.get_user_id_from_username(username.lstrip('@'))
        
        if not user_data:
            print(f"Error: User {username} not found", file=sys.stderr)
            await api.close_gracefully()
            return []
        
        user_id = user_data
        
        # Get user's threads
        profile = await api.get_user_profile_threads(user_id)
        
        posts = []
        if profile and 'data' in profile:
            threads = profile.get('data', {}).get('mediaData', {}).get('threads', [])
            
            for thread in threads[:limit]:
                thread_items = thread.get('thread_items', [])
                if thread_items:
                    post = thread_items[0].get('post', {})
                    user_obj = post.get('user', {})
                    
                    posts.append({
                        'author': user_obj.get('username', username),
                        'text': post.get('caption', {}).get('text', ''),
                        'likes': post.get('like_count', 0),
                        'replies': post.get('text_post_app_info', {}).get('direct_reply_count', 0),
                        'created_at': post.get('taken_at', 0),
                        'url': f"https://www.threads.net/@{user_obj.get('username')}/post/{post.get('code', '')}"
                    })
        
        await api.close_gracefully()
        return posts
        
    except Exception as e:
        print(f"Error fetching user posts: {e}", file=sys.stderr)
        await api.close_gracefully()
        return []

def main():
    parser = argparse.ArgumentParser(description='Threads Reader')
    subparsers = parser.add_subparsers(dest='command', help='Commands')
    
    # Feed command
    feed_parser = subparsers.add_parser('feed', help='Get For You feed')
    feed_parser.add_argument('--limit', type=int, default=10, help='Number of posts')
    
    # Trending command
    trending_parser = subparsers.add_parser('trending', help='Get trending posts')
    trending_parser.add_argument('--limit', type=int, default=5, help='Number of posts')
    
    # User command
    user_parser = subparsers.add_parser('user', help='Get user posts')
    user_parser.add_argument('username', help='Username (with or without @)')
    user_parser.add_argument('--limit', type=int, default=5, help='Number of posts')
    
    args = parser.parse_args()
    
    if not args.command:
        parser.print_help()
        sys.exit(1)
    
    # Run async command
    if args.command == 'feed':
        posts = asyncio.run(get_feed(args.limit))
    elif args.command == 'trending':
        posts = asyncio.run(get_trending(args.limit))
    elif args.command == 'user':
        posts = asyncio.run(get_user_posts(args.username, args.limit))
    else:
        parser.print_help()
        sys.exit(1)
    
    # Output JSON
    print(json.dumps(posts, indent=2))

if __name__ == '__main__':
    main()