Merge pull request #1571 from AnishSarkar22/feat/reddit-scraping

feat(scrapers): add anonymous Reddit scraper

surfsense_backend/app/proprietary/scrapers/reddit/README.md

@@ -0,0 +1,85 @@
+# Reddit scraper (anonymous, no browser)
+
+Platform-native Reddit scraper (anonymous, no browser). Standalone
+module: it depends only on `app.utils.proxy` + `scrapling` and exposes a stable
+public API. It is **not** wired into routes, `connector_service.py`, ingestion,
+or Celery — integration later is a thin `reddit_routes.py` + one `include_router`
+line, identical to the `youtube` / `google_maps` siblings.
+
+## Approach
+
+Reddit deprecated **cold** unauthenticated `.json` (r/modnews, May 2026): a bare
+anonymous GET to `/…/.json` now 403s. What still works — and what maintained
+anonymous scrapers (e.g. yt-dlp) now do — is:
+
+> Bootstrap an anonymous session cookie (`loid`) with one plain HTTP GET to
+> `www.reddit.com/svc/shreddit/<slug>`, then GET
+> `www.reddit.com/<path>/.json?raw_json=1` through that same
+> Chrome-impersonated, sticky-IP session. Rotate the residential IP + re-warm on
+> block.
+
+`loid` is Reddit's equivalent of the Google Maps scraper's `NID`: an anonymous,
+logged-out id that unlocks the public API. **No browser, no Chromium, no
+`solve_cloudflare`** — this collapses Reddit onto the cheap HTTP tier the
+siblings already use. Confirmed live 2026-07-04 through a residential proxy (see
+`scripts/e2e_reddit_scraper.py` step 0): `svc/shreddit` warm-up → 200 + `loid`
+(with `old.reddit` as a fallback), then sequential `.json` fetches → 200.
+
+## Anonymous only — no authentication, ever
+
+No OAuth, no login, no `reddit_session` account cookie, no Devvit. The only
+cookie held is the anonymous `loid`. There is **no** authenticate option in the
+input surface or the fetch layer, by design. A persistent block after IP
+rotation surfaces as `RedditAccessBlockedError` (mirrors google_maps'
+`SignInRequiredError`) rather than a silent empty result.
+
+## Module map
+
+| File | Responsibility |
+|---|---|
+| `__init__.py` | Public exports: `RedditScrapeInput`, `RedditItem`, `iter_reddit`, `scrape_reddit`, `RedditAccessBlockedError`. |
+| `schemas.py` | `RedditScrapeInput` (`extra="allow"`, no auth fields) + single flat `RedditItem` keyed by `dataType` + `StartUrl`. |
+| `fetch.py` | The core. Rotate-on-block sticky `_RotatingSession` + `_current_session` ContextVar + `warm_session` (loid) + `fetch_json`. No browser imports. |
+| `url_resolver.py` | Classify a Reddit URL → `post`/`subreddit`/`user`/`search`; non-Reddit → `None`. |
+| `parsers.py` | Pure JSON→item mapping (`parse_post`, `parse_comment`, `parse_community`, `flatten_comments`, `children`/`after`). I/O-free. |
+| `scraper.py` | Orchestrator: `_post_flow`/`_subreddit_flow`/`_user_flow`/`_search_flow`, `_paginate_listing`, `_dispatch`, `fan_out`, `iter_reddit`, `scrape_reddit`. |
+
+## How it works
+
+1. `iter_reddit` resolves `startUrls` (or builds a search per `searches` entry)
+   and fans them out on a pool of warm proxy sessions (`fan_out`, 16-way). Each
+   worker opens one sticky-IP session and warms `loid` once, reusing it across
+   the sequential targets it pulls.
+2. Each flow pages its listing via the `after` cursor (`_paginate_listing`),
+   filtering by child `kind`, the NSFW gate, and `postDateLimit`.
+3. `fetch_json` warms `loid` on first use, rotates the IP + re-warms on 403,
+   backs off on 429, returns `None` on 404, and paces each sticky IP to ~1 req/s
+   to stay under Reddit's per-IP rate limit.
+4. Parsers map raw `.json` things to flat `RedditItem`s; the orchestrator stamps
+   `scrapedAt` and applies caps as request-time policy.
+
+## Testing
+
+- Offline unit tests: `tests/unit/scrapers/reddit/` — `test_skeleton.py`
+  (schema + URL resolver), `test_parsers.py` (fixture-pinned mapping),
+  `test_fetch_resilience.py` (warm / rotate / backoff loop with fake sessions,
+  no network).
+- Live e2e (needs network + residential proxy): `scripts/e2e_reddit_scraper.py`
+  — step 0 is the go/no-go `loid` probe; later steps exercise the flows and dump
+  trimmed fixtures into `tests/unit/scrapers/reddit/fixtures/`.
+
+```bash
+cd surfsense_backend
+.venv/bin/python -m pytest tests/unit/scrapers/reddit/
+.venv/bin/python scripts/e2e_reddit_scraper.py   # live; regenerates fixtures
+```
+
+## TODO / out of scope (v1)
+
+- Sticky-IP provider support: the fetch layer assumes a sticky exit IP per
+  session (the `loid` binds to it). `anonymous_proxies` does not yet set the
+  `"si"` sticky key — add it (proxy layer) before high-volume production use.
+- `/api/morechildren` deep-comment expansion — `more` stubs terminate the tree
+  walk today.
+- Routes / `connector_service.py` / ingestion / Celery wiring.
+- RSS degraded-mode path (documented, not implemented).

surfsense_backend/app/proprietary/scrapers/reddit/__init__.py

@@ -0,0 +1,13 @@
+"""Platform-native Reddit scraper (anonymous, no browser)."""
+
+from .fetch import RedditAccessBlockedError
+from .schemas import RedditItem, RedditScrapeInput
+from .scraper import iter_reddit, scrape_reddit
+
+__all__ = [
+    "RedditAccessBlockedError",
+    "RedditItem",
+    "RedditScrapeInput",
+    "iter_reddit",
+    "scrape_reddit",
+]

surfsense_backend/app/proprietary/scrapers/reddit/fetch.py

