reddit_search.py

reddit_search.py

# reddit_search.py
#
# Read-only market-research script. Endpoints used: /api/v1/access_token,
# /r/{subreddit}/search. No write actions, no bulk archiving, no redistribution,
# no AI training. Operated manually ~weekly by /u/matad0r33. <100 requests/day.
#
# Setup: scripts/reddit_auth.json (NOT committed) containing:
#   {"client_id": "...", "client_secret": "..."}
#
# Usage: python reddit_search.py <subreddit> "<query>" [new|relevance] [months_back]
#   e.g.: python reddit_search.py Romania "e-factura" new 6

import sys
import time
import json
import datetime
import requests

UA = "idea-scout/2.0 by /u/matad0r33"
AUTH_FILE = "scripts/reddit_auth.json"

def get_token() -> str:
    auth = json.load(open(AUTH_FILE))
    resp = requests.post(
        "https://www.reddit.com/api/v1/access_token",
        auth=(auth["client_id"], auth["client_secret"]),
        data={"grant_type": "client_credentials"},
        headers={"User-Agent": UA},
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json()["access_token"]

def search(sub: str, query: str, sort: str, months_back: int) -> None:
    cutoff = time.time() - months_back * 30 * 86400
    resp = requests.get(
        f"https://oauth.reddit.com/r/{sub}/search",
        params={
            "q": query,
            "restrict_sr": 1,
            "sort": sort,
            "limit": 25,
            "t": "year",
        },
        headers={"User-Agent": UA, "Authorization": f"bearer {get_token()}"},
        timeout=15,
    )
    resp.raise_for_status()

    results = 0
    for child in resp.json()["data"]["children"]:
        d = child["data"]
        if d["created_utc"] < cutoff:
            continue
        results += 1
        print(json.dumps({
            "title": d["title"],
            "url": f"https://www.reddit.com{d['permalink']}",
            "date": datetime.datetime.utcfromtimestamp(
                d["created_utc"]).strftime("%Y-%m-%d"),
            "score": d["score"],
            "num_comments": d["num_comments"],
            "selftext": d.get("selftext", "")[:600],
        }, ensure_ascii=False))

    if results == 0:
        print(f"NO_RESULTS: r/{sub} q={query!r} (last {months_back} months)",
              file=sys.stderr)

if __name__ == "__main__":
    if len(sys.argv) < 3:
        sys.exit("usage: reddit_search.py <subreddit> <query> [new|relevance] [months_back]")
    sub, query = sys.argv[1], sys.argv[2]
    sort = sys.argv[3] if len(sys.argv) > 3 else "new"
    months = int(sys.argv[4]) if len(sys.argv) > 4 else 6
    search(sub, query, sort, months)