Spaces:
Runtime error
Runtime error
File size: 3,392 Bytes
ce4e319 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 |
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)}"} |