DEV Community

Qovia
Qovia

Posted on

Monitor Reddit for mentions of your product in Python

I build Qovia, the API used in this post. The monitoring logic (search, de-duplicate, filter, alert) works with any Reddit data source; the state file, CSV and Slack parts are plain Python.

If you ship a developer tool or a SaaS product, Reddit is where people ask "is there an alternative to X?", complain about your onboarding, or recommend you to someone else, often without tagging you. Finding those threads early is worth a lot: a helpful answer in the first hour of a thread does more than a week of ads. Doing it by hand means refreshing Reddit search every morning, so let's automate it.

In this tutorial we'll write a script that:

  1. searches Reddit for your product name (and competitors) with Reddit's own search syntax,
  2. watches the newest posts in a few subreddits for phrases like "alternative to" or "looking for",
  3. remembers what it has already seen, so each run only reports new posts,
  4. pulls the top comments under every new hit,
  5. writes everything to CSV and posts a short summary to a Slack channel.

Run it from cron every hour and you have a basic social-listening tool for a few dollars a month.

Setup

You need Python 3.9+ and requests:

pip install requests
export QOVIA_KEY=sk-qovia-...   # key from the Qovia console
export SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...   # optional
Enter fullscreen mode Exit fullscreen mode

Get a key from Qovia. Requests authenticate with Authorization: Bearer <key> (an x-api-key header also works). The Reddit search reference shows every parameter with a real response; the other Reddit endpoints are listed next to it.

You don't need a Reddit account, OAuth app or client secret for this. The trade-off is that it's read-only and public content only (more under Limits).

Step 1: a small request helper

Every response has the same envelope: the result is in data, and meta reports the charge (meta.billed, meta.charged_usd, meta.empty). Empty results come back as HTTP 200 with meta.empty: true and are not charged. Errors are {"error": {"code", "message", ...}} and are never charged. A post or subreddit that doesn't exist answers 404, so the helper turns that into None. A 429 carries a Retry-After header, so the helper waits and retries once:

import os
import time
import requests

BASE = "https://api.qovia.ai"
HEADERS = {"Authorization": f"Bearer {os.environ['QOVIA_KEY']}"}

spent = 0.0


def get(path, **params):
    global spent
    params = {k: v for k, v in params.items() if v is not None}
    for attempt in range(2):
        r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
        if r.status_code == 429 and attempt == 0:
            time.sleep(int(r.headers.get("Retry-After", "10")))
            continue
        if r.status_code == 404:
            return None  # missing post or subreddit, not charged
        r.raise_for_status()
        body = r.json()
        spent += float((body.get("meta") or {}).get("charged_usd") or 0)
        return body.get("data") or {}
Enter fullscreen mode Exit fullscreen mode

Unknown parameters are rejected with 400 invalid_params, which is why the helper drops None values instead of sending them.

Step 2: search Reddit for your keywords

GET /v1/raw/reddit/search takes a keyword (1 to 200 characters) and Reddit's search syntax works inside it: AND, OR, NOT, quoted phrases, subreddit:, author:, title: and so on. Two more parameters matter for monitoring:

  • sort: relevance (the default), hot, top, new or comments. For monitoring you want new.
  • time_range: all (the default), year, month, week, day or hour. If you run hourly, day gives you some overlap, which the state file in Step 4 takes care of.

Results come about 7 a page. Pass cursor back for the next page and stop when has_more is false. If Reddit spell-corrected your query, corrected_query tells you what it actually searched, which is worth logging: brand names get "corrected" more often than you'd think.

Each post record has text (title and body), author (username), metrics (likes, which is the post's score, and comments) and extra with source_id (the post's fullname, t3_...), title, subreddit, published_at (UTC), flair, content_type, link_url and url.

QUERIES = {
    "brand": '"acme analytics" OR acmeanalytics',
    "competitor": '"plausible analytics" alternative',
}


def search(query, time_range="day", max_pages=2):
    posts, cursor = [], None
    for _ in range(max_pages):
        page = get("/v1/raw/reddit/search", keyword=query, sort="new",
                   time_range=time_range, cursor=cursor)
        if not page:
            break
        if page.get("corrected_query"):
            print(f"  note: Reddit searched {page['corrected_query']!r} instead of {query!r}")
        posts.extend(page.get("items", []))
        if not page.get("has_more"):
            break
        cursor = page.get("cursor")
    return posts
Enter fullscreen mode Exit fullscreen mode

Quote multi-word names. Without quotes, acme analytics matches any post containing both words anywhere, which for a common word gets noisy fast.

Step 3: watch a few subreddits for intent phrases

Search finds posts that name you. The more valuable posts often don't: "what do you use for privacy-friendly analytics?" in r/webdev never mentions any product. For those, watch the newest posts in the communities your users hang out in and filter locally.