@@ -0,0 +1,343 @@
+"""Proxy-aware fetch seam for the Reddit scraper (no browser).
+
+All network I/O flows through :func:`fetch_json` and always egresses through the
+residential proxy (a direct hit would expose and risk-block the server IP).
+
+Reddit deprecated *cold* unauthenticated ``.json`` (a bare anonymous GET now
+403s). The maintained anonymous recipe (proven live 2026-07-04, see
+``scripts/e2e_reddit_scraper.py`` step 0) is:
+
+    warm one anonymous session cookie (``loid``) with a plain GET to
+    ``old.reddit.com`` (``www.reddit.com/svc/shreddit/<slug>`` fallback), then
+    GET ``www.reddit.com/<path>/.json?raw_json=1`` through that same
+    Chrome-impersonated, sticky-IP session. Which warm URL mints ``loid`` is
+    exit-IP dependent, so the order is a tiebreak — see :func:`warm_session`.
+
+``loid`` is Reddit's equivalent of Google Maps' ``NID`` session cookie: an
+anonymous, logged-out id that unlocks the public API — no account, no browser.
+This module is a direct port of ``../youtube/innertube.py``'s rotate-on-block
+sticky-session pattern (``_RotatingSession`` + ``_current_session`` ContextVar +
+``open_proxy_holder``/``bind_proxy_holder``/``proxy_session``), with a
+Reddit-specific :func:`warm_session` bolted on and a ``.json``-shaped
+:func:`fetch_json` instead of an InnerTube POST.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import random
+import time
+from contextlib import asynccontextmanager, suppress
+from contextvars import ContextVar
+from datetime import UTC, datetime
+from typing import Any
+from urllib.parse import urlencode
+
+from scrapling.fetchers import AsyncFetcher, FetcherSession
+
+from app.utils.proxy import get_proxy_url
+
+logger = logging.getLogger(__name__)
+
+
+class RedditAccessBlockedError(RuntimeError):
+    """Raised when every rotated IP is refused anonymous access.
+
+    This is the yt-dlp "account authentication is required" / "your IP is unable
+    to access the Reddit API" branch. We are anonymous-only and cannot log in,
+    so instead of silently returning nothing we surface it as a clear error
+    (mirrors google_maps' ``SignInRequiredError``). The route can turn it into a
+    403 later; for now it just fails loudly.
+    """
+
+
+# Per-flow proxy session, set by ``bind_proxy_holder`` around one continuation
+# chain. Reusing one keep-alive connection pins a single sticky exit IP so the
+# ``loid`` cookie (bound to that IP) stays valid across the warm-up + every
+# subsequent ``.json`` fetch. A ContextVar keeps each concurrent fan-out flow on
+# its own session/IP without threading a param through every call.
+_current_session: ContextVar[_RotatingSession | None] = ContextVar(
+    "reddit_proxy_session", default=None
+)
+
+# 403 => this IP is blocked; rotate to a fresh one and re-warm. 429 => rate
+# limited; back off on the SAME IP (rotating wouldn't help and burns the pool).
+# Split out from youtube's combined ``_BLOCK_STATUSES`` because Reddit wants
+# different handling per status (spec section 3).
+_ROTATE_STATUS = 403
+_BACKOFF_STATUS = 429
+_MAX_ROTATIONS = 3
+_MAX_BACKOFFS = 4
+_BACKOFF_BASE_S = 5.0
+
+# Reddit 429s aggressively (~60-100 req/min/IP). Pace each sticky session to
+# ~1 req/s + jitter to stay under the per-IP threshold.
+_MIN_INTERVAL_S = 1.0
+_PACE_JITTER_S = 0.5
+
+_HEADERS = {"Accept-Language": "en-US,en;q=0.9"}
+
+# Age-gate opt-in, sent on every ``.json`` fetch so NSFW listings aren't blanked
+# (the caller filters on ``includeNSFW`` downstream). Mirrors the probe.
+_OVER18_COOKIES = {"over18": "1"}
+
+# ``svc/shreddit`` needs *a* path to render; any always-public subreddit mints
+# ``loid`` just the same. ``r/popular`` always exists, so it's a safe default
+# regardless of which target this session ends up serving.
+_WARM_SLUG = "r/popular"
+_SHREDDIT_URL = (
+    "https://www.reddit.com/svc/shreddit/{slug}"
+    "?render-mode=partial&seeker-session=false"
+)
+_OLD_REDDIT_URL = "https://old.reddit.com/"
+_LOID_COOKIE = "loid"
+
+
+def now_iso() -> str:
+    """UTC timestamp in the millisecond ISO shape used by scraper output."""
+    return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+
+def _response_cookie_names(page: Any) -> set[str]:
+    """Cookie names set by a response (best-effort across scrapling shapes)."""
+    cookies = getattr(page, "cookies", None)
+    if isinstance(cookies, dict):
+        return set(cookies.keys())
+    return set()
+
+
+def _parse_json(page: Any) -> Any | None:
+    """Parse a scrapling response body into JSON, or ``None``.
+
+    Prefers ``page.json()``; falls back to ``json.loads`` on the raw body when
+    the impersonated response hands back text. No ``<pre>`` unwrapping — that was
+    a v1 browser artifact and does not apply to plain-HTTP ``.json``.
+    """
+    fn = getattr(page, "json", None)
+    if callable(fn):
+        with suppress(Exception):
+            return fn()
+    for attr in ("body", "text"):
+        val = getattr(page, attr, None)
+        if isinstance(val, bytes):
+            val = val.decode("utf-8", "replace")
+        if isinstance(val, str) and val.strip():
+            with suppress(Exception):
+                return json.loads(val)
+            return None
+    return None
+
+
+def _build_url(path: str, params: dict[str, Any] | None) -> str:
+    """``https://www.reddit.com/<path>/.json?raw_json=1&...`` (always raw_json)."""
+    clean = path.strip("/")
+    query = {"raw_json": "1", **(params or {})}
+    qs = urlencode({k: v for k, v in query.items() if v is not None})
+    return f"https://www.reddit.com/{clean}/.json?{qs}"
+
+
+class _RotatingSession:
+    """Owns one live ``FetcherSession`` (sticky IP) and can swap it for a fresh one.
+
+    ``rotate()`` closes the current keep-alive connection and opens a new one, so
+    the rotating gateway hands out a different residential exit IP. Because the
+    ``loid`` cookie binds to the exit IP, ``rotate()`` also drops the warmed
+    state — the next fetch re-warms on the new IP. Used sequentially within a
+    single flow (never shared across concurrent tasks), so no locking is needed.
+    ``session`` is ``None`` only when no proxy is configured.
+    """
+
+    def __init__(self) -> None:
+        self._cm: Any | None = None
+        self.session: Any | None = None
+        self.rotations = 0
+        self.warmed = False
+        self._last_at = 0.0
+
+    async def _open(self) -> None:
+        proxy = get_proxy_url()
+        self.warmed = False
+        if proxy is None:
+            self._cm = self.session = None
+            return
+        self._cm = FetcherSession(
+            proxy=proxy, stealthy_headers=True, impersonate="chrome"
+        )
+        self.session = await self._cm.__aenter__()
+
+    async def close(self) -> None:
+        if self._cm is not None:
+            with suppress(Exception):  # best-effort teardown
+                await self._cm.__aexit__(None, None, None)
+        self._cm = self.session = None
+
+    async def rotate(self) -> Any | None:
+        """Drop the current IP and connect through a fresh one. Returns new session."""
+        await self.close()
+        self.rotations += 1
+        await self._open()
+        logger.info("[reddit] rotated proxy session (rotation #%d)", self.rotations)
+        return self.session
+
+    async def pace(self) -> None:
+        """Sleep to hold this sticky IP under Reddit's per-IP rate threshold."""
+        wait = _MIN_INTERVAL_S - (time.monotonic() - self._last_at)
+        if wait > 0:
+            await asyncio.sleep(wait + random.uniform(0, _PACE_JITTER_S))
+        self._last_at = time.monotonic()
+
+
+async def open_proxy_holder() -> _RotatingSession:
+    """Open a warm rotate-on-block session holder (caller owns ``close()``)."""
+    holder = _RotatingSession()
+    await holder._open()
+    return holder
+
+
+@asynccontextmanager
+async def bind_proxy_holder(holder: _RotatingSession):
+    """Route this task's fetches through ``holder`` for the enclosed block.
+
+    Does NOT close the holder — enables pooling warm sessions across sequential
+    jobs so each job skips the ~2s proxy handshake AND the ``loid`` warm-up.
+    """
+    token = _current_session.set(holder)
+    try:
+        yield holder
+    finally:
+        _current_session.reset(token)
+
+
+@asynccontextmanager
+async def proxy_session():
+    """Open one reused, rotate-on-block proxy session for a continuation chain."""
+    holder = await open_proxy_holder()
+    try:
+        async with bind_proxy_holder(holder):
+            yield holder
+    finally:
+        await holder.close()
+
+
+async def warm_session(session: Any, *, slug: str = _WARM_SLUG) -> bool:
+    """Mint an anonymous ``loid`` cookie on a freshly opened session.
+
+    Returns ``True`` when a ``loid`` was issued (the session can now reach
+    ``.json``), else ``False`` (caller rotates the IP and retries).
+
+    Tries ``old.reddit`` (yt-dlp's primary), then ``svc/shreddit``. WHICH one
+    mints is exit-IP dependent and roughly random: live probes 2026-07-04 saw
+    both directions across sessions on the rotating residential/custom proxy
+    (one IP 403s ``old.reddit`` but mints on ``shreddit``; another does the
+    reverse, sometimes with an ``rdt`` bot-interstitial). So the order is a
+    tiebreak, not an optimization — a fresh session pays ~one wasted warm 403
+    either way. That cost is amortized: ``fan_out`` reuses one warmed session
+    per worker across many jobs, so warm-up runs once per worker, not per fetch.
+    The fallback is what actually matters — it preserves correctness whichever
+    way a given IP leans.
+
+    ponytail: sequential two-source warm burns 1 wasted request on ~half of new
+    sessions. A parallel warm (gather both, take whichever mints) removes the
+    latency but always spends 2 requests; not worth it while warm-up is
+    once-per-worker. Revisit only if session churn (not reuse) dominates.
+
+    Takes an already-open ``session`` (never constructs one) so tests can drive
+    warm/rotate deterministically with a fake session, exactly like the youtube
+    sibling's fetch-resilience tests.
+    """
+    seen: set[str] = set()
+    with suppress(Exception):
+        page = await session.get(_OLD_REDDIT_URL, headers=_HEADERS)
+        seen |= _response_cookie_names(page)
+    if _LOID_COOKIE in seen:
+        return True
+
+    # Fallback: mints loid on exit IPs where old.reddit 403s instead.
+    with suppress(Exception):
+        page = await session.get(_SHREDDIT_URL.format(slug=slug), headers=_HEADERS)
+        seen |= _response_cookie_names(page)
+    return _LOID_COOKIE in seen
+
+
+async def _get_page(session: Any, url: str) -> Any:
+    """GET through the warmed sticky session, or a one-shot proxied fetch."""
+    if session is not None:
+        return await session.get(url, headers=_HEADERS, cookies=_OVER18_COOKIES)
+    return await AsyncFetcher.get(
+        url,
+        headers=_HEADERS,
+        cookies=_OVER18_COOKIES,
+        proxy=get_proxy_url(),
+        stealthy_headers=True,
+    )
+
+
+async def fetch_json(path: str, params: dict[str, Any] | None = None) -> Any | None:
+    """GET a Reddit ``.json`` endpoint through a ``loid``-warmed HTTP session.
+
+    Returns parsed JSON (dict or list), or ``None`` on 404 / non-block failure.
+    Warms the ``loid`` session once per session; rotates the residential IP and
+    re-warms on 403; backs off on 429. Raises :class:`RedditAccessBlockedError`
+    only when every rotated IP refuses anonymous access (the yt-dlp
+    "login required" branch, which we cannot satisfy).
+    """
+    holder = _current_session.get()
+    if holder is None:
+        # No bound session (e.g. a direct call outside fan_out): open a
+        # short-lived warmed session for this one fetch, then tear it down.
+        async with proxy_session():
+            return await fetch_json(path, params)
+
+    url = _build_url(path, params)
+    attempt = 0
+    backoffs = 0
+    while True:
+        session = holder.session
+        try:
+            if session is not None and not holder.warmed:
+                warmed_ok = await warm_session(session)
+                holder.warmed = True  # attempted; don't re-warm this IP
+                if not warmed_ok:
+                    if attempt < _MAX_ROTATIONS:
+                        attempt += 1
+                        await holder.rotate()
+                        continue
+                    raise RedditAccessBlockedError(
+                        f"could not mint loid after {attempt} IP rotations for {path}"
+                    )
+
+            await holder.pace()
+            page = await _get_page(session, url)
+            status = page.status
+
+            if status == 200:
+                return _parse_json(page)
+            if status == 404:
+                return None
+            if status == _BACKOFF_STATUS and backoffs < _MAX_BACKOFFS:
+                backoffs += 1
+                delay = _BACKOFF_BASE_S * (2 ** (backoffs - 1))
+                logger.warning("[reddit] 429 on %s; backing off %.1fs", path, delay)
+                await asyncio.sleep(delay + random.uniform(0, 1))
+                continue
+            if status == _ROTATE_STATUS and attempt < _MAX_ROTATIONS:
+                attempt += 1
+                await holder.rotate()
+                continue
+            if status == _ROTATE_STATUS:
+                raise RedditAccessBlockedError(
+                    f"Reddit refused {path} on {attempt} rotated IPs (403)"
+                )
+            logger.warning("[reddit] GET %s returned %s", path, status)
+            return None
+        except RedditAccessBlockedError:
+            raise
+        except Exception as e:
+            logger.warning("[reddit] GET %s failed: %s", path, e)
+            if attempt < _MAX_ROTATIONS:
+                attempt += 1
+                await holder.rotate()
+                continue
+            return None

surfsense_backend/app/proprietary/scrapers/reddit/parsers.py

@@ -0,0 +1,250 @@
+"""Pure JSON -> item mapping for the Reddit scraper.
+
+Framework-agnostic and I/O-free so it can be unit-tested against captured
+fixtures. Every function takes raw Reddit ``.json`` data and returns plain
+dicts / lists — no network, no proxy, no ``scrapedAt`` stamp (the orchestrator
+adds the timestamp so these stay deterministic under fixture tests).
+
+Reddit's ``.json`` wraps everything in "things" (``{"kind": "t3", "data":
+{...}}``) and "Listings" (``{"kind": "Listing", "data": {"children": [...],
+"after": ...}}``). ``t3`` = post, ``t1`` = comment, ``more`` = a truncated-reply
+stub (out of scope v1 — treated as a branch stop).
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+_REDDIT_BASE = "https://www.reddit.com"
+
+
+def _int(value: Any) -> int | None:
+    """Coerce to int, or ``None`` (Reddit sometimes sends floats/None)."""
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float):
+        return int(value)
+    return None
+
+
+def _utc_from_sec(value: Any) -> str | None:
+    """Epoch seconds -> millisecond ISO string, or ``None``."""
+    if not isinstance(value, int | float):
+        return None
+    dt = datetime.fromtimestamp(float(value), tz=UTC)
+    return dt.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+
+def _parse_iso(value: str | None) -> datetime | None:
+    """Parse an ISO date (tolerating a trailing ``Z``) to aware UTC, else None."""
+    if not value:
+        return None
+    try:
+        dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return dt if dt.tzinfo else dt.replace(tzinfo=UTC)
+
+
+def _before(created_utc: Any, date_limit: str | None) -> bool:
+    """True if ``created_utc`` (epoch secs) predates the ISO ``date_limit``."""
+    limit = _parse_iso(date_limit)
+    if limit is None or not isinstance(created_utc, int | float):
+        return False
+    return datetime.fromtimestamp(float(created_utc), tz=UTC) < limit
+
+
+def _thumbnail_url(value: Any) -> str | None:
+    """Reddit uses sentinels like 'self'/'default'/'nsfw' for no thumbnail."""
+    return value if isinstance(value, str) and value.startswith("http") else None
+
+
+def _permalink_url(data: dict[str, Any]) -> str | None:
+    permalink = data.get("permalink")
+    return f"{_REDDIT_BASE}{permalink}" if isinstance(permalink, str) else None
+
+
+def _strip_prefix(fullname: Any) -> str | None:
+    """``t3_abc`` -> ``abc`` (Reddit fullname without its type prefix)."""
+    if isinstance(fullname, str) and "_" in fullname:
+        return fullname.split("_", 1)[1]
+    return fullname if isinstance(fullname, str) else None
+
+
+def _image_urls(data: dict[str, Any]) -> list[str]:
+    """Best-effort image URLs from a post's ``preview.images[].source.url``."""
+    preview = data.get("preview")
+    if not isinstance(preview, dict):
+        return []
+    urls: list[str] = []
+    for image in preview.get("images") or []:
+        source = image.get("source") if isinstance(image, dict) else None
+        url = source.get("url") if isinstance(source, dict) else None
+        if isinstance(url, str) and url:
+            urls.append(url)
+    return urls
+
+
+def _video_urls(data: dict[str, Any]) -> list[str]:
+    """Best-effort video URL from ``media.reddit_video.fallback_url``."""
+    for key in ("media", "secure_media"):
+        media = data.get(key)
+        if isinstance(media, dict):
+            rv = media.get("reddit_video")
+            url = rv.get("fallback_url") if isinstance(rv, dict) else None
+            if isinstance(url, str) and url:
+                return [url]
+    return []
+
+
+def children(listing: Any) -> list[dict[str, Any]]:
+    """Return a Listing's ``data.children`` array (empty list if malformed)."""
+    if isinstance(listing, dict):
+        data = listing.get("data")
+        if isinstance(data, dict):
+            kids = data.get("children")
+            if isinstance(kids, list):
+                return kids
+    return []
+
+
+def after(listing: Any) -> str | None:
+    """Return a Listing's ``data.after`` pagination cursor, or ``None``."""
+    if isinstance(listing, dict):
+        data = listing.get("data")
+        if isinstance(data, dict):
+            cursor = data.get("after")
+            return cursor if isinstance(cursor, str) else None
+    return None
+
+
+def _unwrap(thing: dict[str, Any]) -> dict[str, Any]:
+    """Accept a ``{"kind","data"}`` thing or a bare data dict; return the data."""
+    data = thing.get("data")
+    return data if isinstance(data, dict) else thing
+
+
+def parse_post(thing: dict[str, Any]) -> dict[str, Any]:
+    """Map a ``t3`` post thing (or its data dict) to a flat item dict."""
+    data = _unwrap(thing)
+    is_self = bool(data.get("is_self"))
+    external = data.get("url") if not is_self else None
+    return {
+        "dataType": "post",
+        "id": data.get("name"),
+        "parsedId": data.get("id"),
+        "url": _permalink_url(data),
+        "username": data.get("author"),
+        "userId": data.get("author_fullname"),
+        "title": data.get("title"),
+        "body": data.get("selftext") or None,
+        "html": data.get("selftext_html"),
+        "link": external,
+        "externalUrl": external,
+        "communityName": data.get("subreddit_name_prefixed"),
+        "parsedCommunityName": data.get("subreddit"),
+        "numberOfComments": _int(data.get("num_comments")),
+        "upVotes": _int(data.get("score", data.get("ups"))),
+        "upVoteRatio": data.get("upvote_ratio"),
+        "over18": data.get("over_18"),
+        "isVideo": data.get("is_video"),
+        "flair": data.get("link_flair_text"),
+        "authorFlair": data.get("author_flair_text"),
+        "thumbnailUrl": _thumbnail_url(data.get("thumbnail")),
+        "imageUrls": _image_urls(data),
+        "videoUrls": _video_urls(data),
+        "numberOfMembers": _int(data.get("subreddit_subscribers")),
+        "createdAt": _utc_from_sec(data.get("created_utc")),
+    }
+
+
+def parse_community(thing: dict[str, Any]) -> dict[str, Any]:
+    """Map a ``t5`` subreddit thing (``about.json``) to a flat community item."""
+    data = _unwrap(thing)
+    url = data.get("url")
+    return {
+        "dataType": "community",
+        "id": data.get("name"),
+        "parsedId": data.get("id"),
+        "url": f"{_REDDIT_BASE}{url}" if isinstance(url, str) else None,
+        "title": data.get("title"),
+        "body": data.get("public_description") or None,
+        "communityName": data.get("display_name_prefixed"),
+        "parsedCommunityName": data.get("display_name"),
+        "numberOfMembers": _int(data.get("subscribers")),
+        "over18": data.get("over18"),
+        "createdAt": _utc_from_sec(data.get("created_utc")),
+    }
+
+
+def parse_comment(thing: dict[str, Any], *, depth: int = 0) -> dict[str, Any]:
+    """Map a ``t1`` comment thing (or its data dict) to a flat item dict.
+
+    ``numberOfReplies`` is left at ``0`` here; :func:`flatten_comments` fills it
+    with the count of descendant comments it actually emits.
+    """
+    data = _unwrap(thing)
+    return {
+        "dataType": "comment",
+        "id": data.get("name"),
+        "parsedId": data.get("id"),
+        "url": _permalink_url(data),
+        "username": data.get("author"),
+        "userId": data.get("author_fullname"),
+        "body": data.get("body") or None,
+        "html": data.get("body_html"),
+        "communityName": data.get("subreddit_name_prefixed"),
+        "parsedCommunityName": data.get("subreddit"),
+        "upVotes": _int(data.get("score", data.get("ups"))),
+        "over18": data.get("over_18"),
+        "authorFlair": data.get("author_flair_text"),
+        "postId": _strip_prefix(data.get("link_id")),
+        "parentId": data.get("parent_id"),
+        "numberOfReplies": 0,
+        "createdAt": _utc_from_sec(data.get("created_utc")),
+        "depth": depth,
+    }
+
+
+def flatten_comments(
+    comment_children: list[dict[str, Any]] | None,
+    *,
+    max_comments: int,
+    date_limit: str | None = None,
+    depth: int = 0,
+    out: list[dict[str, Any]] | None = None,
+) -> list[dict[str, Any]]:
+    """Depth-first flatten a comment tree into a bounded flat list.
+
+    Skips non-``t1`` children (``more`` stubs terminate that branch — v1),
+    honors ``max_comments`` (total emitted, across all depths) and
+    ``date_limit`` (drops comments older than the ISO cutoff). Each comment's
+    ``numberOfReplies`` is set to the number of its descendants that were
+    emitted.
+    """
+    if out is None:
+        out = []
+    for child in comment_children or []:
+        if len(out) >= max_comments:
+            break
+        if not isinstance(child, dict) or child.get("kind") != "t1":
+            continue  # 'more' stub / non-comment: stop this branch (v1)
+        data = child.get("data") or {}
+        if date_limit and _before(data.get("created_utc"), date_limit):
+            continue
+        item = parse_comment(data, depth=depth)
+        before = len(out)
+        out.append(item)
+        replies = data.get("replies")
+        flatten_comments(
+            children(replies) if isinstance(replies, dict) else [],
+            max_comments=max_comments,
+            date_limit=date_limit,
+            depth=depth + 1,
+            out=out,
+        )
+        item["numberOfReplies"] = len(out) - before - 1
+    return out

surfsense_backend/app/proprietary/scrapers/reddit/schemas.py

@@ -0,0 +1,126 @@
+# ruff: noqa: N815 - field names intentionally use the public camelCase API
+"""Input/output models for the Reddit scraper.
+
+The MVP populates a reliably-sourced subset of fields; every other output field
+is emitted as ``None``/``[]`` so the contract can expand additively.
+
+**Anonymous only.** There is deliberately **no** authentication field on the
+input (no username/password/token/``login*``) — the scraper holds only Reddit's
+anonymous ``loid`` session cookie and can never log in. Anything auth-shaped a
+caller sends lands in ``extra`` and is ignored.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+RedditSort = Literal["relevance", "hot", "top", "new", "rising", "comments"]
+RedditTime = Literal["all", "hour", "day", "week", "month", "year"]
+RedditDataType = Literal["post", "comment", "community", "user"]
+
+
+class StartUrl(BaseModel):
+    """A single direct URL entry (``{"url": ...}``; extra keys ignored)."""
+
+    model_config = ConfigDict(extra="allow")
+
+    url: str
+
+
+class RedditScrapeInput(BaseModel):
+    """Reddit scraper input surface (anonymous, no auth fields).
+
+    Caps (``maxItems``/``maxPostCount``/...) are collector policy applied by
+    :func:`scrape_reddit`, NOT ceilings baked into the streaming flows. Fields
+    the MVP doesn't act on are still accepted via ``extra="allow"`` for parity.
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    # Discovery
+    startUrls: list[StartUrl] = Field(default_factory=list)
+    searches: list[str] = Field(default_factory=list)
+    searchCommunityName: str | None = None
+
+    # Sort / filter
+    sort: RedditSort = "new"
+    time: RedditTime | None = None
+    includeNSFW: bool = True
+
+    # Skips
+    skipComments: bool = False
+    skipUserPosts: bool = False
+    skipCommunity: bool = False
+
+    # Caps (collector policy; enforced by scrape_reddit, not the flows)
+    maxItems: int = Field(default=10, ge=0)
+    maxPostCount: int = Field(default=10, ge=0)
+    maxComments: int = Field(default=10, ge=0)
+    maxCommunitiesCount: int = Field(default=2, ge=0)
+    maxUserCount: int = Field(default=2, ge=0)
+
+    # Incremental scraping (ISO dates)
+    postDateLimit: str | None = None
+    commentDateLimit: str | None = None
+
+
+class RedditItem(BaseModel):
+    """Single flat output item, keyed by ``dataType``.
+
+    One model for post/comment/community/user (union of fields, unsourced
+    default ``None``/``[]``), using a single flat-dataset shape.
+    ``extra="allow"`` keeps the contract open so added fields never break
+    consumers.
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    dataType: RedditDataType | None = None
+
+    # Identity / provenance
+    id: str | None = None
+    parsedId: str | None = None
+    url: str | None = None
+    username: str | None = None
+    userId: str | None = None
+
+    # Content
+    title: str | None = None
+    body: str | None = None
+    html: str | None = None
+    link: str | None = None
+    externalUrl: str | None = None
+
+    # Community
+    communityName: str | None = None
+    parsedCommunityName: str | None = None
+    numberOfMembers: int | None = None
+
+    # Engagement
+    numberOfComments: int | None = None
+    numberOfReplies: int | None = None
+    upVotes: int | None = None
+    upVoteRatio: float | None = None
+
+    # Flags / media
+    over18: bool | None = None
+    isVideo: bool | None = None
+    flair: str | None = None
+    authorFlair: str | None = None
+    thumbnailUrl: str | None = None
+    imageUrls: list[str] = Field(default_factory=list)
+    videoUrls: list[str] = Field(default_factory=list)
+
+    # Threading
+    postId: str | None = None
+    parentId: str | None = None
+
+    # Timestamps
+    createdAt: str | None = None
+    scrapedAt: str | None = None
+
+    def to_output(self) -> dict[str, Any]:
+        """Serialize to the flat output dict shape (keeps extras)."""
+        return self.model_dump(exclude_none=False)

surfsense_backend/app/proprietary/scrapers/reddit/scraper.py

@@ -0,0 +1,411 @@
+"""Orchestrator for the Reddit scraper.
+
+The core is the async generator :func:`iter_reddit` (unbounded, ``after``-cursor
+paged); :func:`scrape_reddit` is a thin collector with a caller-supplied
+``limit`` guard. Any cap is caller policy, never baked into flow logic.
+
+Independent targets (one per ``startUrl`` / search) fan out concurrently on a
+pool of warm ``loid`` sessions (sticky IPs); each target's own ``after`` paging
+stays sequential. ``fan_out`` is ported from ``../youtube/scraper.py`` but bound
+to *this* module's proxy holders so every worker warms its own ``loid`` once and
+reuses it — the ~10-50x throughput win over a browser design.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import AsyncIterator
+from typing import Any
+
+from .fetch import (
+    RedditAccessBlockedError,
+    bind_proxy_holder,
+    fetch_json,
+    now_iso,
+    open_proxy_holder,
+)
+from .parsers import (
+    _before,
+    after,
+    children,
+    flatten_comments,
+    parse_comment,
+    parse_community,
+    parse_post,
+)
+from .schemas import RedditItem, RedditScrapeInput
+from .url_resolver import ResolvedUrl, resolve_url
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "RedditAccessBlockedError",
+    "iter_reddit",
+    "scrape_reddit",
+]
+
+# Independent jobs run concurrently on a pool of warm proxy sessions. Matches
+# the youtube sibling; 16 workers saturate typical job counts while leaving
+# gateway headroom.
+_FANOUT_CONCURRENCY = 16
+
+# Reddit caps any listing at ~1000 items (100/page => ~10 pages). Stop there so
+# a runaway target can't page forever.
+_LISTING_LIMIT = 100
+_MAX_PAGES = 10
+_EMPTY_STREAK_ABORT = 2
+
+# Search sorts differ from listing sorts; fall back to "new" for a listing path
+# when the input carries a search-only sort.
+_LISTING_SORTS = frozenset({"hot", "new", "top", "rising", "controversial", "best"})
+
+
+async def fan_out(
+    jobs: list[AsyncIterator[dict[str, Any]]], *, concurrency: int = _FANOUT_CONCURRENCY
+) -> AsyncIterator[dict[str, Any]]:
+    """Stream items from independent async-iterator jobs via a warm worker pool.
+
+    Each worker opens ONE proxy session and reuses it across the sequential jobs
+    it pulls, so only the first job per worker pays the proxy handshake + the
+    ``loid`` warm-up. A bad job yields nothing rather than aborting the batch;
+    workers are cancelled and their sessions closed if the consumer stops early.
+    """
+    if not jobs:
+        return
+    job_queue: asyncio.Queue[AsyncIterator[dict[str, Any]]] = asyncio.Queue()
+    for job in jobs:
+        job_queue.put_nowait(job)
+    results: asyncio.Queue[list[dict[str, Any]]] = asyncio.Queue()
+
+    async def worker() -> None:
+        holder = None
+        try:
+            holder = await open_proxy_holder()
+        except Exception as e:  # no session: jobs still run via one-shot fetches
+            logger.warning("[reddit] proxy session open failed: %s", e)
+        try:
+            while True:
+                try:
+                    job = job_queue.get_nowait()
+                except asyncio.QueueEmpty:
+                    return
+                items: list[dict[str, Any]] = []
+                try:
+                    if holder is not None:
+                        async with bind_proxy_holder(holder):
+                            items = [item async for item in job]
+                    else:
+                        items = [item async for item in job]
+                except Exception as e:  # one bad target must not kill the run
+                    logger.warning("[reddit] fan-out job failed: %s", e)
+                await results.put(items)
+        finally:
+            if holder is not None:
+                await holder.close()
+
+    tasks = [asyncio.create_task(worker()) for _ in range(min(concurrency, len(jobs)))]
+    try:
+        for _ in range(len(jobs)):
+            for item in await results.get():
+                yield item
+    finally:
+        for task in tasks:
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+
+def _emit(partial: dict[str, Any], *, include_nsfw: bool) -> dict[str, Any] | None:
+    """Stamp ``scrapedAt``, apply the NSFW gate, and wrap as an output dict."""
+    if not include_nsfw and partial.get("over18") is True:
+        return None
+    return RedditItem(**{**partial, "scrapedAt": now_iso()}).to_output()
+
+
+async def _paginate_listing(
+    path: str,
+    base_params: dict[str, Any],
+    kinds: frozenset[str],
+    *,
+    max_items: int,
+    include_nsfw: bool,
+    date_limit: str | None = None,
+) -> AsyncIterator[dict[str, Any]]:
+    """Yield raw child ``data`` dicts across pages via the ``after`` cursor.
+
+    Filters by child ``kind`` (``t3``/``t1``), the NSFW gate, and ``date_limit``
+    (drops older items and, since ``date_limit`` forces newest-first, stops once
+    a page crosses the cutoff). Aborts on an empty-streak, a null ``after``, or
+    the ~1000-item page ceiling.
+    """
+    if max_items <= 0:
+        return
+    emitted = 0
+    cursor: str | None = None
+    empty_streak = 0
+    for _page in range(_MAX_PAGES):
+        params = {**base_params, "limit": _LISTING_LIMIT}
+        if cursor:
+            params["after"] = cursor
+        listing = await fetch_json(path, params)
+        kids = children(listing)
+        if not kids:
+            empty_streak += 1
+            if empty_streak >= _EMPTY_STREAK_ABORT:
+                break
+        else:
+            empty_streak = 0
+        crossed_cutoff = False
+        for child in kids:
+            if not isinstance(child, dict) or child.get("kind") not in kinds:
+                continue
+            data = child.get("data") or {}
+            if date_limit and _before(data.get("created_utc"), date_limit):
+                crossed_cutoff = True
+                continue
+            if not include_nsfw and data.get("over_18") is True:
+                continue
+            yield data
+            emitted += 1
+            if emitted >= max_items:
+                return
+        cursor = after(listing)
+        if not cursor or crossed_cutoff:
+            break
+
+
+async def _post_flow(
+    post_id: str,
+    *,
+    input_model: RedditScrapeInput,
+    subreddit: str | None = None,
+    include_post: bool = True,
+) -> AsyncIterator[dict[str, Any]]:
+    """Emit a post (unless ``include_post`` is False) plus its comment tree."""
+    path = f"r/{subreddit}/comments/{post_id}" if subreddit else f"comments/{post_id}"
+    data = await fetch_json(path)
+    if not isinstance(data, list) or not data:
+        return
+    post_children = children(data[0])
+    if include_post and post_children:
+        item = _emit(parse_post(post_children[0]), include_nsfw=input_model.includeNSFW)
+        if item is not None:
+            yield item
+    if input_model.skipComments or len(data) < 2:
+        return
+    flat = flatten_comments(
+        children(data[1]),
+        max_comments=input_model.maxComments,
+        date_limit=input_model.commentDateLimit,
+    )
+    for comment in flat:
+        item = _emit(comment, include_nsfw=input_model.includeNSFW)
+        if item is not None:
+            yield item
+
+
+async def _subreddit_flow(
+    subreddit: str,
+    *,
+    input_model: RedditScrapeInput,
+    sort: str | None = None,
+) -> AsyncIterator[dict[str, Any]]:
+    """Emit the community, then paged posts (descending into comments if asked)."""
+    if not input_model.skipCommunity:
+        about = await fetch_json(f"r/{subreddit}/about")
+        if isinstance(about, dict):
+            item = _emit(parse_community(about), include_nsfw=input_model.includeNSFW)
+            if item is not None:
+                yield item
+
+    # postDateLimit forces newest-first so the early-stop is correct.
+    sort = "new" if input_model.postDateLimit else (sort or input_model.sort)
+    if sort not in _LISTING_SORTS:
+        sort = "new"
+    params: dict[str, Any] = {}
+    if sort == "top" and input_model.time:
+        params["t"] = input_model.time
+
+    async for data in _paginate_listing(
+        f"r/{subreddit}/{sort}",
+        params,
+        frozenset({"t3"}),
+        max_items=input_model.maxPostCount,
+        include_nsfw=input_model.includeNSFW,
+        date_limit=input_model.postDateLimit,
+    ):
+        item = _emit(parse_post(data), include_nsfw=input_model.includeNSFW)
+        if item is not None:
+            yield item
+        if not input_model.skipComments and isinstance(data.get("id"), str):
+            async for comment in _post_flow(
+                data["id"],
+                input_model=input_model,
+                subreddit=subreddit,
+                include_post=False,
+            ):
+                yield comment
+
+
+async def _user_flow(
+    username: str,
+    *,
+    input_model: RedditScrapeInput,
+    content: str | None = None,
+) -> AsyncIterator[dict[str, Any]]:
+    """Page a user's overview/submitted/comments listing (mixed t3 + t1)."""
+    if content == "submitted":
+        path, kinds = f"user/{username}/submitted", frozenset({"t3"})
+    elif content == "comments":
+        path, kinds = f"user/{username}/comments", frozenset({"t1"})
+    else:
+        path = f"user/{username}"
+        kinds = frozenset({"t1"} if input_model.skipUserPosts else {"t3", "t1"})
+
+    async for data in _paginate_listing(
+        path,
+        {},
+        kinds,
+        max_items=input_model.maxItems,
+        include_nsfw=input_model.includeNSFW,
+        date_limit=input_model.postDateLimit,
+    ):
+        # A user listing mixes posts (t3) and comments (t1); a post has a title.
+        parsed = parse_post(data) if data.get("title") is not None else parse_comment(
+            data
+        )
+        item = _emit(parsed, include_nsfw=input_model.includeNSFW)
+        if item is not None:
+            yield item
+
+
+async def _search_flow(
+    query: str,
+    *,
+    input_model: RedditScrapeInput,
+    subreddit: str | None = None,
+) -> AsyncIterator[dict[str, Any]]:
+    """Global search, or in-subreddit when ``subreddit`` is set. De-dupes by id."""
+    params: dict[str, Any] = {"q": query, "sort": input_model.sort}
+    if input_model.time:
+        params["t"] = input_model.time
+    if subreddit:
+        path = f"r/{subreddit}/search"
+        params["restrict_sr"] = "on"
+    else:
+        path = "search"
+
+    seen: set[str] = set()
+    async for data in _paginate_listing(
+        path,
+        params,
+        frozenset({"t3"}),
+        max_items=input_model.maxItems,
+        include_nsfw=input_model.includeNSFW,
+        date_limit=input_model.postDateLimit,
+    ):
+        post_id = data.get("id")
+        if isinstance(post_id, str):
+            if post_id in seen:
+                continue
+            seen.add(post_id)
+        item = _emit(parse_post(data), include_nsfw=input_model.includeNSFW)
+        if item is not None:
+            yield item
+
+
+def _dispatch(
+    resolved: ResolvedUrl, input_model: RedditScrapeInput
+) -> AsyncIterator[dict[str, Any]]:
+    """Route a resolved URL to its flow (returns the flow's async generator)."""
+    if resolved.kind == "post":
+        return _post_flow(
+            resolved.value, input_model=input_model, subreddit=resolved.subreddit
+        )
+    if resolved.kind == "subreddit":
+        return _subreddit_flow(
+            resolved.value, input_model=input_model, sort=resolved.sort
+        )
+    if resolved.kind == "user":
+        return _user_flow(
+            resolved.value, input_model=input_model, content=resolved.content
+        )
+    return _search_flow(
+        resolved.value, input_model=input_model, subreddit=resolved.subreddit
+    )
+
+
+def _capped_targets(
+    resolved: list[ResolvedUrl], input_model: RedditScrapeInput
+) -> list[ResolvedUrl]:
+    """Apply the target-count caps (``maxCommunitiesCount`` / ``maxUserCount``).
+
+    These bound how many subreddit / user *targets* are scraped; per-target item
+    counts are bounded inside each flow (maxPostCount / maxItems / maxComments).
+    """
+    subs = users = 0
+    out: list[ResolvedUrl] = []
+    for r in resolved:
+        if r.kind == "subreddit":
+            if subs >= input_model.maxCommunitiesCount:
+                continue
+            subs += 1
+        elif r.kind == "user":
+            if users >= input_model.maxUserCount:
+                continue
+            users += 1
+        out.append(r)
+    return out
+
+
+async def iter_reddit(
+    input_model: RedditScrapeInput,
+) -> AsyncIterator[dict[str, Any]]:
+    """Yield flat Reddit items. ``startUrls`` override ``searches``.
+
+    Independent targets fan out concurrently; each target's ``after`` paging
+    stays sequential.
+    """
+    if input_model.startUrls:
+        resolved: list[ResolvedUrl] = []
+        for entry in input_model.startUrls:
+            r = resolve_url(entry.url)
+            if r is None:
+                logger.warning("[reddit] unrecognized URL: %s", entry.url)
+                continue
+            resolved.append(r)
+        jobs = [
+            _dispatch(r, input_model)
+            for r in _capped_targets(resolved, input_model)
+        ]
+        async for item in fan_out(jobs):
+            yield item
+        return
+
+    jobs = [
+        _search_flow(
+            query,
+            input_model=input_model,
+            subreddit=input_model.searchCommunityName,
+        )
+        for query in input_model.searches
+    ]
+    async for item in fan_out(jobs):
+        yield item
+
+
+async def scrape_reddit(
+    input_model: RedditScrapeInput, *, limit: int | None = None
+) -> list[dict[str, Any]]:
+    """Collect :func:`iter_reddit` into a list, honoring an optional ``limit``.
+
+    ``limit`` is a request-time policy guard, NOT a ceiling in the streaming
+    core.
+    """
+    results: list[dict[str, Any]] = []
+    async for item in iter_reddit(input_model):
+        results.append(item)
+        if limit is not None and len(results) >= limit:
+            break
+    return results

