post.py

"""Fetch one Reddit post and its comment tree from the Chocodata API.

    export CHOCODATA_API_KEY="your_key"
    python reddit_post_scraper_api_codes/post.py \
        "https://www.reddit.com/r/askscience/comments/627akk/do_giraffes_get_struck_by_lightning_more_often/"
    python reddit_post_scraper_api_codes/post.py 627akk askscience --sort new
"""
from __future__ import annotations

import json
import os
import sys

import requests

ENDPOINT = "https://api.chocodata.com/api/v1/reddit/post"
TIMEOUT = 90


def check(r: requests.Response) -> None:
    """Turn the statuses this endpoint documents into something actionable."""
    if r.status_code == 400:
        sys.exit("400 invalid_params: pass either url, or post_id together with "
                 f"subreddit. Body: {r.text[:200]}")
    if r.status_code == 401:
        sys.exit("401 INVALID_API_KEY: key missing or not recognised. "
                 "Get one at chocodata.com")
    if r.status_code == 402:
        sys.exit("402 INSUFFICIENT_CREDITS: balance exhausted. "
                 "Top up or upgrade at chocodata.com")
    if r.status_code == 404:
        sys.exit("404 item_not_found: no post resolved from that input. "
                 f"Check the URL or the post_id. Body: {r.text[:200]}")
    if r.status_code == 429:
        sys.exit("429 RATE_LIMITED: over your plan's concurrency. Back off and retry.")
    if r.status_code == 502:
        sys.exit("502: Reddit did not serve this post on this request. "
                 "This one is retryable, so try again shortly.")
    r.raise_for_status()


def fetch(url: str | None = None, post_id: str | None = None,
          subreddit: str | None = None, sort: str = "top") -> dict:
    key = os.environ.get("CHOCODATA_API_KEY", "your_key")
    params = {"api_key": key, "sort": sort}
    if url:
        params["url"] = url
    if post_id:
        params["post_id"] = post_id
    if subreddit:
        params["subreddit"] = subreddit
    r = requests.get(ENDPOINT, params=params, timeout=TIMEOUT)
    check(r)
    return r.json()


def walk(comments: list[dict], depth: int = 0):
    """Depth-first walk of the nested comment tree."""
    for c in comments:
        yield depth, c
        yield from walk(c.get("replies", []), depth + 1)


def main(argv: list[str]) -> int:
    args = [a for a in argv[1:] if not a.startswith("--")]
    sort = "top"
    for a in argv[1:]:
        if a.startswith("--sort"):
            sort = a.split("=", 1)[1] if "=" in a else argv[argv.index(a) + 1]
    if not args:
        print(__doc__)
        return 64

    if args[0].startswith("http"):
        data = fetch(url=args[0], sort=sort)
    else:
        if len(args) < 2:
            print("post_id needs a subreddit alongside it: post.py <post_id> <subreddit>")
            return 64
        data = fetch(post_id=args[0], subreddit=args[1], sort=sort)

    post, meta = data["post"], data["_meta"]

    # The post object is a stub when the upstream post page did not render for
    # this request. _meta.source says so: it lists post-page only when the real
    # post tag was found. Check it before trusting a null title.
    if post.get("title") is None:
        print(f"post object is a stub (title is null). _meta.source={meta['source']!r}")
        print(f"num_comments={post['num_comments']}  "
              f"comments_returned={data['comments_returned']}")
        print("A stub with comments means the post page did not render: retry, and "
              "check the subreddit's canonical casing.")
        print("A stub with num_comments 0 and no comments means no post at that id.")
        return 1

    print(post["title"])
    print(f"  r/{post['permalink'].split('/r/')[1].split('/')[0]}  "
          f"score={post['score']}  ratio={post['upvote_ratio']}  "
          f"comments={post['num_comments']}")
    print(f"  posted {post['created']}")
    print(f"  link={post['external_url'] or '(text post)'}  domain={post['domain']}")
    if post.get("body"):
        print(f"  selftext: {post['body'][:200]}")
    print(f"  source={meta['source']}  sort={meta['sort']}  "
          f"truncated={meta['truncated']}")
    print(f"\n{data['comments_returned']} comments returned "
          f"({len(data['comments'])} top level):")
    for depth, c in walk(data["comments"]):
        body = (c["body"] or "").replace("\n", " ")
        print(f"  {'  ' * depth}[{str(c['score']):>6}] "
              f"{c['author']['username']}: {body[:90]}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))