How to Scrape Reddit Posts, Subreddits and Profiles

BASE_CONFIG = {
    # enable the anti scraping protection
    "asp": True,
    # set the proxy country to US
    "country": "US",
    # bypassing reddit requires enabling JavaScript and using the residential proxy pool
    "render_js": True,
    "proxy_pool": "public_residential_pool"
}

import os
from typing import Dict, List, Union
from datetime import datetime
from loguru import logger as log
from scrapfly import ScrapeConfig, ScrapflyClient, ScrapeApiResponse

SCRAPFLY = ScrapflyClient(key=os.environ["SCRAPFLY_KEY"])

$ poetry run pytest test.py

$ poetry run pytest test.py -k test_subreddit_scraping
$ poetry run pytest test.py -k test_post_scraping
$ poetry run pytest test.py -k test_user_post_scraping
$ poetry run pytest test.py -k test_user_comment_scraping

def parse_subreddit(response: ScrapeApiResponse) -> Dict:
    """parse article data from HTML"""
    selector = response.selector
    url = response.context["url"]
    info = {}
    info["id"] = url.split("/r")[-1].replace("/", "")
    info["description"] = selector.xpath("//shreddit-subreddit-header/@description").get()
    members_text = selector.xpath("//faceplate-number[following-sibling::text()[contains(., 'members')]]/@number").get()
    weekly_active = selector.xpath("//shreddit-subreddit-header/@weekly-active-users").get()
    rank = selector.xpath("//strong[@id='position']/text()").get()
    info["rank"] = rank.strip() if rank else None
    info["members"] = int(members_text) if members_text else (int(weekly_active) if weekly_active else None)
    info["bookmarks"] = {}
    for item in selector.xpath("//div[faceplate-tracker[@source='community_menu']]/faceplate-tracker"):
        name = item.xpath(".//a/span/span/span/text()").get()
        link = item.xpath(".//a/@href").get()
        if name and link:
            info["bookmarks"][name] = link
    info["url"] = url

    post_data = []
    for box in selector.xpath("//article[@data-post-id]"):
        link = box.xpath(".//a/@href").get()
        author = box.xpath(".//shreddit-post/@author").get()
        post_label = box.xpath(".//span[contains(@class, 'bg-tone-4')]/div/text()").get()
        upvotes = box.xpath(".//shreddit-post/@score").get()
        comment_count = box.xpath(".//shreddit-post/@comment-count").get()
        attachment_type = box.xpath(".//shreddit-post/@post-type").get()
        # simplified here, the repository file also covers gallery posts and a content-href fallback
        attachment_link = None
        if attachment_type == "image":
            attachment_link = box.xpath(".//img[contains(@class, 'media-lightbox-img')]/@src").get()
        elif attachment_type == "video":
            attachment_link = box.xpath(".//shreddit-player/@preview").get()
        post_data.append({
            "authorProfile": "https://www.reddit.com/user/" + author if author else None,
            "authorId": box.xpath(".//shreddit-post/@author-id").get(),
            "title": box.xpath(".//shreddit-post/@post-title").get(),
            "link": "https://www.reddit.com" + link if link and link.startswith("/") else link,
            "publishingDate": box.xpath(".//shreddit-post/@created-timestamp").get(),
            "postId": box.xpath(".//shreddit-post/@id").get(),
            "postLabel": post_label.strip() if post_label else None,
            "postUpvotes": int(upvotes) if upvotes else None,
            "commentCount": int(comment_count) if comment_count else None,
            "attachmentType": attachment_type,
            "attachmentLink": attachment_link,
        })
    cursor_id = selector.xpath("//shreddit-post/@more-posts-cursor").get()
    return {"post_data": post_data, "info": info, "cursor": cursor_id}


async def scrape_subreddit(subreddit_id: str, max_pages: int = None) -> Dict:
    """scrape a subreddit's first page (pagination covered next)"""
    base_url = f"https://www.reddit.com/r/{subreddit_id}/"
    response = await SCRAPFLY.async_scrape(ScrapeConfig(base_url, **BASE_CONFIG))
    data = parse_subreddit(response)
    subreddit_data = {"info": data["info"], "posts": data["post_data"]}
    log.success(f"scraped {len(subreddit_data['posts'])} posts from r/{subreddit_id}")
    return subreddit_data

def make_pagination_url(cursor_id: str, subreddit_id: str):
    return f"https://www.reddit.com/svc/shreddit/community-more-posts/hot/?after={cursor_id}%3D%3D&t=DAY&name={subreddit_id}&feedLength=3"