surfsense_backend/app/proprietary/scrapers/reddit/url_resolver.py

@@ -0,0 +1,89 @@
+"""Classify a Reddit URL into a scrape job.
+
+Covers the supported ``startUrls`` shapes: a post (permalink), a
+subreddit listing, a user profile, and a search-results page. Non-Reddit hosts
+resolve to ``None`` so the orchestrator can skip them. Mirrors the frozen
+``ResolvedUrl`` dataclass shape of ``../youtube/url_resolver.py``, widened with
+the extra context Reddit flows need (subreddit, sort, user content type).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Literal
+from urllib.parse import parse_qs, unquote, urlparse
+
+ResolvedKind = Literal["post", "subreddit", "user", "search"]
+UserContent = Literal["overview", "submitted", "comments"]
+
+_REDDIT_HOSTS = frozenset(
+    {
+        "reddit.com",
+        "www.reddit.com",
+        "old.reddit.com",
+        "new.reddit.com",
+        "np.reddit.com",
+        "m.reddit.com",
+    }
+)
+
+# Listing sorts that can appear as a trailing subreddit path segment.
+_SUBREDDIT_SORTS = frozenset(
+    {"hot", "new", "top", "rising", "controversial", "best"}
+)
+_USER_CONTENT = frozenset({"overview", "submitted", "comments"})
+
+
+@dataclass(frozen=True)
+class ResolvedUrl:
+    kind: ResolvedKind
+    value: str  # post id, subreddit, username, or search query
+    url: str
+    subreddit: str | None = None  # carried for posts and in-sub searches
+    sort: str | None = None  # trailing subreddit sort, if present
+    content: UserContent | None = None  # user profile tab
+
+
+def _is_reddit_host(hostname: str | None) -> bool:
+    return bool(hostname) and hostname.lower() in _REDDIT_HOSTS
+
+
+def resolve_url(url: str) -> ResolvedUrl | None:
+    """Classify a Reddit URL into a scrape job, or ``None`` if unrecognized."""
+    parsed = urlparse(url)
+    if not _is_reddit_host(parsed.hostname):
+        return None
+
+    segments = [s for s in (parsed.path or "").split("/") if s]
+
+    # Search: /search?q=... or /r/<sub>/search?q=...
+    if segments and segments[-1] == "search":
+        query = parse_qs(parsed.query).get("q", [None])[0]
+        if not query:
+            return None
+        sub = segments[1] if len(segments) >= 3 and segments[0] == "r" else None
+        return ResolvedUrl("search", unquote(query), url, subreddit=sub)
+
+    # Subreddit-scoped URLs: /r/<sub>/...
+    if len(segments) >= 2 and segments[0] == "r":
+        sub = segments[1]
+        # Post permalink: /r/<sub>/comments/<id>[/slug]
+        if len(segments) >= 4 and segments[2] == "comments":
+            return ResolvedUrl("post", segments[3], url, subreddit=sub)
+        # Subreddit listing, optional trailing sort: /r/<sub>[/<sort>]
+        sort = (
+            segments[2]
+            if len(segments) >= 3 and segments[2] in _SUBREDDIT_SORTS
+            else None
+        )
+        return ResolvedUrl("subreddit", sub, url, sort=sort)
+
+    # User profile: /user/<name>[/tab] or /u/<name>[/tab]
+    if len(segments) >= 2 and segments[0] in ("user", "u"):
+        name = segments[1]
+        content: UserContent | None = None
+        if len(segments) >= 3 and segments[2] in _USER_CONTENT:
+            content = segments[2]  # type: ignore[assignment]
+        return ResolvedUrl("user", name, url, content=content)
+
+    return None