reddit_cli.py

import argparse
import sys
import textwrap
from datetime import datetime, timezone
from shutil import get_terminal_size
from typing import Any, Callable, Sequence, TextIO

from redditAPY import NotFoundError, RateLimitError, Reddit, RedditError

TIME_FILTER_CHOICES = ("hour", "day", "week", "month", "year", "all")
SUBREDDIT_SORT_CHOICES = ("best", "hot", "new", "rising", "top", "controversial")
SEARCH_SORT_CHOICES = ("relevance", "hot", "top", "new", "comments")
USER_SORT_CHOICES = ("hot", "new", "top", "controversial")
DEFAULT_PREVIEW_LENGTH = 280
Handler = Callable[[argparse.Namespace, Reddit, TextIO], int]


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Human-friendly CLI for redditAPY.",
        epilog=(
            "Examples:\n"
            "  python reddit_cli.py subreddit about Python\n"
            "  python reddit_cli.py subreddit posts Python --sort top --time week --limit 10\n"
            "  python reddit_cli.py search posts \"python web scraping\" --subreddit Python\n"
            "  python reddit_cli.py user comments spez --limit 5\n"
            "  python reddit_cli.py post /r/Python/comments/abc123/example_post/"
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--timeout", type=int, default=10, help="Request timeout in seconds.")
    parser.add_argument(
        "--cache-ttl",
        type=int,
        default=3600,
        help="In-memory GET cache TTL in seconds. Use 0 to disable caching.",
    )
    parser.add_argument(
        "--cache-max-entries",
        type=int,
        default=512,
        help="Maximum number of cached GET responses kept in memory.",
    )
    parser.add_argument(
        "--max-retries",
        type=int,
        default=2,
        help="Retries for transient transport, auth, and server errors.",
    )
    parser.add_argument(
        "--backoff-factor",
        type=float,
        default=0.5,
        help="Base delay for retry backoff.",
    )
    parser.add_argument(
        "--no-cache-warning",
        action="store_true",
        help="Suppress the warning about cached GET responses.",
    )

    commands = parser.add_subparsers(dest="command", required=True)

    post_parser = commands.add_parser("post", help="Show a single Reddit post from a permalink or full URL.")
    post_parser.add_argument("permalink", help="Relative permalink or full Reddit URL.")
    post_parser.set_defaults(handler=handle_post)

    subreddit_parser = commands.add_parser("subreddit", help="Commands for subreddit data.")
    subreddit_commands = subreddit_parser.add_subparsers(dest="subreddit_command", required=True)

    subreddit_about = subreddit_commands.add_parser("about", help="Show basic subreddit details.")
    subreddit_about.add_argument("name", help="Subreddit name, for example Python.")
    subreddit_about.set_defaults(handler=handle_subreddit_about)

    subreddit_posts = subreddit_commands.add_parser("posts", help="List posts from a subreddit.")
    subreddit_posts.add_argument("name", help="Subreddit name, for example Python.")
    subreddit_posts.add_argument(
        "--sort",
        choices=SUBREDDIT_SORT_CHOICES,
        default="hot",
        help="Listing sort.",
    )
    subreddit_posts.add_argument(
        "--time",
        choices=TIME_FILTER_CHOICES,
        default="day",
        help="Time filter for top or controversial listings.",
    )
    subreddit_posts.add_argument("--limit", type=int, default=10, help="Number of posts to show.")
    subreddit_posts.add_argument("--after", help="Pagination token for the next page.")
    subreddit_posts.add_argument("--before", help="Pagination token for the previous page.")
    subreddit_posts.add_argument(
        "--exclude-stickied",
        action="store_true",
        help="Hide stickied posts and backfill with more results when possible.",
    )
    subreddit_posts.set_defaults(handler=handle_subreddit_posts)

    search_parser = commands.add_parser("search", help="Search posts or subreddits.")
    search_commands = search_parser.add_subparsers(dest="search_command", required=True)

    search_posts = search_commands.add_parser("posts", help="Search Reddit posts.")
    search_posts.add_argument("query", help="Search query.")
    search_posts.add_argument("--subreddit", help="Restrict search to a single subreddit.")
    search_posts.add_argument(
        "--sort",
        choices=SEARCH_SORT_CHOICES,
        default="relevance",
        help="Search sort.",
    )
    search_posts.add_argument(
        "--time",
        choices=TIME_FILTER_CHOICES,
        default="all",
        help="Time filter.",
    )
    search_posts.add_argument("--limit", type=int, default=10, help="Number of results to show.")
    search_posts.add_argument("--after", help="Pagination token for the next page.")
    search_posts.add_argument("--before", help="Pagination token for the previous page.")
    search_posts.set_defaults(handler=handle_search_posts)

    search_subreddits = search_commands.add_parser("subreddits", help="Search for matching subreddits.")
    search_subreddits.add_argument("query", help="Search query.")
    search_subreddits.add_argument("--limit", type=int, default=10, help="Number of results to show.")
    search_subreddits.set_defaults(handler=handle_search_subreddits)

    user_parser = commands.add_parser("user", help="Commands for Reddit user data.")
    user_commands = user_parser.add_subparsers(dest="user_command", required=True)

    user_about = user_commands.add_parser("about", help="Show basic user details.")
    user_about.add_argument("username", help="Reddit username.")
    user_about.set_defaults(handler=handle_user_about)

    user_overview = user_commands.add_parser("overview", help="Show a mixed overview of a user's activity.")
    user_overview.add_argument("username", help="Reddit username.")
    add_user_listing_flags(user_overview)
    user_overview.set_defaults(handler=handle_user_overview)

    user_posts = user_commands.add_parser(
        "posts",
        aliases=["submitted"],
        help="Show posts submitted by a user.",
    )
    user_posts.add_argument("username", help="Reddit username.")
    add_user_listing_flags(user_posts)
    user_posts.set_defaults(handler=handle_user_posts)

    user_comments = user_commands.add_parser("comments", help="Show comments made by a user.")
    user_comments.add_argument("username", help="Reddit username.")
    add_user_listing_flags(user_comments)
    user_comments.set_defaults(handler=handle_user_comments)

    return parser


def add_user_listing_flags(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--limit", type=int, default=10, help="Number of items to show.")
    parser.add_argument("--after", help="Pagination token for the next page.")
    parser.add_argument("--before", help="Pagination token for the previous page.")
    parser.add_argument(
        "--sort",
        choices=USER_SORT_CHOICES,
        default="new",
        help="Listing sort.",
    )
    parser.add_argument(
        "--time",
        choices=TIME_FILTER_CHOICES,
        default="all",
        help="Time filter for top or controversial listings.",
    )


def emit_cache_warning(args: argparse.Namespace, err: TextIO) -> None:
    cache_enabled = args.cache_ttl > 0 and args.cache_max_entries > 0
    if args.no_cache_warning or not cache_enabled:
        return
    print(
        (
            f"Warning: redditAPY caches GET responses in memory for {args.cache_ttl} seconds "
            "during this process. Use --cache-ttl 0 if you need fresh reads only."
        ),
        file=err,
    )


def create_client(args: argparse.Namespace, reddit_factory: type[Reddit]) -> Reddit:
    return reddit_factory(
        timeout=args.timeout,
        cache_ttl=args.cache_ttl,
        cache_max_entries=args.cache_max_entries,
        max_retries=args.max_retries,
        backoff_factor=args.backoff_factor,
    )


def dispatch(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    handler: Handler = args.handler
    return handler(args, reddit, out)


def handle_subreddit_about(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    payload = reddit.subreddit(args.name).about()
    render_subreddit_about(payload["data"], out)
    return 0


def handle_subreddit_posts(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    methods = reddit.subreddit(args.name)
    options: dict[str, Any] = {
        "limit": positive_limit(args.limit),
        "after": args.after,
        "before": args.before,
        "include_stickied": not args.exclude_stickied,
    }
    if args.sort in {"top", "controversial"}:
        payload = getattr(methods, args.sort)(time_filter=args.time, **options)
    else:
        payload = getattr(methods, args.sort)(**options)
    render_post_listing(
        payload["data"]["children"],
        out,
        heading=f"r/{normalize_name(args.name)} | {args.sort} posts",
    )
    render_pagination_hint(payload["data"], out)
    return 0


def handle_search_posts(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    payload = reddit.search(
        args.query,
        sort=args.sort,
        time_filter=args.time,
        limit=positive_limit(args.limit),
        after=args.after,
        before=args.before,
        subreddit=args.subreddit,
    )
    scope = f" in r/{normalize_name(args.subreddit)}" if args.subreddit else ""
    render_post_listing(
        payload["data"]["children"],
        out,
        heading=f"Search results{scope} for {args.query!r}",
    )
    render_pagination_hint(payload["data"], out)
    return 0


def handle_search_subreddits(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    payload = reddit.subreddits().search(args.query, limit=positive_limit(args.limit))
    render_subreddit_search_results(
        payload["data"]["children"],
        out,
        heading=f"Subreddit results for {args.query!r}",
    )
    render_pagination_hint(payload["data"], out)
    return 0


def handle_user_about(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    payload = reddit.user(args.username).about()
    render_user_about(payload["data"], out)
    return 0


def handle_user_overview(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    payload = reddit.user(args.username).overview(**build_user_listing_kwargs(args))
    render_activity_listing(
        payload["data"]["children"],
        out,
        heading=f"u/{args.username} | overview",
    )
    render_pagination_hint(payload["data"], out)
    return 0


def handle_user_posts(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    payload = reddit.user(args.username).submitted(**build_user_listing_kwargs(args))
    render_post_listing(
        payload["data"]["children"],
        out,
        heading=f"u/{args.username} | submitted posts",
    )
    render_pagination_hint(payload["data"], out)
    return 0


def handle_user_comments(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    payload = reddit.user(args.username).comments(**build_user_listing_kwargs(args))
    render_activity_listing(
        payload["data"]["children"],
        out,
        heading=f"u/{args.username} | comments",
    )
    render_pagination_hint(payload["data"], out)
    return 0


def handle_post(args: argparse.Namespace, reddit: Reddit, out: TextIO) -> int:
    payload = reddit.post(args.permalink)
    render_post_details(payload, out)
    return 0


def build_user_listing_kwargs(args: argparse.Namespace) -> dict[str, Any]:
    return {
        "limit": positive_limit(args.limit),
        "after": args.after,
        "before": args.before,
        "sort": args.sort,
        "time_filter": args.time,
    }


def render_subreddit_about(data: dict[str, Any], out: TextIO) -> None:
    title = data.get("display_name_prefixed") or f"r/{data.get('display_name', 'unknown')}"
    write_heading(out, title)
    write_fields(
        out,
        [
            ("Title", data.get("title")),
            ("Subscribers", format_number(data.get("subscribers"))),
            ("Active users", format_number(data.get("active_user_count"))),
            ("Created", format_timestamp(data.get("created_utc"))),
            ("Over 18", format_bool(data.get("over18"))),
            ("Quarantined", format_bool(data.get("quarantine"))),
            ("URL", f"https://www.reddit.com{data.get('url', '')}" if data.get("url") else None),
        ],
    )
    description = data.get("public_description") or data.get("description")
    if description:
        out.write("\n")
        out.write(wrap_paragraph(description))
        out.write("\n")


def render_user_about(data: dict[str, Any], out: TextIO) -> None:
    write_heading(out, f"u/{data.get('name', 'unknown')}")
    write_fields(
        out,
        [
            ("Link karma", format_number(data.get("link_karma"))),
            ("Comment karma", format_number(data.get("comment_karma"))),
            ("Created", format_timestamp(data.get("created_utc"))),
            ("Verified email", format_bool(data.get("has_verified_email"))),
            ("Is employee", format_bool(data.get("is_employee"))),
            ("Is moderator", format_bool(data.get("is_mod"))),
        ],
    )
    if data.get("subreddit", {}).get("public_description"):
        out.write("\n")
        out.write(wrap_paragraph(data["subreddit"]["public_description"]))
        out.write("\n")


def render_post_listing(children: list[dict[str, Any]], out: TextIO, heading: str) -> None:
    write_heading(out, heading)
    if not children:
        out.write("No posts found.\n")
        return
    for index, child in enumerate(children, start=1):
        data = child.get("data", {})
        out.write(f"{index}. {data.get('title', '<no title>')}\n")
        out.write(
            "   "
            f"r/{data.get('subreddit', 'unknown')} | "
            f"u/{data.get('author', 'unknown')} | "
            f"score {format_number(data.get('score'))} | "
            f"{format_number(data.get('num_comments'))} comments\n"
        )
        out.write(
            "   "
            f"{format_timestamp(data.get('created_utc'))} | "
            f"https://www.reddit.com{data.get('permalink', '')}\n"
        )
        preview = data.get("selftext") or data.get("url_overridden_by_dest")
        if preview:
            out.write(indent_lines(truncate(preview, DEFAULT_PREVIEW_LENGTH), prefix="   ") + "\n")
        out.write("\n")


def render_subreddit_search_results(children: list[dict[str, Any]], out: TextIO, heading: str) -> None:
    write_heading(out, heading)
    if not children:
        out.write("No subreddits found.\n")
        return
    for index, child in enumerate(children, start=1):
        data = child.get("data", {})
        name = data.get("display_name_prefixed") or f"r/{data.get('display_name', 'unknown')}"
        out.write(f"{index}. {name}\n")
        out.write(
            "   "
            f"{format_number(data.get('subscribers'))} subscribers | "
            f"{format_number(data.get('active_user_count'))} active users\n"
        )
        if data.get("public_description"):
            out.write(indent_lines(truncate(data["public_description"], DEFAULT_PREVIEW_LENGTH), prefix="   ") + "\n")
        out.write("\n")


def render_activity_listing(children: list[dict[str, Any]], out: TextIO, heading: str) -> None:
    write_heading(out, heading)
    if not children:
        out.write("No activity found.\n")
        return
    for index, child in enumerate(children, start=1):
        kind = child.get("kind")
        data = child.get("data", {})
        if kind == "t1":
            title = f"Comment in r/{data.get('subreddit', 'unknown')}"
            body = data.get("body", "")
        else:
            title = data.get("title", "<no title>")
            body = data.get("selftext") or data.get("url_overridden_by_dest") or ""
        out.write(f"{index}. {title}\n")
        out.write(
            "   "
            f"u/{data.get('author', 'unknown')} | "
            f"score {format_number(data.get('score'))} | "
            f"{format_timestamp(data.get('created_utc'))}\n"
        )
        if data.get("permalink"):
            out.write(f"   https://www.reddit.com{data['permalink']}\n")
        if body:
            out.write(indent_lines(truncate(body, DEFAULT_PREVIEW_LENGTH), prefix="   ") + "\n")
        out.write("\n")


def render_post_details(payload: list[dict[str, Any]], out: TextIO) -> None:
    if not payload or not payload[0].get("data", {}).get("children"):
        out.write("Post not found.\n")
        return
    data = payload[0]["data"]["children"][0].get("data", {})
    write_heading(out, data.get("title", "<no title>"))
    write_fields(
        out,
        [
            ("Subreddit", f"r/{data.get('subreddit')}" if data.get("subreddit") else None),
            ("Author", f"u/{data.get('author')}" if data.get("author") else None),
            ("Score", format_number(data.get("score"))),
            ("Comments", format_number(data.get("num_comments"))),
            ("Created", format_timestamp(data.get("created_utc"))),
            ("URL", f"https://www.reddit.com{data.get('permalink', '')}" if data.get("permalink") else None),
            ("Outbound", data.get("url_overridden_by_dest")),
        ],
    )
    body = data.get("selftext")
    if body:
        out.write("\n")
        out.write(wrap_paragraph(body))
        out.write("\n")


def render_pagination_hint(data: dict[str, Any], out: TextIO) -> None:
    after = data.get("after")
    before = data.get("before")
    if after is None and before is None:
        return
    out.write("Pagination:\n")
    if after:
        out.write(f"  next page: --after {after}\n")
    if before:
        out.write(f"  previous page: --before {before}\n")


def write_heading(out: TextIO, heading: str) -> None:
    out.write(f"{heading}\n")
    out.write(f"{'=' * len(heading)}\n")


def write_fields(out: TextIO, fields: list[tuple[str, str | None]]) -> None:
    present = [(label, value) for label, value in fields if value not in {None, ""}]
    if not present:
        return
    width = max(len(label) for label, _ in present)
    for label, value in present:
        out.write(f"{label:<{width}} : {value}\n")


def wrap_paragraph(text: str) -> str:
    return textwrap.fill(" ".join(text.split()), width=terminal_width())


def indent_lines(text: str, prefix: str) -> str:
    return textwrap.indent(wrap_paragraph(text), prefix)


def terminal_width() -> int:
    return max(72, min(get_terminal_size((100, 20)).columns, 120))


def truncate(text: str, limit: int) -> str:
    compact = " ".join(text.split())
    if len(compact) <= limit:
        return compact
    return compact[: limit - 1].rstrip() + "…"


def format_timestamp(value: Any) -> str | None:
    if value in (None, ""):
        return None
    try:
        timestamp = float(value)
    except (TypeError, ValueError):
        return str(value)
    return datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")


def format_number(value: Any) -> str | None:
    if value in (None, ""):
        return None
    try:
        if isinstance(value, bool):
            return "1" if value else "0"
        return f"{int(value):,}"
    except (TypeError, ValueError):
        return str(value)


def format_bool(value: Any) -> str | None:
    if value is None:
        return None
    return "yes" if bool(value) else "no"


def normalize_name(value: str | None) -> str:
    if not value:
        return "unknown"
    cleaned = value.strip()
    return cleaned[2:] if cleaned.lower().startswith("r/") else cleaned


def positive_limit(value: int) -> int:
    return max(1, value)


def execute(
    argv: Sequence[str] | None = None,
    *,
    stdout: TextIO | None = None,
    stderr: TextIO | None = None,
    reddit_factory: type[Reddit] = Reddit,
) -> int:
    out = stdout or sys.stdout
    err = stderr or sys.stderr
    parser = build_parser()
    args = parser.parse_args(argv)
    emit_cache_warning(args, err)
    with create_client(args, reddit_factory) as reddit:
        return dispatch(args, reddit, out)


def main(
    argv: Sequence[str] | None = None,
    *,
    stdout: TextIO | None = None,
    stderr: TextIO | None = None,
    reddit_factory: type[Reddit] = Reddit,
) -> int:
    err = stderr or sys.stderr
    try:
        return execute(argv, stdout=stdout, stderr=err, reddit_factory=reddit_factory)
    except NotFoundError as exc:
        print(f"Not found: {exc}", file=err)
        return 2
    except RateLimitError as exc:
        print(f"Rate limited: {exc}", file=err)
        if exc.retry_after is not None:
            print(f"Retry after: {exc.retry_after} seconds", file=err)
        return 3
    except RedditError as exc:
        print(f"Reddit error: {exc}", file=err)
        return 4


if __name__ == "__main__":
    raise SystemExit(main())