import logging from typing import Union, Dict import praw import random import string from tenacity import retry, stop_after_attempt, wait_exponential # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class RedditError(Exception): """Custom exception for Reddit API errors""" pass def validate_reddit_credentials(client_id: str, client_secret: str, username: str, password: str) -> bool: """Validate Reddit credentials""" return all([client_id, client_secret, username, password]) def generate_random_user_agent(): """Generate a random user agent.""" prefix = "my_script_by_u/" username = "your_reddit_username" # Replace with your Reddit username random_suffix = ''.join(random.choices(string.ascii_letters + string.digits, k=10)) return f"{prefix}{username}_{random_suffix}" @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True ) def reddit_post( client_id: str, client_secret: str, username: str, password: str, subreddit: str, title: str, body: str = "" ) -> Union[Dict[str, str], Dict[str, str]]: """ Create a Reddit post with improved error handling and retries Args: client_id (str): Reddit API client ID client_secret (str): Reddit API client secret username (str): Reddit username password (str): Reddit password subreddit (str): Target subreddit title (str): Post title body (str): Post content Returns: dict: Post information if successful dict: Error information if failed """ try: if not validate_reddit_credentials(client_id, client_secret, username, password): return {"error": "Invalid or missing Reddit credentials"} if not title: return {"error": "Post title is required"} if not subreddit: return {"error": "Subreddit is required"} # Initialize Reddit client reddit = praw.Reddit( client_id=client_id, client_secret=client_secret, username=username, password=password, user_agent=f"python:flowify:v1.0 (by /u/{username})" ) # Verify credentials try: reddit.user.me() except Exception: return {"error": "Failed to authenticate with Reddit"} # Create post try: subreddit_instance = reddit.subreddit(subreddit) post = subreddit_instance.submit( title=title, selftext=body, send_replies=True ) return { "success": True, "post_id": post.id, "url": f"https://reddit.com{post.permalink}" } except praw.exceptions.RedditAPIException as e: error_messages = [f"{error.error_type}: {error.message}" for error in e.items] return {"error": f"Reddit API error: {', '.join(error_messages)}"} except praw.exceptions.PRAWException as e: logger.error(f"PRAW error: {str(e)}") return {"error": f"Reddit error: {str(e)}"} except Exception as e: logger.error(f"Unexpected error creating Reddit post: {str(e)}") return {"error": f"Failed to create post: {str(e)}"}