Add Kodexis Reddit API provider as fallback with exam/cert filter

.env.example

@@ -17,6 +17,8 @@ REDDIT_CLIENT_ID="your_client_id"
 REDDIT_CLIENT_SECRET="your_client_secret"
 REDDIT_USER_AGENT="VoucherBot/0.1.0"
 REDDIT_INGESTION_ENABLED="false"  # false = collect via public RSS feeds, never call the OAuth API
+# Kodexis API (used when REDDIT_INGESTION_ENABLED=false)
+KODEXIS_API_KEY="kp_live_example_key"
 
 # Email Automation
 # See docs/setup.md for how to get your Resend API key and configure your email.

Sources/Subreddit.txt

@@ -10,7 +10,6 @@ FreeUdemyCoupons
 
 # Tier B — every 60 min
 O365Certification
-mcsa
 CompTIA
 ccna
 ccnp

voucherbot/config/settings.py

@@ -138,6 +138,9 @@ class Settings(BaseSettings):
     reddit_fetch_limit: int = 25
     reddit_ingestion_enabled: bool = False
 
+    # Kodexis API (fallback when REDDIT_INGESTION_ENABLED=false)
+    kodexis_api_key: Optional[str] = None
+
     # Scraping policy (see deep-research-report)
     # Identifying UA; empty → built from email contact. Do not spoof browsers.
     scraper_user_agent: Optional[str] = None

voucherbot/database/bootstrap.py

@@ -117,7 +117,6 @@ def _warn_on_invalid_selectors(config: dict[str, Any], source_name: str) -> None
     "AWSCertifications",
     "AzureCertification",
     "O365Certification",
-    "mcsa",
     "CompTIA",
     "ccna",
     "ccnp",

voucherbot/providers/Kodexis/__init__.py

@@ -0,0 +1,6 @@
+"""Kodexis Reddit provider for fetching posts via Kodexis API."""
+
+from .client import KodexisRedditClient
+
+
+__all__ = ["KodexisRedditClient"]

voucherbot/providers/Kodexis/client.py

@@ -0,0 +1,63 @@
+import httpx
+from typing import Any, Dict, List
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+
+KODEXIS_API_BASE = "https://app.kodexisapi.com"
+
+
+class KodexisRedditClient:
+    def __init__(self, api_key: str) -> None:
+        self.api_key = api_key
+        self.base_url = KODEXIS_API_BASE
+        self.http_client = httpx.AsyncClient(
+            base_url=self.base_url,
+            headers={"X-API-Key": self.api_key},
+            timeout=30.0,
+        )
+
+    async def __aenter__(self) -> "KodexisRedditClient":
+        return self
+
+    async def __aexit__(self, *args: Any) -> None:
+        await self.http_client.aclose()
+
+    async def fetch_latest_posts(
+        self, subreddit: str, limit: int = 10
+    ) -> List[Dict[str, Any]]:
+        """Fetch latest posts from a subreddit using Kodexis API."""
+        try:
+            response = await self.http_client.get(
+                f"/marketplace/reddit/api/reddit/{subreddit}",
+                params={
+                    "sort": "new",
+                    "limit": min(limit, 10),  # API max is 10
+                },
+            )
+            response.raise_for_status()
+            data = response.json()
+            # Parse Reddit's Listing format
+            posts = []
+            for child in data.get("data", {}).get("children", []):
+                if child.get("kind") == "t3":  # t3 = link/post
+                    posts.append(child.get("data", {}))
+            return posts
+        except httpx.HTTPStatusError as e:
+            logger.error(
+                "Kodexis: HTTP error fetching posts from {subreddit}",
+                subreddit=subreddit,
+                error=str(e),
+            )
+            raise
+        except Exception as e:
+            logger.error(
+                "Kodexis: error fetching posts from {subreddit}",
+                subreddit=subreddit,
+                error=str(e),
+            )
+            raise
+
+    async def close(self) -> None:
+        await self.http_client.aclose()

voucherbot/providers/reddit/collector.py

@@ -1,6 +1,7 @@
 from datetime import datetime, timezone
 from typing import Any
 from urllib.parse import quote_plus
+import re
 
 import asyncio
 
@@ -9,6 +10,7 @@
 
 from voucherbot.config.settings import settings
 from voucherbot.providers.base import BaseCollector, NormalizedPost