def parse_post_info(response: ScrapeApiResponse) -> Dict:
    """parse post data from a subreddit post"""
    selector = response.selector
    info = {}
    label = selector.xpath("//faceplate-tracker[@source='post']/a/span/div/text()").get()
    comments = selector.xpath("//shreddit-post/@comment-count").get()
    upvotes = selector.xpath("//shreddit-post/@score").get()
    info["authorId"] = selector.xpath("//shreddit-post/@author-id").get()
    info["author"] = selector.xpath("//shreddit-post/@author").get()
    info["authorProfile"] = "https://www.reddit.com/user/" + info["author"] if info["author"] else None
    subreddit = selector.xpath("//shreddit-post/@subreddit-prefixed-name").get()
    info["subreddit"] = subreddit.replace("r/", "") if subreddit else None
    info["postId"] = selector.xpath("//shreddit-post/@id").get()
    info["postLabel"] = label.strip() if label else None
    info["publishingDate"] = selector.xpath("//shreddit-post/@created-timestamp").get()
    info["postTitle"] = selector.xpath("//shreddit-post/@post-title").get()
    info["postLink"] = selector.xpath("//shreddit-canonical-url-updater/@value").get()
    info["commentCount"] = int(comments) if comments else None
    info["upvoteCount"] = int(upvotes) if upvotes else None
    info["attachmentType"] = selector.xpath("//shreddit-post/@post-type").get()
    info["attachmentLink"] = selector.xpath("//shreddit-post/@content-href").get()
    return info

def parse_comment(parent_selector) -> Dict:
    """parse a comment object"""
    author = parent_selector.xpath("./@data-author").get()
    link = parent_selector.xpath("./@data-permalink").get()
    dislikes = parent_selector.xpath(".//span[contains(@class, 'dislikes')]/@title").get()
    upvotes = parent_selector.xpath(".//span[contains(@class, 'likes')]/@title").get()
    downvotes = parent_selector.xpath(".//span[contains(@class, 'unvoted')]/@title").get()
    return {
        "authorId": parent_selector.xpath("./@data-author-fullname").get(),
        "author": author,
        "authorProfile": "https://www.reddit.com/user/" + author if author else None,
        "commentId": parent_selector.xpath("./@data-fullname").get(),
        "link": "https://www.reddit.com" + link if link else None,
        "publishingDate": parent_selector.xpath(".//time/@datetime").get(),
        "commentBody": parent_selector.xpath(".//div[@class='md']/p/text()").get(),
        "upvotes": int(upvotes) if upvotes else None,
        "dislikes": int(dislikes) if dislikes else None,
        "downvotes": int(downvotes) if downvotes else None,
    }


def parse_replies(what) -> List[Dict]:
    """recursively parse replies"""
    replies = []
    for reply_box in what.xpath(".//div[@data-type='comment']"):
        reply_comment = parse_comment(reply_box)
        child_replies = parse_replies(reply_box)
        if child_replies:
            reply_comment["replies"] = child_replies
        replies.append(reply_comment)
    return replies


def parse_post_comments(response: ScrapeApiResponse) -> List[Dict]:
    """parse post comments"""
    selector = response.selector
    data = []
    for item in selector.xpath("//div[@class='sitetable nestedlisting']/div[@data-type='comment']"):
        comment_data = parse_comment(item)
        replies = parse_replies(item)
        if replies:
            comment_data["replies"] = replies
        data.append(comment_data)
    return data


async def scrape_post(url: str, sort: Union["old", "new", "top"]) -> Dict:
    """scrape a post's info and comments"""
    response = await SCRAPFLY.async_scrape(ScrapeConfig(url, **BASE_CONFIG))
    post_data = {"info": parse_post_info(response)}
    post_link = post_data["info"]["postLink"] or post_data["info"]["attachmentLink"]
    comments_url = post_link.replace("www", "old") + f"?sort={sort}&limit=500"
    response = await SCRAPFLY.async_scrape(ScrapeConfig(comments_url, **BASE_CONFIG))
    post_data["comments"] = parse_post_comments(response)
    log.success(f"scraped {len(post_data['comments'])} comments from {url}")
    return post_data

def parse_user_posts(response: ScrapeApiResponse) -> Dict:
    """parse user posts from user profiles"""
    selector = response.selector
    data = []
    for box in selector.xpath("//div[@id='siteTable']/div[contains(@class, 'thing')]"):
        author = box.xpath("./@data-author").get()
        link = box.xpath("./@data-permalink").get()
        timestamp = box.xpath("./@data-timestamp").get()
        publishing_date = (
            datetime.fromtimestamp(int(timestamp) / 1000.0).strftime("%Y-%m-%dT%H:%M:%S.%f%z")
            if timestamp else None
        )
        comment_count = box.xpath("./@data-comments-count").get()
        post_score = box.xpath("./@data-score").get()
        data.append({
            "authorId": box.xpath("./@data-author-fullname").get(),
            "author": author,
            "authorProfile": "https://www.reddit.com/user/" + author if author else None,
            "postId": box.xpath("./@data-fullname").get(),
            "postLink": "https://www.reddit.com" + link if link else None,
            "postTitle": box.xpath(".//p[@class='title']/a/text()").get(),
            "postSubreddit": box.xpath("./@data-subreddit-prefixed").get(),
            "publishingDate": publishing_date,
            "commentCount": int(comment_count) if comment_count else None,
            "postScore": int(post_score) if post_score else None,
            "attachmentType": box.xpath("./@data-type").get(),
            "attachmentLink": box.xpath("./@data-url").get(),
        })
    next_page_url = selector.xpath("//span[@class='next-button']/a/@href").get()
    return {"data": data, "url": next_page_url}


