bluesky_poster.py

#!/usr/bin/env python3
"""
Bluesky posting wrapper with app password authentication and session persistence.

Usage:
    python bluesky_poster.py "Your post text here"
    
Or import and use programmatically:
    from bluesky_poster import BlueskyPoster
    poster = BlueskyPoster()
    poster.post('Hello Bluesky!')
"""

import os
import sys
import json
import time
from typing import Optional
from pathlib import Path
from atproto import Client
from atproto.exceptions import AtProtocolError
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()


class BlueskyPoster:
    """Bluesky posting wrapper with session persistence."""
    
    def __init__(self, session_file: Optional[str] = None, verbose: int = 0):
        """
        Initialize the Bluesky poster.
        
        Args:
            session_file: Path to store session data (default: ~/.bluesky_session.json)
            verbose: Verbosity level (0=quiet, 1=show post URL and session info)
        """
        self.session_file = session_file or os.path.expanduser('~/.bluesky_session.json')
        self.verbose = verbose
        self.client = Client()
        self._authenticated = False
        self.session_data = {}
        self._load_session()
    
    def _load_session(self) -> None:
        """Load existing session data from file."""
        if os.path.exists(self.session_file):
            try:
                with open(self.session_file, 'r') as f:
                    self.session_data = json.load(f)
            except (json.JSONDecodeError, IOError):
                self.session_data = {}
    
    def _save_session(self, session_string: str, handle: str) -> None:
        """Save session data to file."""
        try:
            self.session_data = {
                'session_string': session_string,
                'handle': handle,
                'timestamp': time.time()
            }
            
            os.makedirs(os.path.dirname(self.session_file), exist_ok=True)
            with open(self.session_file, 'w') as f:
                json.dump(self.session_data, f, indent=2)
            os.chmod(self.session_file, 0o600)  # Secure permissions
        except IOError as e:
            print(f"Warning: Could not save session data: {e}")
    
    def _try_existing_session(self) -> bool:
        """Try to use existing session from file."""
        session_string = self.session_data.get('session_string')
        handle = self.session_data.get('handle')
        
        if not session_string or not handle:
            return False
        
        try:
            # Try to restore the session
            self.client.login(handle, session_string=session_string)
            self._authenticated = True
            
            # Export and save the potentially refreshed session
            new_session_string = self.client.export_session_string()
            if new_session_string != session_string:
                self._save_session(new_session_string, handle)
            
            return True
        except Exception as e:
            print(f"Session restoration failed, will re-authenticate: {e}")
            return False
    
    def authenticate(self) -> bool:
        """
        Authenticate with Bluesky using app password from environment variables.
        
        Returns:
            bool: True if authentication successful
        """
        # Try existing session first
        if self._try_existing_session():
            print("✅ Restored existing session")
            if self.verbose >= 1:
                print(f"📁 Session file: {self.session_file}")
                print(f"👤 Handle: {self.session_data.get('handle', 'unknown')}")
            return True
        
        # Get credentials from environment
        handle = os.getenv('BLUESKY_HANDLE')
        password = os.getenv('BLUESKY_PASSWORD')
        
        if not handle or not password:
            print("❌ Missing BLUESKY_HANDLE or BLUESKY_PASSWORD in .env file")
            print("Please add these to your .env file:")
            print("BLUESKY_HANDLE=yourhandle.bsky.social")
            print("BLUESKY_PASSWORD=your-app-password")
            return False
        
        try:
            # Login with app password
            self.client.login(handle, password)
            self._authenticated = True
            
            # Save the session for future use
            session_string = self.client.export_session_string()
            self._save_session(session_string, handle)
            
            print("✅ Authentication successful")
            if self.verbose >= 1:
                print(f"📁 Session saved to: {self.session_file}")
                print(f"👤 Handle: {handle}")
            return True
            
        except AtProtocolError as e:
            if 'Invalid identifier or password' in str(e):
                print("❌ Invalid credentials. Make sure you're using an app password, not your main password.")
                print("Generate an app password at: https://bsky.app/settings/app-passwords")
            else:
                print(f"❌ Authentication failed: {e}")
            return False
        except Exception as e:
            print(f"❌ Authentication failed: {e}")
            return False
    
    def post(self, text: str) -> Optional[str]:
        """
        Post text to Bluesky.
        
        Args:
            text: The text content to post
            
        Returns:
            Optional[str]: Post URI if successful, None if failed
        """
        if not self._authenticated:
            if not self.authenticate():
                return None
        
        try:
            response = self.client.send_post(text)
            
            # Try to save refreshed session if tokens were updated
            try:
                session_string = self.client.export_session_string()
                handle = self.session_data.get('handle') or os.getenv('BLUESKY_HANDLE')
                if session_string and handle:
                    self._save_session(session_string, handle)
            except:
                pass  # Session save is optional, don't fail the post
            
            if self.verbose >= 1 and response.uri:
                # Convert AT URI to web URL
                handle = self.session_data.get('handle') or os.getenv('BLUESKY_HANDLE')
                post_id = response.uri.split('/')[-1]
                web_url = f"https://bsky.app/profile/{handle}/post/{post_id}"
                print(f"🔗 Post URL: {web_url}")
            
            return response.uri
        except Exception as e:
            print(f"Failed to post: {e}")
            
            # Try to re-authenticate once if it failed
            if "Invalid token" in str(e) or "ExpiredToken" in str(e):
                print("Session expired, re-authenticating...")
                self._authenticated = False
                if self.authenticate():
                    try:
                        response = self.client.send_post(text)
                        return response.uri
                    except Exception as e2:
                        print(f"Failed to post after re-authentication: {e2}")
            
            return None
    
    def post_with_link(self, text: str, link_url: str, link_title: str = "") -> Optional[str]:
        """
        Post text with an embedded link.
        
        Args:
            text: The text content to post
            link_url: URL to embed
            link_title: Optional title for the link
            
        Returns:
            Optional[str]: Post URI if successful, None if failed
        """
        if not self._authenticated:
            if not self.authenticate():
                return None
        
        try:
            # Create embed for the link
            embed = {
                '$type': 'app.bsky.embed.external',
                'external': {
                    'uri': link_url,
                    'title': link_title or link_url,
                    'description': ''
                }
            }
            
            response = self.client.send_post(text, embed=embed)
            
            # Try to save refreshed session
            try:
                session_string = self.client.export_session_string()
                handle = self.session_data.get('handle') or os.getenv('BLUESKY_HANDLE')
                if session_string and handle:
                    self._save_session(session_string, handle)
            except:
                pass
            
            return response.uri
        except Exception as e:
            print(f"Failed to post with link: {e}")
            return None


def main():
    """Command line interface for posting to Bluesky."""
    import argparse
    
    parser = argparse.ArgumentParser(description='Post to Bluesky')
    parser.add_argument('text', help='The text to post')
    parser.add_argument('-v', '--verbose', action='count', default=0,
                        help='Increase verbosity (use -v for verbose output)')
    
    args = parser.parse_args()
    
    poster = BlueskyPoster(verbose=args.verbose)
    
    if not poster.authenticate():
        print("❌ Authentication failed")
        sys.exit(1)
    
    post_uri = poster.post(args.text)
    
    if post_uri:
        print(f"✅ Posted successfully!")
        if args.verbose == 0:  # Only show URI if not verbose (verbose shows web URL)
            print(f"Post URI: {post_uri}")
    else:
        print("❌ Failed to post")
        sys.exit(1)


if __name__ == '__main__':
    main()