GET /v1/raw/reddit/subreddit_posts takes a subreddit (the name after r/) and a sort (hot by default; also best, new, top, controversial, rising). It's cursor-paginated: about 6 posts on the first page, then about 24. The list records carry the title and the first 300 characters of the body (extra.truncated tells you when it was cut), which is plenty for a phrase filter. Fields the list doesn't carry (language, author ID, images, upvote ratio) are null; GET /v1/raw/reddit/post has them if you need them.

import re

SUBREDDITS = ["webdev", "SaaS", "selfhosted"]
TOPIC = re.compile(r"analytics|tracking|pageviews", re.I)
INTENT = re.compile(r"alternative to|looking for|recommend|what do you use|switch(ed|ing)? from", re.I)


def intent_posts(subreddit, max_pages=2):
    hits, cursor = [], None
    for _ in range(max_pages):
        page = get("/v1/raw/reddit/subreddit_posts", subreddit=subreddit,
                   sort="new", cursor=cursor)
        if not page:
            break  # 404: no such subreddit
        for p in page.get("items", []):
            text = p.get("text") or ""
            if TOPIC.search(text) and INTENT.search(text):
                hits.append(p)
        if not page.get("has_more"):
            break
        cursor = page.get("cursor")
    return hits
Enter fullscreen mode Exit fullscreen mode

The two regexes are deliberately simple: a post must be about the topic and sound like someone is choosing a tool. Tune them on a week of output; this is the part of the script you'll edit most.

Step 4: remember what you've already seen

Both sources return the same record shape with the same IDs, so one set of seen fullnames is enough. A JSON file next to the script is fine for an hourly job:

import json
from pathlib import Path

STATE = Path("reddit_seen.json")


def load_seen():
    return set(json.loads(STATE.read_text())) if STATE.exists() else set()


def save_seen(seen):
    STATE.write_text(json.dumps(sorted(seen)))


def only_new(posts, seen):
    fresh = []
    for p in posts:
        pid = p["extra"]["source_id"]
        if pid not in seen:
            seen.add(pid)
            fresh.append(p)
    return fresh
Enter fullscreen mode Exit fullscreen mode

This also de-duplicates a post that shows up in both a search and a subreddit feed during the same run.

Step 5: read the room under each new post

A title tells you a thread exists; the comments tell you whether it's worth answering. GET /v1/raw/reddit/comments takes the post's post_id (the same t3_... fullname) and an optional sort (best by default, or top, new, old, controversial). A page holds 8 comments: a few top-level ones plus the replies Reddit expands under them, flattened in Reddit's order. Each has cid, text, likes (the comment's score), reply_count, created_at (UTC), pinned, author, depth (0 for top-level) and parent_cid. The page also has total, cursor and has_more. A post without comments answers empty and isn't charged.

def top_comments(post_id, keywords=None):
    page = get("/v1/raw/reddit/comments", post_id=post_id, sort="top") or {}
    out = []
    for c in page.get("items", []):
        text = (c.get("text") or "").replace("\n", " ")
        out.append({
            "cid": c["cid"],
            "depth": c.get("depth"),
            "score": c.get("likes"),
            "replies": c.get("reply_count"),
            "mentions": bool(keywords and keywords.search(text)),
            "text": text[:300],
        })
    return out, page.get("total")
Enter fullscreen mode Exit fullscreen mode

One page per post keeps the cost predictable. If you need a full thread, keep paging with cursor; replies Reddit collapses come with a replies_cursor that GET /v1/raw/reddit/comment_replies expands (it takes post_id and that cursor). For alerts, the top 8 is enough to see whether people are already recommending someone.

Step 6: CSV, Slack, and the run loop

import csv
from datetime import datetime, timezone

FIELDS = ["checked_at", "source", "post_id", "subreddit", "title", "author",
          "score", "comments", "published_at", "url", "top_comment"]
MENTION = re.compile(r"acme", re.I)  # flags comments that name you


def now():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def row(source, p, comments):
    ex, m = p.get("extra") or {}, p.get("metrics") or {}
    best = comments[0]["text"] if comments else ""
    return {
        "checked_at": now(),
        "source": source,
        "post_id": ex.get("source_id"),
        "subreddit": ex.get("subreddit"),
        "title": ex.get("title"),
        "author": (p.get("author") or {}).get("username"),
        "score": m.get("likes"),
        "comments": m.get("comments"),
        "published_at": ex.get("published_at"),
        "url": ex.get("url"),
        "top_comment": best[:200],
    }


def append_csv(path, rows):
    path = Path(path)
    new_file = not path.exists()
    with path.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=FIELDS)
        if new_file:
            w.writeheader()
        w.writerows(rows)


def notify(rows):
    url = os.environ.get("SLACK_WEBHOOK_URL")
    if not url or not rows:
        return
    lines = [f"*{len(rows)} new Reddit posts*"]
    for r in rows[:15]:
        lines.append(f"• [{r['source']}] r/{r['subreddit']}: <{r['url']}|{r['title']}> "
                     f"({r['score']} pts, {r['comments']} comments)")
    requests.post(url, json={"text": "\n".join(lines)}, timeout=30)


