main.py

"""Reddit Posts + Comments Scraper — customer pain-point research in Python.

Use case: you are validating a product idea (here: invoicing software for
freelancers). The script
  1. searches Reddit for the problem phrases your buyers actually type,
     ranked by top score over the past year,
  2. picks the threads that generated the most discussion,
  3. pulls the full comment tree of each one,
  4. counts the recurring words in those comments and saves everything
     to output.json + output.csv for a spreadsheet or an LLM summary.

Run:
    pip install -r requirements.txt
    cp .env.example .env
    python main.py

Actor: https://apify.com/pro100chok/reddit-scraper-all-in-one?utm_source=github_pro100&utm_medium=readme&utm_campaign=reddit-posts-comments-scraper-python
"""
from __future__ import annotations

import csv
import json
import os
import re
import sys
from collections import Counter
from pathlib import Path

from apify_client import ApifyClient
from apify_client.errors import ApifyApiError
from dotenv import load_dotenv

ACTOR_ID = "pro100chok/reddit-scraper-all-in-one"

# Phrases a frustrated buyer would type, not marketing keywords.
PROBLEM_QUERIES = [
    "freelance invoicing software",
    "clients paying invoices late",
    "best way to send invoices freelancer",
]
THREADS_TO_EXPAND = 3        # how many of the busiest threads to open fully
COMMENTS_PER_THREAD = 200    # comment cap per thread

STOPWORDS = {
    "the", "and", "for", "that", "this", "with", "you", "your", "not", "but",
    "have", "has", "are", "was", "were", "can", "get", "its", "it's", "they",
    "their", "them", "when", "what", "which", "there", "would", "could", "just",
    "like", "from", "about", "will", "then", "than", "been", "also", "some",
    "more", "only", "into", "because", "don't", "i'm", "really", "much", "any",
    "one", "out", "all", "use", "used", "using", "even", "make", "know", "think",
}


def themes(texts: list[str], n: int = 15) -> list[tuple[str, int]]:
    counter: Counter[str] = Counter()
    for text in texts:
        for word in re.findall(r"[a-z][a-z'-]{3,}", (text or "").lower()):
            if word not in STOPWORDS:
                counter[word] += 1
    return counter.most_common(n)


def run_actor(client: ApifyClient, run_input: dict, label: str) -> list[dict]:
    try:
        run = client.actor(ACTOR_ID).call(run_input=run_input, timeout_secs=1200)
    except ApifyApiError as exc:
        sys.exit(f"{label} failed: {exc}")
    return list(client.dataset(run["defaultDatasetId"]).iterate_items())


def main() -> int:
    load_dotenv()
    token = os.environ.get("APIFY_API_TOKEN")
    if not token:
        sys.exit("APIFY_API_TOKEN missing — copy .env.example to .env first.")
    client = ApifyClient(token)
    out = Path(__file__).parent

    # Step 1: find the discussions
    print(f"Step 1/3 — searching Reddit for {len(PROBLEM_QUERIES)} problem phrases")
    posts = run_actor(client, {
        "scrapeType": "search",
        "items": PROBLEM_QUERIES,
        "sort": "top",
        "time": "year",
        "searchPosts": True,
        "includeNSFW": False,
        "maxPostsPerSource": 40,
        "maxItems": 120,
    }, "Search step")
    posts = [p for p in posts if p.get("dataType") == "post"]
    # The same thread can match several queries — keep one copy.
    posts = list({p["id"]: p for p in posts if p.get("id")}.values())
    posts.sort(key=lambda p: -(p.get("numberOfComments") or 0))
    print(f"  {len(posts)} unique posts")

    subreddits = Counter(p.get("community") or "—" for p in posts)
    print("  where the conversation happens:")
    for name, count in subreddits.most_common(6):
        print(f"    {count:>3}  {name}")

    if not posts:
        return 0

    # Step 2: open the busiest threads with their whole comment tree
    busiest = posts[:THREADS_TO_EXPAND]
    print(f"\nStep 2/3 — full comment trees for the {len(busiest)} busiest threads")
    for p in busiest:
        print(f"  {p.get('numberOfComments'):>5} comments  {p.get('community')}  "
              f"{(p.get('title') or '')[:70]}")
    records = run_actor(client, {
        "scrapeType": "post",
        "items": [p["url"] for p in busiest],
        "scrapeComments": True,
        "maxCommentsPerPost": COMMENTS_PER_THREAD,
        "expandMoreComments": True,
        "maxItems": THREADS_TO_EXPAND * (COMMENTS_PER_THREAD + 1),
    }, "Thread step")
    comments = [r for r in records if r.get("dataType") == "comment"]
    print(f"  {len(comments)} comments collected")

    # Step 3: what keeps coming up
    print("\nStep 3/3 — recurring words in the comments")
    for word, count in themes([c.get("body") for c in comments]):
        print(f"    {count:>4}x  {word}")

    top_comments = sorted(comments, key=lambda c: -(c.get("score") or 0))[:5]
    print("\n  Highest-voted comments:")
    for c in top_comments:
        print(f"    [{c.get('score')}] {(c.get('body') or '').replace(chr(10), ' ')[:160]}")

    (out / "output.json").write_text(
        json.dumps({"posts": posts, "comments": comments}, indent=2, ensure_ascii=False),
        encoding="utf-8")
    fields = ["dataType", "community", "postId", "id", "author", "score",
              "numberOfComments", "depth", "createdAt", "title", "body", "url"]
    with (out / "output.csv").open("w", newline="", encoding="utf-8") as fh:
        writer = csv.DictWriter(fh, fieldnames=fields, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(posts + comments)
    print("\nSaved output.json and output.csv")
    return 0


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