From a623e5c12dce20e7f4bda992d4a1c20e527b61ea Mon Sep 17 00:00:00 2001 From: Sanjeev Kumar Date: Sat, 13 Dec 2025 22:35:09 +0530 Subject: [PATCH] feat: Async scraper, cloud upload, subreddit stats --- analytics/subreddit_stats.py | 243 ++++++++++++++++++++++ export/cloud.py | 313 ++++++++++++++++++++++++++++ requirements.txt | 4 + scraper/__init__.py | 2 + scraper/async_scraper.py | 381 +++++++++++++++++++++++++++++++++++ 5 files changed, 943 insertions(+) create mode 100644 analytics/subreddit_stats.py create mode 100644 export/cloud.py create mode 100644 scraper/__init__.py create mode 100644 scraper/async_scraper.py diff --git a/analytics/subreddit_stats.py b/analytics/subreddit_stats.py new file mode 100644 index 0000000..aca7b6b --- /dev/null +++ b/analytics/subreddit_stats.py @@ -0,0 +1,243 @@ +""" +Subreddit Statistics - Subscribers, rules, mods, and metadata +""" +import requests +from datetime import datetime +import json + +USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + +def get_subreddit_about(subreddit): + """ + Fetch subreddit metadata (subscribers, description, rules, etc.) + + Args: + subreddit: Subreddit name (without r/) + + Returns: + Dictionary with subreddit info + """ + url = f"https://old.reddit.com/r/{subreddit}/about.json" + + try: + response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=15) + + if response.status_code != 200: + print(f"āŒ Failed to fetch r/{subreddit} info: {response.status_code}") + return None + + data = response.json()['data'] + + return { + "name": data.get('display_name'), + "title": data.get('title'), + "description": data.get('public_description'), + "subscribers": data.get('subscribers', 0), + "active_users": data.get('accounts_active', 0), + "created_utc": datetime.fromtimestamp(data.get('created_utc', 0)).isoformat(), + "over_18": data.get('over18', False), + "subreddit_type": data.get('subreddit_type'), # public, private, restricted + "lang": data.get('lang'), + "icon_url": data.get('icon_img', '').split('?')[0] if data.get('icon_img') else None, + "banner_url": data.get('banner_img', '').split('?')[0] if data.get('banner_img') else None, + "header_url": data.get('header_img'), + "community_icon": data.get('community_icon', '').split('?')[0] if data.get('community_icon') else None, + "wiki_enabled": data.get('wiki_enabled', False), + "spoilers_enabled": data.get('spoilers_enabled', False), + "allow_videos": data.get('allow_videos', False), + "allow_images": data.get('allow_images', False), + "allow_polls": data.get('allow_polls', False), + } + except Exception as e: + print(f"āŒ Error fetching subreddit info: {e}") + return None + +def get_subreddit_rules(subreddit): + """ + Fetch subreddit rules. + + Args: + subreddit: Subreddit name + + Returns: + List of rule dictionaries + """ + url = f"https://old.reddit.com/r/{subreddit}/about/rules.json" + + try: + response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=15) + + if response.status_code != 200: + return [] + + data = response.json() + rules = [] + + for rule in data.get('rules', []): + rules.append({ + "short_name": rule.get('short_name'), + "description": rule.get('description'), + "priority": rule.get('priority'), + "kind": rule.get('kind'), # link, comment, all + "created_utc": datetime.fromtimestamp(rule.get('created_utc', 0)).isoformat() + }) + + return rules + except Exception as e: + print(f"āŒ Error fetching rules: {e}") + return [] + +def get_subreddit_mods(subreddit): + """ + Fetch subreddit moderators. + + Args: + subreddit: Subreddit name + + Returns: + List of moderator usernames + """ + url = f"https://old.reddit.com/r/{subreddit}/about/moderators.json" + + try: + response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=15) + + if response.status_code != 200: + return [] + + data = response.json() + mods = [] + + for mod in data.get('data', {}).get('children', []): + mods.append({ + "name": mod.get('name'), + "permissions": mod.get('mod_permissions', []), + "added_utc": datetime.fromtimestamp(mod.get('date', 0)).isoformat() if mod.get('date') else None + }) + + return mods + except Exception as e: + print(f"āŒ Error fetching mods: {e}") + return [] + +def get_subreddit_flairs(subreddit): + """ + Fetch available post flairs. + + Args: + subreddit: Subreddit name + + Returns: + List of flair options + """ + url = f"https://old.reddit.com/r/{subreddit}/api/link_flair_v2.json" + + try: + response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=15) + + if response.status_code != 200: + return [] + + flairs = [] + for flair in response.json(): + flairs.append({ + "text": flair.get('text'), + "id": flair.get('id'), + "background_color": flair.get('background_color'), + "text_color": flair.get('text_color'), + "type": flair.get('type') + }) + + return flairs + except Exception as e: + return [] + +def get_full_subreddit_stats(subreddit): + """ + Get comprehensive subreddit statistics. + + Args: + subreddit: Subreddit name + + Returns: + Dictionary with all stats + """ + print(f"šŸ“Š Fetching stats for r/{subreddit}...") + + about = get_subreddit_about(subreddit) + + if not about: + return None + + rules = get_subreddit_rules(subreddit) + mods = get_subreddit_mods(subreddit) + flairs = get_subreddit_flairs(subreddit) + + stats = { + **about, + "rules": rules, + "rules_count": len(rules), + "moderators": mods, + "moderator_count": len(mods), + "flairs": flairs, + "flair_count": len(flairs), + "fetched_at": datetime.now().isoformat() + } + + # Print summary + print(f"\nšŸ“Š r/{subreddit} Statistics:") + print(f" šŸ‘„ Subscribers: {stats['subscribers']:,}") + print(f" šŸŸ¢ Active Users: {stats['active_users']:,}") + print(f" šŸ“œ Rules: {stats['rules_count']}") + print(f" šŸ‘® Moderators: {stats['moderator_count']}") + print(f" šŸ·ļø Flairs: {stats['flair_count']}") + print(f" šŸ“… Created: {stats['created_utc'][:10]}") + print(f" šŸ”ž NSFW: {stats['over_18']}") + + return stats + +def save_subreddit_stats(subreddit, output_dir="data"): + """ + Fetch and save subreddit stats to JSON. + + Args: + subreddit: Subreddit name + output_dir: Output directory + + Returns: + Path to saved file + """ + import os + + stats = get_full_subreddit_stats(subreddit) + + if not stats: + return None + + save_dir = f"{output_dir}/r_{subreddit}" + os.makedirs(save_dir, exist_ok=True) + + filepath = f"{save_dir}/subreddit_stats.json" + + with open(filepath, 'w', encoding='utf-8') as f: + json.dump(stats, f, indent=2, ensure_ascii=False) + + print(f"\nšŸ’¾ Saved to {filepath}") + return filepath + +# CLI for testing +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Subreddit Statistics") + parser.add_argument("subreddit", help="Subreddit name") + parser.add_argument("--save", action="store_true", help="Save to JSON") + + args = parser.parse_args() + + if args.save: + save_subreddit_stats(args.subreddit) + else: + stats = get_full_subreddit_stats(args.subreddit) + if stats: + print(f"\nšŸ“ Description: {stats['description'][:200]}..." if stats['description'] else "") diff --git a/export/cloud.py b/export/cloud.py new file mode 100644 index 0000000..132234e --- /dev/null +++ b/export/cloud.py @@ -0,0 +1,313 @@ +""" +Cloud Upload Module - S3 and Google Drive integration +""" +import os +import json +from pathlib import Path +from datetime import datetime + +# Try importing boto3 for S3 +try: + import boto3 + from botocore.exceptions import ClientError + HAS_BOTO3 = True +except ImportError: + HAS_BOTO3 = False + +# Try importing Google Drive API +try: + from google.oauth2.credentials import Credentials + from googleapiclient.discovery import build + from googleapiclient.http import MediaFileUpload + HAS_GDRIVE = True +except ImportError: + HAS_GDRIVE = False + + +class S3Uploader: + """Upload scraped data to AWS S3.""" + + def __init__(self, bucket_name, aws_access_key=None, aws_secret_key=None, region='us-east-1'): + """ + Initialize S3 uploader. + + Args: + bucket_name: S3 bucket name + aws_access_key: Optional, uses env/config if not provided + aws_secret_key: Optional, uses env/config if not provided + region: AWS region + """ + if not HAS_BOTO3: + raise ImportError("boto3 not installed. Run: pip install boto3") + + self.bucket_name = bucket_name + self.region = region + + # Use provided credentials or fall back to env vars + self.s3 = boto3.client( + 's3', + aws_access_key_id=aws_access_key or os.getenv('AWS_ACCESS_KEY_ID'), + aws_secret_access_key=aws_secret_key or os.getenv('AWS_SECRET_ACCESS_KEY'), + region_name=region + ) + + def upload_file(self, local_path, s3_key=None): + """ + Upload a single file to S3. + + Args: + local_path: Local file path + s3_key: S3 object key (default: same as filename) + + Returns: + S3 URL or None on failure + """ + local_path = Path(local_path) + + if not local_path.exists(): + print(f"āŒ File not found: {local_path}") + return None + + s3_key = s3_key or local_path.name + + try: + self.s3.upload_file(str(local_path), self.bucket_name, s3_key) + url = f"https://{self.bucket_name}.s3.{self.region}.amazonaws.com/{s3_key}" + print(f"āœ… Uploaded: {s3_key}") + return url + except ClientError as e: + print(f"āŒ S3 upload failed: {e}") + return None + + def upload_directory(self, local_dir, s3_prefix=""): + """ + Upload entire directory to S3. + + Args: + local_dir: Local directory path + s3_prefix: Prefix for S3 keys + + Returns: + Dictionary of uploaded files + """ + local_dir = Path(local_dir) + + if not local_dir.exists(): + print(f"āŒ Directory not found: {local_dir}") + return {} + + uploaded = {} + + for file_path in local_dir.rglob('*'): + if file_path.is_file(): + relative_path = file_path.relative_to(local_dir) + s3_key = f"{s3_prefix}/{relative_path}" if s3_prefix else str(relative_path) + s3_key = s3_key.replace('\\', '/') # Windows path fix + + url = self.upload_file(file_path, s3_key) + if url: + uploaded[str(relative_path)] = url + + print(f"\nšŸ“¤ Uploaded {len(uploaded)} files to S3") + return uploaded + + def upload_subreddit_data(self, subreddit, prefix="u"): + """ + Upload all data for a subreddit. + + Args: + subreddit: Subreddit name + prefix: "r" for subreddit, "u" for user + + Returns: + Upload results + """ + data_dir = Path(f"data/{prefix}_{subreddit}") + + if not data_dir.exists(): + print(f"āŒ Data not found for {prefix}/{subreddit}") + return {} + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + s3_prefix = f"reddit/{prefix}_{subreddit}/{timestamp}" + + return self.upload_directory(data_dir, s3_prefix) + + def list_uploads(self, prefix="reddit/"): + """List all uploaded data in S3.""" + try: + response = self.s3.list_objects_v2( + Bucket=self.bucket_name, + Prefix=prefix + ) + + objects = response.get('Contents', []) + + print(f"\nšŸ“ S3 Contents ({self.bucket_name}/{prefix}):") + for obj in objects[:50]: # Limit to 50 + size_kb = obj['Size'] / 1024 + print(f" {obj['Key']} ({size_kb:.1f} KB)") + + if len(objects) > 50: + print(f" ... and {len(objects) - 50} more") + + return objects + except ClientError as e: + print(f"āŒ S3 list failed: {e}") + return [] + + +class GDriveUploader: + """Upload scraped data to Google Drive.""" + + def __init__(self, credentials_file='credentials.json', token_file='token.json'): + """ + Initialize Google Drive uploader. + + Args: + credentials_file: Path to OAuth credentials JSON + token_file: Path to token JSON + """ + if not HAS_GDRIVE: + raise ImportError("Google API client not installed. Run: pip install google-api-python-client google-auth-oauthlib") + + self.credentials_file = credentials_file + self.token_file = token_file + self.service = None + self._authenticate() + + def _authenticate(self): + """Authenticate with Google Drive API.""" + creds = None + + if os.path.exists(self.token_file): + creds = Credentials.from_authorized_user_file(self.token_file) + + if not creds or not creds.valid: + if creds and creds.expired and creds.refresh_token: + creds.refresh(Request()) + else: + from google_auth_oauthlib.flow import InstalledAppFlow + SCOPES = ['https://www.googleapis.com/auth/drive.file'] + flow = InstalledAppFlow.from_client_secrets_file(self.credentials_file, SCOPES) + creds = flow.run_local_server(port=0) + + with open(self.token_file, 'w') as token: + token.write(creds.to_json()) + + self.service = build('drive', 'v3', credentials=creds) + print("āœ… Google Drive authenticated") + + def create_folder(self, name, parent_id=None): + """Create a folder in Google Drive.""" + metadata = { + 'name': name, + 'mimeType': 'application/vnd.google-apps.folder' + } + + if parent_id: + metadata['parents'] = [parent_id] + + folder = self.service.files().create(body=metadata, fields='id').execute() + return folder.get('id') + + def upload_file(self, local_path, folder_id=None): + """Upload a file to Google Drive.""" + local_path = Path(local_path) + + if not local_path.exists(): + print(f"āŒ File not found: {local_path}") + return None + + metadata = {'name': local_path.name} + if folder_id: + metadata['parents'] = [folder_id] + + media = MediaFileUpload(str(local_path), resumable=True) + + try: + file = self.service.files().create( + body=metadata, + media_body=media, + fields='id,webViewLink' + ).execute() + + print(f"āœ… Uploaded: {local_path.name}") + return file.get('webViewLink') + except Exception as e: + print(f"āŒ Upload failed: {e}") + return None + + def upload_subreddit_data(self, subreddit, prefix="r"): + """Upload all data for a subreddit.""" + data_dir = Path(f"data/{prefix}_{subreddit}") + + if not data_dir.exists(): + print(f"āŒ Data not found for {prefix}/{subreddit}") + return {} + + # Create folder structure + root_folder = self.create_folder(f"reddit_{prefix}_{subreddit}_{datetime.now().strftime('%Y%m%d')}") + + uploaded = {} + + for file_path in data_dir.rglob('*'): + if file_path.is_file(): + url = self.upload_file(file_path, root_folder) + if url: + uploaded[str(file_path.name)] = url + + print(f"\nšŸ“¤ Uploaded {len(uploaded)} files to Google Drive") + return uploaded + + +def upload_to_s3(subreddit, bucket_name, prefix="r"): + """ + Convenience function to upload subreddit data to S3. + + Args: + subreddit: Subreddit name + bucket_name: S3 bucket name + prefix: "r" or "u" + + Returns: + Upload results + """ + uploader = S3Uploader(bucket_name) + return uploader.upload_subreddit_data(subreddit, prefix) + + +def upload_to_gdrive(subreddit, prefix="r"): + """ + Convenience function to upload subreddit data to Google Drive. + + Args: + subreddit: Subreddit name + prefix: "r" or "u" + + Returns: + Upload results + """ + uploader = GDriveUploader() + return uploader.upload_subreddit_data(subreddit, prefix) + + +# CLI for testing +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Cloud Upload") + parser.add_argument("subreddit", help="Subreddit to upload") + parser.add_argument("--s3-bucket", help="S3 bucket name") + parser.add_argument("--gdrive", action="store_true", help="Upload to Google Drive") + parser.add_argument("--user", action="store_true", help="Is a user profile") + + args = parser.parse_args() + prefix = "u" if args.user else "r" + + if args.s3_bucket: + upload_to_s3(args.subreddit, args.s3_bucket, prefix) + elif args.gdrive: + upload_to_gdrive(args.subreddit, prefix) + else: + print("Please specify --s3-bucket or --gdrive") diff --git a/requirements.txt b/requirements.txt index a713acf..b3ed494 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,6 +2,10 @@ pandas requests +# Async +aiohttp +aiofiles + # Dashboard streamlit diff --git a/scraper/__init__.py b/scraper/__init__.py new file mode 100644 index 0000000..2424006 --- /dev/null +++ b/scraper/__init__.py @@ -0,0 +1,2 @@ +# Scraper module +from .async_scraper import run_async_scraper, scrape_async diff --git a/scraper/async_scraper.py b/scraper/async_scraper.py new file mode 100644 index 0000000..be146f7 --- /dev/null +++ b/scraper/async_scraper.py @@ -0,0 +1,381 @@ +""" +Async Reddit Scraper - 10x Speed Boost with aiohttp +""" +import asyncio +import aiohttp +import aiofiles +import pandas as pd +import datetime +import time +import os +import random +from pathlib import Path +from urllib.parse import urlparse +import sys + +sys.path.insert(0, str(Path(__file__).parent.parent)) +from config import USER_AGENT, MIRRORS, ASYNC_MAX_CONCURRENT, ASYNC_BATCH_SIZE + +# Semaphore to limit concurrent requests +semaphore = None + +async def fetch_json(session, url, retries=3): + """Fetch JSON with retry logic.""" + for attempt in range(retries): + try: + async with session.get(url, timeout=aiohttp.ClientTimeout(total=15)) as response: + if response.status == 200: + return await response.json() + elif response.status == 429: # Rate limited + await asyncio.sleep(5 * (attempt + 1)) + except Exception as e: + if attempt < retries - 1: + await asyncio.sleep(2) + return None + +async def fetch_posts_page(session, base_url, target, after=None, is_user=False): + """Fetch a single page of posts.""" + if is_user: + path = f"/user/{target}/submitted.json" + else: + path = f"/r/{target}/new.json" + + url = f"{base_url}{path}?limit=100&raw_json=1" + if after: + url += f"&after={after}" + + return await fetch_json(session, url) + +async def download_media_async(session, url, save_path): + """Download media file asynchronously.""" + global semaphore + + if os.path.exists(save_path): + return True + + async with semaphore: + try: + async with session.get(url, timeout=aiohttp.ClientTimeout(total=60)) as response: + if response.status == 200: + async with aiofiles.open(save_path, 'wb') as f: + async for chunk in response.content.iter_chunked(8192): + await f.write(chunk) + return True + except: + pass + return False + +async def fetch_comments_async(session, permalink): + """Fetch comments asynchronously.""" + global semaphore + + async with semaphore: + url = f"https://old.reddit.com{permalink}.json?limit=100" + data = await fetch_json(session, url) + + if data and len(data) > 1: + return parse_comments_sync(data[1]['data']['children'], permalink) + return [] + +def parse_comments_sync(comment_list, post_permalink, depth=0, max_depth=3): + """Parse comments (sync helper).""" + comments = [] + + if depth > max_depth: + return comments + + for item in comment_list: + if item['kind'] != 't1': + continue + + c = item['data'] + comments.append({ + "post_permalink": post_permalink, + "comment_id": c.get('id'), + "parent_id": c.get('parent_id'), + "author": c.get('author'), + "body": c.get('body', ''), + "score": c.get('score', 0), + "created_utc": datetime.datetime.fromtimestamp(c.get('created_utc', 0)).isoformat(), + "depth": depth, + "is_submitter": c.get('is_submitter', False), + }) + + replies = c.get('replies') + if replies and isinstance(replies, dict): + comments.extend(parse_comments_sync( + replies.get('data', {}).get('children', []), + post_permalink, depth + 1, max_depth + )) + + return comments + +def extract_media_urls(post_data): + """Extract all media URLs from a post.""" + media = {"images": [], "videos": [], "galleries": []} + + url = post_data.get('url', '') + + if any(ext in url.lower() for ext in ['.jpg', '.jpeg', '.png', '.gif', '.webp']): + media["images"].append(url) + + if 'i.redd.it' in url: + media["images"].append(url) + + if post_data.get('is_video'): + reddit_video = post_data.get('media', {}) + if reddit_video and 'reddit_video' in reddit_video: + video_url = reddit_video['reddit_video'].get('fallback_url', '') + if video_url: + media["videos"].append(video_url.split('?')[0]) + + preview = post_data.get('preview', {}) + if preview and 'images' in preview: + for img in preview['images']: + source = img.get('source', {}) + if source.get('url'): + media["images"].append(source['url'].replace('&', '&')) + + if post_data.get('is_gallery'): + gallery_data = post_data.get('gallery_data', {}) + media_metadata = post_data.get('media_metadata', {}) + + if gallery_data and media_metadata: + for item in gallery_data.get('items', []): + media_id = item.get('media_id') + if media_id and media_id in media_metadata: + meta = media_metadata[media_id] + if meta.get('s', {}).get('u'): + media["galleries"].append(meta['s']['u'].replace('&', '&')) + + return media + +def extract_post_data(p): + """Extract post data from JSON.""" + post_type = "text" + if p.get('is_video'): + post_type = "video" + elif p.get('is_gallery'): + post_type = "gallery" + elif any(ext in p.get('url', '').lower() for ext in ['.jpg', '.jpeg', '.png', '.gif', '.webp']) or 'i.redd.it' in p.get('url', ''): + post_type = "image" + elif p.get('is_self'): + post_type = "text" + else: + post_type = "link" + + return { + "id": p.get('id'), + "title": p.get('title'), + "author": p.get('author'), + "created_utc": datetime.datetime.fromtimestamp(p.get('created_utc', 0)).isoformat(), + "permalink": p.get('permalink'), + "url": p.get('url_overridden_by_dest', p.get('url')), + "score": p.get('score', 0), + "upvote_ratio": p.get('upvote_ratio', 0), + "num_comments": p.get('num_comments', 0), + "num_crossposts": p.get('num_crossposts', 0), + "selftext": p.get('selftext', ''), + "post_type": post_type, + "is_nsfw": p.get('over_18', False), + "is_spoiler": p.get('spoiler', False), + "flair": p.get('link_flair_text', ''), + "total_awards": p.get('total_awards_received', 0), + "has_media": p.get('is_video', False) or p.get('is_gallery', False) or 'i.redd.it' in p.get('url', ''), + "media_downloaded": False, + "source": "Async-Scraper" + } + +async def scrape_async(target, limit=100, is_user=False, download_media=True, scrape_comments=True): + """ + Main async scraping function. + + Args: + target: Subreddit or username + limit: Max posts to scrape + is_user: True if scraping a user + download_media: Download images/videos + scrape_comments: Scrape comments + """ + global semaphore + semaphore = asyncio.Semaphore(ASYNC_MAX_CONCURRENT) + + prefix = "u" if is_user else "r" + print(f"šŸš€ ASYNC Scraper starting for {prefix}/{target}") + print(f" Target: {limit} posts | Media: {download_media} | Comments: {scrape_comments}") + print(f" Concurrency: {ASYNC_MAX_CONCURRENT} simultaneous requests") + print("-" * 50) + + # Setup directories + base_dir = f"data/{prefix}_{target}" + media_dir = f"{base_dir}/media" + images_dir = f"{media_dir}/images" + videos_dir = f"{media_dir}/videos" + + for d in [base_dir, media_dir, images_dir, videos_dir]: + os.makedirs(d, exist_ok=True) + + start_time = time.time() + all_posts = [] + all_comments = [] + media_tasks = [] + seen_permalinks = set() + + # Load existing data + posts_file = f"{base_dir}/posts.csv" + if os.path.exists(posts_file): + try: + df = pd.read_csv(posts_file) + seen_permalinks = set(df['permalink'].astype(str).tolist()) + print(f"šŸ“š Loaded {len(seen_permalinks)} existing posts") + except: + pass + + async with aiohttp.ClientSession(headers={"User-Agent": USER_AGENT}) as session: + after = None + total_fetched = 0 + + while total_fetched < limit: + # Try mirrors + mirrors = MIRRORS.copy() + random.shuffle(mirrors) + + data = None + for mirror in mirrors: + data = await fetch_posts_page(session, mirror, target, after, is_user) + if data: + print(f"āœ… Fetched from {mirror}") + break + + if not data: + print("āŒ All mirrors failed") + break + + children = data.get('data', {}).get('children', []) + if not children: + print("šŸ No more posts") + break + + print(f" Processing {len(children)} posts...") + + # Process posts + batch_posts = [] + comment_tasks = [] + + for child in children: + p = child['data'] + post = extract_post_data(p) + + if post['permalink'] in seen_permalinks: + continue + + seen_permalinks.add(post['permalink']) + batch_posts.append(post) + + # Queue media downloads + if download_media: + media = extract_media_urls(p) + + for i, img_url in enumerate(media['images'][:5]): + ext = os.path.splitext(urlparse(img_url).path)[1] or '.jpg' + save_path = f"{images_dir}/{post['id']}_{i}{ext}" + media_tasks.append(download_media_async(session, img_url, save_path)) + + for i, img_url in enumerate(media['galleries'][:10]): + save_path = f"{images_dir}/{post['id']}_gallery_{i}.jpg" + media_tasks.append(download_media_async(session, img_url, save_path)) + + for i, vid_url in enumerate(media['videos'][:2]): + if 'youtube' not in vid_url: + save_path = f"{videos_dir}/{post['id']}_{i}.mp4" + media_tasks.append(download_media_async(session, vid_url, save_path)) + + # Queue comment fetching + if scrape_comments and post['num_comments'] > 0: + comment_tasks.append(fetch_comments_async(session, post['permalink'])) + + all_posts.extend(batch_posts) + total_fetched += len(batch_posts) + + # Fetch comments in parallel + if comment_tasks: + print(f" šŸ’¬ Fetching comments for {len(comment_tasks)} posts...") + comment_results = await asyncio.gather(*comment_tasks, return_exceptions=True) + for result in comment_results: + if isinstance(result, list): + all_comments.extend(result) + + print(f" šŸ“Š Progress: {total_fetched}/{limit} posts | {len(all_comments)} comments") + + after = data.get('data', {}).get('after') + if not after: + print("šŸ Reached end of available posts") + break + + await asyncio.sleep(1) # Small delay between pages + + # Download all media in parallel + if media_tasks: + print(f"\nšŸ–¼ļø Downloading {len(media_tasks)} media files in parallel...") + media_results = await asyncio.gather(*media_tasks, return_exceptions=True) + downloaded = sum(1 for r in media_results if r is True) + print(f" āœ… Downloaded {downloaded}/{len(media_tasks)} files") + + # Save data + if all_posts: + df = pd.DataFrame(all_posts) + if os.path.exists(posts_file): + df.to_csv(posts_file, mode='a', header=False, index=False) + else: + df.to_csv(posts_file, index=False) + print(f"\nšŸ’¾ Saved {len(all_posts)} posts to {posts_file}") + + if all_comments: + comments_file = f"{base_dir}/comments.csv" + df = pd.DataFrame(all_comments) + if os.path.exists(comments_file): + df.to_csv(comments_file, mode='a', header=False, index=False) + else: + df.to_csv(comments_file, index=False) + print(f"šŸ’¾ Saved {len(all_comments)} comments") + + duration = time.time() - start_time + + print("\n" + "=" * 50) + print("āœ… ASYNC SCRAPE COMPLETE!") + print(f" šŸ“Š Posts: {len(all_posts)}") + print(f" šŸ’¬ Comments: {len(all_comments)}") + print(f" šŸ–¼ļø Media: {len(media_tasks)} queued") + print(f" ā±ļø Duration: {duration:.1f}s") + print(f" āš” Speed: {len(all_posts) / duration:.1f} posts/sec") + + return { + 'posts': len(all_posts), + 'comments': len(all_comments), + 'duration': duration + } + +def run_async_scraper(target, limit=100, is_user=False, download_media=True, scrape_comments=True): + """Wrapper to run async scraper from sync code.""" + return asyncio.run(scrape_async(target, limit, is_user, download_media, scrape_comments)) + +# CLI for testing +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Async Reddit Scraper") + parser.add_argument("target", help="Subreddit or username") + parser.add_argument("--limit", type=int, default=100) + parser.add_argument("--user", action="store_true") + parser.add_argument("--no-media", action="store_true") + parser.add_argument("--no-comments", action="store_true") + + args = parser.parse_args() + + run_async_scraper( + args.target, + args.limit, + args.user, + not args.no_media, + not args.no_comments + )