async def scrape_user_posts(username: str, sort: Union["new", "top", "controversial"], max_pages: int = None) -> List[Dict]:
    """scrape user posts"""
    url = f"https://old.reddit.com/user/{username}/submitted/?sort={sort}"
    response = await SCRAPFLY.async_scrape(ScrapeConfig(url, **BASE_CONFIG))
    data = parse_user_posts(response)
    post_data, next_page_url = data["data"], data["url"]
    while next_page_url and (max_pages is None or max_pages > 0):
        response = await SCRAPFLY.async_scrape(ScrapeConfig(next_page_url, **BASE_CONFIG))
        data = parse_user_posts(response)
        next_page_url = data["url"]
        post_data.extend(data["data"])
        if max_pages is not None:
            max_pages -= 1
    log.success(f"scraped {len(post_data)} posts from the {username} reddit profile")
    return post_data

def parse_user_comments(response: ScrapeApiResponse) -> Dict:
    """parse user comments from user profiles"""
    selector = response.selector
    data = []
    for box in selector.xpath("//div[@id='siteTable']/div[contains(@class, 'thing')]"):
        author = box.xpath("./@data-author").get()
        link = box.xpath("./@data-permalink").get()
        dislikes = box.xpath(".//span[contains(@class, 'dislikes')]/@title").get()
        upvotes = box.xpath(".//span[contains(@class, 'likes')]/@title").get()
        downvotes = box.xpath(".//span[contains(@class, 'unvoted')]/@title").get()
        data.append({
            "authorId": box.xpath("./@data-author-fullname").get(),
            "author": author,
            "authorProfile": "https://www.reddit.com/user/" + author if author else None,
            "commentId": box.xpath("./@data-fullname").get(),
            "commentLink": "https://www.reddit.com" + link if link else None,
            "commentBody": "".join(box.xpath(".//div[contains(@class, 'usertext-body')]/div/p/text()").getall()).replace("\n", ""),
            "attachedCommentLinks": box.xpath(".//div[contains(@class, 'usertext-body')]/div/p/a/@href").getall(),
            "publishingDate": box.xpath(".//time/@datetime").get(),
            "dislikes": int(dislikes) if dislikes else None,
            "upvotes": int(upvotes) if upvotes else None,
            "downvotes": int(downvotes) if downvotes else None,
            "replyTo": {
                "postTitle": box.xpath(".//p[@class='parent']/a[@class='title']/text()").get(),
                "postLink": "https://www.reddit.com" + box.xpath(".//p[@class='parent']/a[@class='title']/@href").get(),
                "postAuthor": box.xpath(".//p[@class='parent']/a[contains(@class, 'author')]/text()").get(),
                "postSubreddit": box.xpath("./@data-subreddit-prefixed").get(),
            }
        })
    next_page_url = selector.xpath("//span[@class='next-button']/a/@href").get()
    return {"data": data, "url": next_page_url}


async def scrape_user_comments(username: str, sort: Union["new", "top", "controversial"], max_pages: int = None) -> List[Dict]:
    """scrape user comments"""
    url = f"https://old.reddit.com/user/{username}/comments/?sort={sort}"
    response = await SCRAPFLY.async_scrape(ScrapeConfig(url, **BASE_CONFIG))
    data = parse_user_comments(response)
    post_data, next_page_url = data["data"], data["url"]
    while next_page_url and (max_pages is None or max_pages > 0):
        response = await SCRAPFLY.async_scrape(ScrapeConfig(next_page_url, **BASE_CONFIG))
        data = parse_user_comments(response)
        next_page_url = data["url"]
        post_data.extend(data["data"])
        if max_pages is not None:
            max_pages -= 1
    log.success(f"scraped {len(post_data)} comments from the {username} reddit profile")
    return post_data

from cerberus import Validator

def check_output(items: list, schema: dict, min_items: int, label: str) -> bool:
    """conceptual adaptation of the canonical Cerberus checks in test.py"""
    validator = Validator(schema, allow_unknown=True)
    for item in items:
        if not validator.validate(item):
            print(f"[{label}] schema failure: {validator.errors}")
            return False
    if len(items) < min_items:
        print(f"[{label}] only {len(items)} items, expected at least {min_items}")
        return False
    print(f"[{label}] passed: {len(items)} items validated")
    return True

# one check per page type, so a single failing route doesn't mask the other three
results = {
    "subreddit": check_output(subreddit_posts, subreddit_post_schema, min_items=50, label="subreddit"),
    "post_comments": check_output(post_comments, comment_schema, min_items=50, label="post"),
    "profile_posts": check_output(profile_posts, user_post_schema, min_items=25, label="profile_posts"),
    "profile_comments": check_output(profile_comments, user_comment_schema, min_items=2, label="profile_comments"),
}