Spaces:
Sleeping
Sleeping
File size: 2,079 Bytes
c71926b e5a6e85 7c0c559 c71926b e5a6e85 8280ddc c71926b 8280ddc cbe7e39 1f4bc54 cbe7e39 dc9605a cbe7e39 8280ddc e07522e 8280ddc a25e20b dc9605a 8280ddc 7c0c559 8280ddc cbe7e39 c71926b 8280ddc 60c6493 8280ddc c71926b dc9605a 8280ddc c71926b 8280ddc c71926b | 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 | import os
import time
import praw
import logging
from dotenv import load_dotenv
# Load environment variables
load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), '..', '.env'))
# Initialize logging (optional: configure this in your app entrypoint)
logging.basicConfig(level=logging.INFO)
# Load multiple Reddit clients
clients = []
for i in range(1, 4):
cid = os.getenv(f"REDDIT_CLIENT_ID_{i}")
csecret = os.getenv(f"REDDIT_CLIENT_SECRET_{i}")
agent = os.getenv(f"REDDIT_USER_AGENT_{i}")
if cid and csecret and agent:
try:
reddit = praw.Reddit(
client_id=cid,
client_secret=csecret,
user_agent=agent
)
reddit.read_only = True
clients.append(reddit)
except Exception as e:
logging.error(f"Failed to initialize Reddit client #{i}: {e}")
if not clients:
raise ValueError("❌ No valid Reddit clients loaded from environment variables.")
# Rotate between clients to avoid rate limits
def get_reddit_client():
return clients[int(time.time()) % len(clients)]
# Main function to search Reddit
def search_reddit(brand, limit=10, retries=3):
"""
Search Reddit for a given brand keyword.
Args:
brand (str): Keyword to search.
limit (int): Number of posts to fetch. Default is 10.
retries (int): Retry attempts in case of failure.
Returns:
list: List of PRAW submission objects.
"""
for attempt in range(retries):
try:
reddit = get_reddit_client()
posts = list(
reddit.subreddit("all").search(
brand, sort="new", limit=limit
)
)
logging.info(f"✅ Fetched {len(posts)} Reddit posts for brand: {brand}")
return posts
except Exception as e:
logging.warning(f"[Retry {attempt + 1}/{retries}] Reddit search failed: {e}")
time.sleep(2)
logging.error(f"❌ All retries failed for brand '{brand}'")
return []
|