+from voucherbot.providers.Kodexis.client import KodexisRedditClient
 from voucherbot.providers.reddit.client import RedditClient
 from voucherbot.providers.http_policy import (
     RobotsDisallowedError,
@@ -20,10 +22,15 @@
 
 
 class RedditCollector(BaseCollector):
-    """Collects posts from a subreddit via asyncpraw."""
+    """Collects posts from a subreddit via asyncpraw or Kodexis fallback."""
 
-    def __init__(self, client: RedditClient) -> None:
-        self.client = client
+    def __init__(
+        self,
+        reddit_client: RedditClient,
+        kodexis_client: KodexisRedditClient | None = None,
+    ) -> None:
+        self.client = reddit_client
+        self.kodexis_client = kodexis_client
 
     async def collect(
         self, source_config: dict[str, Any], limit: int = 25
@@ -35,46 +42,118 @@ async def collect(
             )
             return []
 
-        if not settings.reddit_ingestion_enabled or not self.client.is_configured:
-            return await self._collect_via_rss(source_config, limit)
+        if settings.reddit_ingestion_enabled and self.client.is_configured:
+            query_terms = source_config.get("query_terms") or []
+            if query_terms:
+                query = " OR ".join(
+                    f'"{term}"' if " " in term else term for term in query_terms
+                )
+                raw_posts = await self.client.search_posts(
+                    query=query,
+                    subreddit_name=subreddit_name,
+                    limit=limit,
+                )
+            else:
+                raw_posts = await self.client.fetch_new_posts(
+                    subreddit_name, limit=limit
+                )
 
-        query_terms = source_config.get("query_terms") or []
-        if query_terms:
-            query = " OR ".join(
-                f'"{term}"' if " " in term else term for term in query_terms
+            return self._normalize_praw_posts(subreddit_name, raw_posts)
+
+        # Fallback to Kodexis API when REDDIT_INGESTION_ENABLED is false
+        if self.kodexis_client:
+            return await self._collect_via_kodexis(source_config, limit)
+
+        # Fallback to RSS when Kodexis is not configured
+        return await self._collect_via_rss(source_config, limit)
+
+    async def _collect_via_kodexis(
+        self, source_config: dict[str, Any], limit: int
+    ) -> list[NormalizedPost]:
+        subreddit_name = source_config.get("subreddit", "")
+        if not subreddit_name:
+            return []
+
+        # self.kodexis_client is guaranteed non-None by the caller check
+        client = self.kodexis_client
+        assert client is not None
+
+        try:
+            raw_posts = await client.fetch_latest_posts(
+                subreddit=subreddit_name, limit=limit
             )
-            raw_posts = await self.client.search_posts(
-                query=query,
-                subreddit_name=subreddit_name,
-                limit=limit,
+        except Exception as e:
+            logger.error(
+                "Kodexis fallback failed for {subreddit}",
+                subreddit=subreddit_name,
+                error=str(e),
             )
-        else:
-            raw_posts = await self.client.fetch_new_posts(subreddit_name, limit=limit)
+            return []
 
-        return self._normalize_praw_posts(subreddit_name, raw_posts)
+        filtered_posts = self._filter_exam_pass_posts(raw_posts)
+        return self._normalize_kodexis_posts(subreddit_name, filtered_posts, limit)
 
-    def _normalize_praw_posts(
-        self, subreddit_name: str, raw_posts: Any
+    def _filter_exam_pass_posts(
+        self, raw_posts: list[dict[str, Any]]
+    ) -> list[dict[str, Any]]:
+        """Filter out posts mentioning 'exam' or 'certification' + 'pass'/'passed'.
+
+        Only filters when 'exam' or 'certification' appear as standalone terms,
+        not as part of 'free exam', 'exam voucher', 'certification voucher', etc.
+        """
+        filtered = []
+        pass_regex = re.compile(r"\b(pass|passed|passing)\b")
+        exam_regex = re.compile(r"\bexam\b")
+        cert_regex = re.compile(r"\bcertification\b")
+
+        exclude_exam = re.compile(
+            r"\b(free exam|exam voucher|exam credit|beta exam|retake)\b"
+        )
+        exclude_cert = re.compile(r"\b(free certification|certification voucher)\b")
+
+        for post in raw_posts:
+            title = (post.get("title", "") or "").lower()
+            content = (post.get("selftext", "") or "").lower()
+            text = f"{title} {content}"
+
+            has_exam = bool(exam_regex.search(text)) and not exclude_exam.search(text)
+            has_cert = bool(cert_regex.search(text)) and not exclude_cert.search(text)
+            has_pass = bool(pass_regex.search(text))
+
+            if (has_exam or has_cert) and has_pass:
+                logger.info(
+                    "Kodexis: filtered exam/cert + pass post",
+                    title=post.get("title", "")[:80],
+                    subreddit=post.get("subreddit", ""),
+                )
+                continue
+
+            filtered.append(post)
+
+        return filtered
+
+    def _normalize_kodexis_posts(
+        self, subreddit_name: str, raw_posts: list[dict[str, Any]], limit: int
     ) -> list[NormalizedPost]:
         results: list[NormalizedPost] = []
 
-        for post in raw_posts:
+        for post in raw_posts[:limit]:
             results.append(
                 NormalizedPost(
-                    url=f"https://www.reddit.com{post.permalink}",
-                    title=post.title,
-                    content=post.selftext or None,
+                    url=f"https://www.reddit.com{post.get('permalink', '')}",
+                    title=post.get("title", "(no title)"),
+                    content=post.get("selftext") or None,
                     summary=None,
                     author=None,
                     published_at=datetime.fromtimestamp(
-                        post.created_utc, tz=timezone.utc
+                        post.get("created_utc", 0), tz=timezone.utc
                     ),
                     raw_data={
-                        "score": post.score,
-                        "num_comments": post.num_comments,
-                        "url": post.url,
+                        "score": post.get("score", 0),
+                        "num_comments": post.get("num_comments", 0),
+                        "url": post.get("url", ""),
                         "subreddit": subreddit_name,
-                        "flair": post.link_flair_text,
+                        "auth_mode": "kodexis",
                     },
                 )
             )
@@ -142,3 +221,31 @@ async def _collect_via_rss(
             )
 
         return results
+
+    def _normalize_praw_posts(
+        self, subreddit_name: str, raw_posts: Any
+    ) -> list[NormalizedPost]:
+        results: list[NormalizedPost] = []
+
+        for post in raw_posts:
+            results.append(
+                NormalizedPost(
+                    url=f"https://www.reddit.com{post.permalink}",
+                    title=post.title,
+                    content=post.selftext or None,
+                    summary=None,
+                    author=None,
+                    published_at=datetime.fromtimestamp(
+                        post.created_utc, tz=timezone.utc
+                    ),
+                    raw_data={
+                        "score": post.score,
+                        "num_comments": post.num_comments,
+                        "url": post.url,
+                        "subreddit": subreddit_name,
+                        "flair": post.link_flair_text,
+                    },
+                )
+            )
+
+        return results

voucherbot/services/scheduler.py

@@ -16,9 +16,11 @@
 import structlog
 from sqlalchemy import func, or_, select
 
+from voucherbot.config.settings import settings
 from voucherbot.database.connection import session_scope
 from voucherbot.models.notification import NotificationOutbox, NotificationStatus
 from voucherbot.models.source import Source
+from voucherbot.providers.Kodexis.client import KodexisRedditClient
 from voucherbot.providers.reddit.client import RedditClient
 from voucherbot.providers.reddit.collector import RedditCollector
 from voucherbot.providers.rss.collector import RssCollector
@@ -33,8 +35,11 @@
 logger = structlog.get_logger(__name__)
 
 _reddit_client = RedditClient()
+_kodexis_client = (
+    KodexisRedditClient(settings.kodexis_api_key) if settings.kodexis_api_key else None
+)
 _collectors = {
-    "reddit": RedditCollector(_reddit_client),
+    "reddit": RedditCollector(_reddit_client, _kodexis_client),
     "rss": RssCollector(),
     "web": WebsiteCollector(),
     "pearsonvue": PearsonVUECollector(),
@@ -180,3 +185,5 @@ async def stop_scheduler() -> None:
         _loop_task.cancel()
         await asyncio.gather(_loop_task, return_exceptions=True)
     await _reddit_client.close()
+    if _kodexis_client:
+        await _kodexis_client.close()