def run():
    seen = load_seen()
    found = []
    for name, q in QUERIES.items():
        found += [(name, p) for p in only_new(search(q), seen)]
    for sub in SUBREDDITS:
        found += [(f"r/{sub}", p) for p in only_new(intent_posts(sub), seen)]

    rows = []
    for source, p in found:
        comments, _ = top_comments(p["extra"]["source_id"], MENTION)
        if any(c["mentions"] for c in comments):
            source += " +mentioned in comments"
        rows.append(row(source, p, comments))

    append_csv("reddit_mentions.csv", rows)
    notify(rows)
    save_seen(seen)
    print(f"{len(rows)} new posts, spent this run: ${spent:.4f}")


if __name__ == "__main__":
    run()
Enter fullscreen mode Exit fullscreen mode

The state is saved only after the CSV and the Slack message go out, so a crash mid-run means the next run reports those posts again instead of losing them.

Run it once by hand, then schedule it:

python3 reddit_monitor.py
# crontab -e
# .env holds: export QOVIA_KEY=... and export SLACK_WEBHOOK_URL=...
0 * * * * cd /path/to/monitor && . ./.env && python3 reddit_monitor.py >> monitor.log 2>&1
Enter fullscreen mode Exit fullscreen mode

Output looks something like this (illustrative, not real data):

  note: Reddit searched 'acme analytic' instead of 'acmeanalytics'
4 new posts, spent this run: $0.0140
Enter fullscreen mode Exit fullscreen mode

And in Slack (also illustrative):

4 new Reddit posts
• [brand] r/webdev: Anyone tried Acme Analytics for a static site? (12 pts, 9 comments)
• [competitor] r/selfhosted: Plausible alternative that runs on SQLite? (31 pts, 22 comments)
• [r/SaaS] r/SaaS: Looking for privacy-friendly analytics for a B2B app (5 pts, 3 comments)
• [r/webdev +mentioned in comments] r/webdev: What do you use for pageview tracking in 2026? (48 pts, 61 comments)
Enter fullscreen mode Exit fullscreen mode

The last one is the interesting kind: a post that never names you, where someone in the comments already did.

Doing the same from Claude or Cursor

For one-off research ("what did r/selfhosted say about analytics tools this month?") a script is overkill. The same endpoints are available through Qovia's MCP server at https://api.qovia.ai/mcp (Streamable HTTP, same API key in an Authorization: Bearer header, same billing as the HTTP API). In Claude Code:

claude mcp add --transport http qovia https://api.qovia.ai/mcp \
  --header "Authorization: Bearer $QOVIA_KEY"
Enter fullscreen mode Exit fullscreen mode

Then ask something like "search Reddit for posts from the last week asking for a Plausible alternative, and summarize what the top comments recommend." Configs for Cursor, VS Code and Claude Desktop are in the MCP docs. The assistant is good for digging into one question; the cron job is better for not missing anything.

Cost

Every Reddit endpoint used here (search, subreddit_posts, comments, and post / comment_replies if you add them) is a fixed $0.001 per successful call ($1 per 1,000). Empty results, errors and 404s are not charged, so a quiet hour where search finds nothing costs nothing for that query.

As a rough example (arithmetic only): with 2 queries x up to 2 search pages, 3 subreddits x up to 2 pages, and one comment page per new post, a run is at most 10 calls plus one per new post. With 4 new posts that's $0.014. Run hourly, the worst case is 240 calls a day ($0.24) plus one call per new post, so roughly $7 to $8 a month; it's less in practice because a search with no new results returns empty and isn't charged. The subreddit feeds are the fixed part: one page per subreddit, or running every three hours, cuts that by half or more. The spent counter prints the real number each run. New accounts get 50 trial calls, counted only when a call returns data; after that it's a prepaid balance with no monthly fee. Details are on the pricing page, and every parameter is in the API docs.

The default rate limit is 60 requests per minute per account, far more than an hourly job needs.

Limits

  • Public content only, read-only: you can't post, vote or reply through this.
  • The Reddit endpoints are marked beta.
  • Search covers posts, not comments. A comment that names you under a post that doesn't is only found if that post is in one of your watched subreddits (Step 3 + Step 5).
  • subreddit_posts carries only the first 300 characters of the body; fetch GET /v1/raw/reddit/post when you need the full text.
  • Scores and comment counts are a snapshot at the time of the call; a post found an hour after it went up will have very different numbers a day later.
  • Reddit's own search decides what matches. Check corrected_query and test your queries on reddit.com once before trusting them.

If you build on this, tell me in the comments what you'd want it to flag besides mentions.


Drafted with AI assistance and checked against the live API docs.

Top comments (0)