diff --git a/src/create_context_graph/cli.py b/src/create_context_graph/cli.py index f34fa7b..b448097 100644 --- a/src/create_context_graph/cli.py +++ b/src/create_context_graph/cli.py @@ -590,6 +590,11 @@ def main( try: conn = get_connector(conn_id) creds = config.saas_credentials.get(conn_id, {}) + # Inject API keys so connectors can use them for enrichment + if config.anthropic_api_key: + creds.setdefault("anthropic_api_key", config.anthropic_api_key) + if config.openai_api_key: + creds.setdefault("openai_api_key", config.openai_api_key) console.print(f" Connecting to {conn.service_name}...") conn.authenticate(creds) console.print(f" Fetching data from {conn.service_name}...") diff --git a/src/create_context_graph/config.py b/src/create_context_graph/config.py index 5e9b427..6f89358 100644 --- a/src/create_context_graph/config.py +++ b/src/create_context_graph/config.py @@ -64,6 +64,21 @@ } +# --------------------------------------------------------------------------- +# Reddit connector defaults +# --------------------------------------------------------------------------- + +REDDIT_DEFAULT_SUBREDDITS: list[str] = [ + "neo4j", +] + +REDDIT_DEFAULT_KEYWORDS: list[str] = [ + "rag", + "agentic ai", + "knowledge graph", +] + + class ProjectConfig(BaseModel): """All configuration collected from the wizard or CLI flags.""" diff --git a/src/create_context_graph/connectors/__init__.py b/src/create_context_graph/connectors/__init__.py index 698a0d3..4b071d3 100644 --- a/src/create_context_graph/connectors/__init__.py +++ b/src/create_context_graph/connectors/__init__.py @@ -170,4 +170,5 @@ def merge_connector_results(results: list[NormalizedData]) -> NormalizedData: from create_context_graph.connectors.claude_code_connector import ClaudeCodeConnector # noqa: E402, F401 from create_context_graph.connectors.claude_ai_connector import ClaudeAIConnector # noqa: E402, F401 from create_context_graph.connectors.chatgpt_connector import ChatGPTConnector # noqa: E402, F401 +from create_context_graph.connectors.reddit_connector import RedditConnector # noqa: E402, F401 from create_context_graph.connectors.local_file_connector import LocalFileConnector # noqa: E402, F401 diff --git a/src/create_context_graph/connectors/reddit_connector.py b/src/create_context_graph/connectors/reddit_connector.py new file mode 100644 index 0000000..c934a37 --- /dev/null +++ b/src/create_context_graph/connectors/reddit_connector.py @@ -0,0 +1,848 @@ +# Copyright 2026 Neo4j Labs +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Reddit connector — imports posts, comments, and authors from public subreddits. + +Works in two modes: + +Unauthenticated (default): + Uses the Reddit public JSON API — no credentials required. + Rate limit: ~10 requests/min. A 6-second delay is applied automatically. + +Authenticated (Reddit script app): + Supply client_id, client_secret, reddit_username, and reddit_password via + the credential prompts. Uses OAuth2 password grant to obtain a bearer token. + Rate limit increases to ~60 requests/min. Delay drops to 1 second. + + To create a script app: https://www.reddit.com/prefs/apps + Select "script", set redirect URI to http://localhost:8080. + +Designed as a general-purpose product discovery and community intelligence +connector. Configure target subreddits and search keywords to track any +product, technology, or topic across Reddit communities. + +LLM enrichment follows the same provider pattern as the rest of create-context-graph: +Anthropic is used if ANTHROPIC_API_KEY / anthropic_api_key is available, +OpenAI is the fallback if openai_api_key is available. + +Subreddits and keywords are configured in config.py via +``REDDIT_DEFAULT_SUBREDDITS`` and ``REDDIT_DEFAULT_KEYWORDS``, or can be +overridden at runtime through the credential prompts. +""" + +from __future__ import annotations + +import json +import logging +import os +import time +import urllib.error +import urllib.parse +import urllib.request +from datetime import datetime, timezone +from typing import Any + +from create_context_graph.connectors import ( + BaseConnector, + NormalizedData, + register_connector, +) +from create_context_graph.config import REDDIT_DEFAULT_KEYWORDS, REDDIT_DEFAULT_SUBREDDITS + +logger = logging.getLogger(__name__) + +BASE_URL = "https://www.reddit.com" +OAUTH_URL = "https://oauth.reddit.com" +TOKEN_URL = "https://www.reddit.com/api/v1/access_token" +RESULTS_PER_PAGE = 100 + +# Rate limits: unauthenticated ~10 req/min (6s delay), authenticated ~60 req/min (1s delay) +REQUEST_DELAY_UNAUTHED = 6.0 +REQUEST_DELAY_AUTHED = 1.0 +COMMENT_DELAY_UNAUTHED = 3.0 +COMMENT_DELAY_AUTHED = 0.5 + +USER_AGENT = "python:neo4j-context-graph:v1.0 (research pipeline; neo4j-labs)" + +HEADERS = { + "User-Agent": USER_AGENT, + "Accept": "application/json", +} + +_ENRICHMENT_SYSTEM_PROMPT = """You are a data extraction assistant. Extract structured product discovery insights from Reddit posts. + +Return ONLY valid JSON with this exact structure: +{ + "pain_points": ["concise description of a specific problem or frustration mentioned"], + "use_cases": ["concise description of a specific application or use case mentioned"], + "topics": ["topic tag"], + "technologies": ["ProductOrToolName"] +} + +Rules: +- pain_points: real problems, frustrations, or unmet needs the author describes, max 3, empty list if none +- use_cases: concrete applications, workflows, or patterns being built or described, max 3, empty list if none +- topics: 1-4 word subject tags describing the discussion theme (e.g. "performance", "getting started", "integration"), max 5 +- technologies: proper-cased product, tool, or service names explicitly mentioned, max 10 +- Return empty lists if nothing relevant found, never null""" + + +def _get_llm_client(anthropic_api_key: str | None, openai_api_key: str | None): + """Return (client, provider) following the same pattern as generator.py.""" + if anthropic_api_key: + try: + import anthropic + return anthropic.Anthropic(api_key=anthropic_api_key), "anthropic" + except ImportError: + pass + if openai_api_key: + try: + import openai + return openai.OpenAI(api_key=openai_api_key), "openai" + except ImportError: + pass + return None, None + + +def _enrich_post( + title: str, + body: str, + anthropic_api_key: str | None, + openai_api_key: str | None, +) -> dict: + """Extract structured product insights using the configured LLM provider. + + Follows the same Anthropic-first, OpenAI-fallback pattern as generator.py. + """ + client, provider = _get_llm_client(anthropic_api_key, openai_api_key) + if client is None: + return {} + try: + text = f"Title: {title}\n\n{body[:2000]}" + if provider == "anthropic": + response = client.messages.create( + model="claude-haiku-4-5-20251001", + max_tokens=512, + system=_ENRICHMENT_SYSTEM_PROMPT, + messages=[{"role": "user", "content": text}], + ) + raw = response.content[0].text.strip() + else: + response = client.chat.completions.create( + model="gpt-4o-mini", + max_tokens=512, + messages=[ + {"role": "system", "content": _ENRICHMENT_SYSTEM_PROMPT}, + {"role": "user", "content": text}, + ], + ) + raw = response.choices[0].message.content.strip() + # Strip markdown code fences if present + if raw.startswith("```"): + raw = raw.split("```")[1] + if raw.startswith("json"): + raw = raw[4:] + return json.loads(raw) + except Exception as exc: + logger.debug("Enrichment failed for post '%s': %s", title[:50], exc) + return {} + + +def _fetch_oauth_token(client_id: str, client_secret: str, username: str, password: str) -> str | None: + """Obtain a Reddit OAuth2 bearer token via password grant (script apps only).""" + import base64 + credentials = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() + data = urllib.parse.urlencode({ + "grant_type": "password", + "username": username, + "password": password, + }).encode() + req = urllib.request.Request( + TOKEN_URL, + data=data, + headers={ + "Authorization": f"Basic {credentials}", + "User-Agent": USER_AGENT, + "Content-Type": "application/x-www-form-urlencoded", + }, + ) + try: + with urllib.request.urlopen(req, timeout=15) as resp: + result = json.loads(resp.read()) + token = result.get("access_token") + if token: + logger.info("Reddit OAuth token obtained — authenticated mode active.") + else: + logger.warning("Reddit OAuth response missing access_token: %s", result) + return token + except Exception as exc: + logger.warning("Failed to obtain Reddit OAuth token: %s", exc) + return None + + +def _get_json( + url: str, + params: dict | None = None, + retries: int = 3, + extra_headers: dict | None = None, +) -> dict | None: + """Fetch a Reddit JSON endpoint, respecting rate limits.""" + if params: + query = "&".join(f"{k}={urllib.parse.quote(str(v))}" for k, v in params.items()) + url = f"{url}?{query}" + + headers = {**HEADERS, **(extra_headers or {})} + for attempt in range(1, retries + 1): + req = urllib.request.Request(url, headers=headers) + try: + with urllib.request.urlopen(req, timeout=15) as resp: + return json.loads(resp.read()) + except urllib.error.HTTPError as exc: + if exc.code == 429: + wait = 30 * attempt + logger.warning("Reddit rate-limited. Sleeping %ds…", wait) + time.sleep(wait) + continue + if exc.code in (403, 404): + return None + logger.warning("HTTP %d on attempt %d/%d: %s", exc.code, attempt, retries, url) + except Exception as exc: + logger.warning("Request error (attempt %d/%d): %s", attempt, retries, exc) + time.sleep(5 * attempt) + return None + + +def _parse_timestamp(ts: Any) -> str: + """Convert a Unix timestamp (int/float) to an ISO-8601 string.""" + try: + return datetime.fromtimestamp(float(ts), tz=timezone.utc).isoformat() + except (ValueError, TypeError, OSError): + return datetime.now(tz=timezone.utc).isoformat() + + +@register_connector("reddit") +class RedditConnector(BaseConnector): + """Import posts and comments from Reddit via the public JSON API. + + No API key or OAuth is required. Configure target subreddits and keywords + in ``config.py`` (``REDDIT_DEFAULT_SUBREDDITS`` / ``REDDIT_DEFAULT_KEYWORDS``), + or override them through the credential prompts at runtime. + """ + + service_name = "Reddit" + service_description = ( + "Import posts and comments from Reddit subreddits for product discovery and community intelligence. " + "Works unauthenticated (public API) or with a Reddit script app (higher rate limits). " + "Configure target subreddits and search keywords in config.py." + ) + requires_oauth = False + + def __init__(self) -> None: + self._subreddits: list[str] = list(REDDIT_DEFAULT_SUBREDDITS) + self._keywords: list[str] = list(REDDIT_DEFAULT_KEYWORDS) + self._max_pages: int = 1 + self._fetch_comments: bool = True + self._enrich_posts: bool = True + self._max_post_age_days: int = 1095 # ~3 years + # Auth state (set by authenticate()) + self._auth_headers: dict[str, str] = {} + self._request_delay: float = REQUEST_DELAY_UNAUTHED + self._comment_delay: float = COMMENT_DELAY_UNAUTHED + self._anthropic_api_key: str | None = None + self._openai_api_key: str | None = None + + # ------------------------------------------------------------------ + # BaseConnector interface + # ------------------------------------------------------------------ + + def get_credential_prompts(self) -> list[dict[str, Any]]: + return [ + { + "name": "subreddits", + "prompt": "Subreddits to scrape (comma-separated, or leave blank for defaults):", + "secret": False, + "optional": True, + "description": ( + f"Defaults: {', '.join(REDDIT_DEFAULT_SUBREDDITS[:4])} " + "Override by listing subreddits without the r/ prefix." + ), + }, + { + "name": "keywords", + "prompt": "Search keywords (comma-separated, or leave blank for defaults):", + "secret": False, + "optional": True, + "description": ( + f"Defaults: {', '.join(REDDIT_DEFAULT_KEYWORDS)} " + "Posts matching any keyword in any target subreddit are imported." + ), + }, + { + "name": "max_pages", + "prompt": "Max pages per keyword/subreddit combination (default: 1, ~100 posts/page):", + "secret": False, + "optional": True, + "description": "1 page ≈ 100 posts. Increase for deeper historical data.", + }, + { + "name": "fetch_comments", + "prompt": "Fetch post comments? (yes/no, default: yes):", + "secret": False, + "optional": True, + "description": "Fetching comments makes the import slower (extra delay per post).", + }, + { + "name": "enrich_posts", + "prompt": "Enrich posts with LLM extraction of pain points, use cases, and topics? (yes/no, default: yes):", + "secret": False, + "optional": True, + "description": ( + "Uses the configured LLM provider (Anthropic or OpenAI) to extract " + "PainPoint, UseCase, Topic, and Technology entities from each post." + ), + }, + # Optional Reddit OAuth credentials (script app) + { + "name": "client_id", + "prompt": "Reddit app client_id (optional — leave blank for unauthenticated access):", + "secret": False, + "optional": True, + "description": ( + "Create a script app at https://www.reddit.com/prefs/apps. " + "Authenticated requests allow ~60 req/min vs 10 req/min unauthenticated." + ), + }, + { + "name": "client_secret", + "prompt": "Reddit app client_secret (optional):", + "secret": True, + "optional": True, + "description": "Secret shown under your Reddit script app.", + }, + { + "name": "reddit_username", + "prompt": "Reddit username (optional — required with client_id):", + "secret": False, + "optional": True, + "description": "Your Reddit account username (without u/).", + }, + { + "name": "reddit_password", + "prompt": "Reddit password (optional — required with client_id):", + "secret": True, + "optional": True, + "description": "Your Reddit account password.", + }, + ] + + def authenticate(self, credentials: dict[str, str]) -> None: + """Store scrape configuration and optionally obtain a Reddit OAuth token.""" + raw_subs = (credentials.get("subreddits") or "").strip() + raw_kws = (credentials.get("keywords") or "").strip() + raw_pages = (credentials.get("max_pages") or "").strip() + raw_comments = (credentials.get("fetch_comments") or "yes").strip().lower() + raw_enrich = (credentials.get("enrich_posts") or "yes").strip().lower() + + self._subreddits = ( + [s.strip() for s in raw_subs.split(",") if s.strip()] + or list(REDDIT_DEFAULT_SUBREDDITS) + ) + self._keywords = ( + [k.strip() for k in raw_kws.split(",") if k.strip()] + or list(REDDIT_DEFAULT_KEYWORDS) + ) + self._max_pages = max(1, int(raw_pages)) if raw_pages.isdigit() else 1 + self._fetch_comments = raw_comments not in ("no", "false", "0", "n") + self._enrich_posts = raw_enrich not in ("no", "false", "0", "n") + + # LLM keys — injected by cli.py or read from environment + self._anthropic_api_key = ( + credentials.get("anthropic_api_key") + or os.environ.get("ANTHROPIC_API_KEY") + ) + self._openai_api_key = ( + credentials.get("openai_api_key") + or os.environ.get("OPENAI_API_KEY") + ) + + # Reddit OAuth — optional, enables higher rate limits + client_id = (credentials.get("client_id") or "").strip() + client_secret = (credentials.get("client_secret") or "").strip() + reddit_username = (credentials.get("reddit_username") or "").strip() + reddit_password = (credentials.get("reddit_password") or "").strip() + + if client_id and client_secret and reddit_username and reddit_password: + token = _fetch_oauth_token(client_id, client_secret, reddit_username, reddit_password) + if token: + self._auth_headers = {"Authorization": f"bearer {token}"} + self._request_delay = REQUEST_DELAY_AUTHED + self._comment_delay = COMMENT_DELAY_AUTHED + else: + logger.warning("Reddit auth failed — falling back to unauthenticated access.") + self._auth_headers = {} + self._request_delay = REQUEST_DELAY_UNAUTHED + self._comment_delay = COMMENT_DELAY_UNAUTHED + else: + self._auth_headers = {} + self._request_delay = REQUEST_DELAY_UNAUTHED + self._comment_delay = COMMENT_DELAY_UNAUTHED + + logger.info( + "Reddit connector configured: %d subreddits, %d keywords, " + "max_pages=%d, fetch_comments=%s", + len(self._subreddits), + len(self._keywords), + self._max_pages, + self._fetch_comments, + ) + + def fetch(self, **kwargs: Any) -> NormalizedData: + """Scrape configured subreddits and return normalised data. + + Entity labels produced (matches the product-discovery domain): + Person — Reddit authors (aligned with _base.yaml Person type) + Subreddit — communities (ORGANIZATION pole_type) + Post — Reddit posts + Comment — post comments + Technology — products/tools detected in post text + Topic — subject tags from flair and LLM extraction + PainPoint — problems/frustrations extracted via LLM enrichment + UseCase — applications/patterns extracted via LLM enrichment + """ + # Determine base URL — OAuth uses a different host + base = OAUTH_URL if self._auth_headers else BASE_URL + authed = bool(self._auth_headers) + + # LLM enrichment setup — Anthropic first, OpenAI fallback (mirrors generator.py) + do_enrich = self._enrich_posts and (self._anthropic_api_key or self._openai_api_key) + if self._enrich_posts and do_enrich: + provider = "Anthropic" if self._anthropic_api_key else "OpenAI" + logger.info("LLM enrichment enabled via %s — PainPoints, UseCases, Technologies, Topics.", provider) + elif self._enrich_posts and not do_enrich: + logger.warning("enrich_posts=yes but no LLM API key found — skipping enrichment.") + else: + logger.info("LLM enrichment disabled.") + + entities: dict[str, list[dict]] = { + "Person": [], + "Subreddit": [], + "Post": [], + "Comment": [], + "Technology": [], + "Topic": [], + "PainPoint": [], + "UseCase": [], + } + + seen_pain_points: set[str] = set() + seen_use_cases: set[str] = set() + relationships: list[dict] = [] + documents: list[dict] = [] + + seen_post_ids: set[str] = set() + seen_users: set[str] = set() + seen_subs: set[str] = set() + seen_techs: set[str] = set() + seen_topics: set[str] = set() + + # ---------------------------------------------------------------- + # Helpers + # ---------------------------------------------------------------- + + def _ensure_subreddit(name: str) -> None: + key = name.lower() + if key not in seen_subs: + seen_subs.add(key) + entities["Subreddit"].append({ + "name": key, + "url": f"https://www.reddit.com/r/{key}/", + "description": f"r/{key} community", + }) + + def _ensure_user(username: str) -> None: + if username and username not in seen_users: + seen_users.add(username) + entities["Person"].append({ + "name": username, + "role": "reddit-user", + "description": f"Reddit user u/{username}", + }) + + def _ensure_technology(name: str) -> None: + if name and name not in seen_techs: + seen_techs.add(name) + entities["Technology"].append({ + "name": name, + "description": f"Technology/product mentioned in community posts", + }) + + def _ensure_topic(name: str) -> None: + if name and name not in seen_topics: + seen_topics.add(name) + entities["Topic"].append({ + "name": name, + "description": f"Discussion topic found in r/ communities", + }) + + # Pre-populate Product/Technology nodes directly from the keyword list. + # Keywords are treated as product/topic names — proper-cased for display. + for kw in self._keywords: + if len(kw) > 2: + _ensure_technology(kw.title()) + + # ---------------------------------------------------------------- + # Scrape + # ---------------------------------------------------------------- + for sub_idx, subreddit in enumerate(self._subreddits, 1): + sub_key = subreddit.lower() + logger.info("[%d/%d] Scraping r/%s", sub_idx, len(self._subreddits), sub_key) + _ensure_subreddit(sub_key) + + for keyword in self._keywords: + logger.info(" Keyword '%s' in r/%s…", keyword, sub_key) + params: dict[str, Any] = { + "q": keyword, + "sort": "new", + "restrict_sr": "on", + "limit": RESULTS_PER_PAGE, + "t": "all", + "raw_json": 1, + } + url = f"{base}/r/{sub_key}/search.json" + after: str | None = None + page = 0 + + while page < self._max_pages: + if after: + params["after"] = after + time.sleep(self._request_delay) + data = _get_json(url, params=params, extra_headers=self._auth_headers) + if not data or not isinstance(data, dict): + break + + listing = data.get("data", {}) + children = listing.get("children", []) + if not children: + break + + new_posts: list[dict] = [] + for child in children: + d = child.get("data", {}) + pid = d.get("id") + if not pid or pid in seen_post_ids: + continue + + # Age filter + try: + age = ( + datetime.now(tz=timezone.utc) + - datetime.fromtimestamp(float(d.get("created_utc", 0)), tz=timezone.utc) + ).total_seconds() / 86400 + if age > self._max_post_age_days: + continue + except Exception: + pass + + seen_post_ids.add(pid) + + author = d.get("author", "") + if author in ("[deleted]", ""): + author = None + flair = d.get("link_flair_text") or "" + created = _parse_timestamp(d.get("created_utc", 0)) + title = d.get("title", "") + body = d.get("selftext", "") + + new_posts.append({ + "pid": pid, + "subreddit": sub_key, + "title": title, + "body": body, + "author": author, + "score": int(d.get("score", 0)), + "upvote_ratio": float(d.get("upvote_ratio", 0.0)), + "num_comments": int(d.get("num_comments", 0)), + "flair": flair, + "permalink": f"https://www.reddit.com{d.get('permalink', '')}", + "created": created, + "is_self": bool(d.get("is_self", True)), + }) + + # Fetch comments before building entities (progress logging) + if self._fetch_comments and new_posts: + logger.info( + " Fetching comments for %d posts (~%ds)…", + len(new_posts), + int(len(new_posts) * COMMENT_DELAY), + ) + + for i, post in enumerate(new_posts, 1): + pid = post["pid"] + title = post["title"] + body = post["body"] + author = post["author"] + sub_key = post["subreddit"] + + # Post entity + entities["Post"].append({ + "name": f"{pid}: {title[:80]}", + "post_id": pid, + "subreddit": sub_key, + "title": title, + "body": body[:2000], + "score": post["score"], + "upvote_ratio": post["upvote_ratio"], + "num_comments": post["num_comments"], + "flair": post["flair"], + "permalink": post["permalink"], + "created_utc": post["created"], + "is_self": post["is_self"], + }) + + # Author + if author: + _ensure_user(author) + relationships.append({ + "type": "POSTED", + "source_name": author, + "source_label": "Person", + "target_name": f"{pid}: {title[:80]}", + "target_label": "Post", + }) + relationships.append({ + "type": "ACTIVE_IN", + "source_name": author, + "source_label": "Person", + "target_name": sub_key, + "target_label": "Subreddit", + }) + + # Post → Subreddit + relationships.append({ + "type": "IN_SUBREDDIT", + "source_name": f"{pid}: {title[:80]}", + "source_label": "Post", + "target_name": sub_key, + "target_label": "Subreddit", + }) + + # Keyword → Technology link + tech = keyword.title() if len(keyword) > 2 else None + if tech: + _ensure_technology(tech) + relationships.append({ + "type": "MENTIONS", + "source_name": f"{pid}: {title[:80]}", + "source_label": "Post", + "target_name": tech, + "target_label": "Technology", + }) + + # Flair → Topic + if post["flair"]: + _ensure_topic(post["flair"]) + relationships.append({ + "type": "TAGGED_WITH", + "source_name": f"{pid}: {title[:80]}", + "source_label": "Post", + "target_name": post["flair"], + "target_label": "Topic", + }) + + # LLM enrichment — PainPoints, UseCases, Topics, Technologies + if do_enrich: + enriched = _enrich_post(title, body, self._anthropic_api_key, self._openai_api_key) + post_name = f"{pid}: {title[:80]}" + + for pp in enriched.get("pain_points", []): + pp = pp.strip() + if pp: + if pp not in seen_pain_points: + seen_pain_points.add(pp) + entities["PainPoint"].append({ + "name": pp, + "description": pp, + "frequency": 1, + }) + relationships.append({ + "type": "HAS_PAIN_POINT", + "source_name": post_name, + "source_label": "Post", + "target_name": pp, + "target_label": "PainPoint", + }) + + for uc in enriched.get("use_cases", []): + uc = uc.strip() + if uc: + if uc not in seen_use_cases: + seen_use_cases.add(uc) + entities["UseCase"].append({ + "name": uc, + "description": uc, + "frequency": 1, + }) + relationships.append({ + "type": "DEMONSTRATES", + "source_name": post_name, + "source_label": "Post", + "target_name": uc, + "target_label": "UseCase", + }) + + for topic in enriched.get("topics", []): + topic = topic.strip() + if topic: + _ensure_topic(topic) + relationships.append({ + "type": "TAGGED_WITH", + "source_name": post_name, + "source_label": "Post", + "target_name": topic, + "target_label": "Topic", + }) + + for tech in enriched.get("technologies", []): + tech = tech.strip() + if tech: + _ensure_technology(tech) + relationships.append({ + "type": "MENTIONS", + "source_name": post_name, + "source_label": "Post", + "target_name": tech, + "target_label": "Technology", + }) + + # Post body as document + full_text = f"{title}\n\n{body}".strip() + if full_text: + documents.append({ + "title": f"r/{sub_key}: {title[:80]}", + "content": full_text[:4000], + "type": "reddit-post", + "metadata": { + "post_id": pid, + "subreddit": sub_key, + "author": author or "[deleted]", + "score": post["score"], + "permalink": post["permalink"], + "keyword": keyword, + }, + }) + + # Comments + if self._fetch_comments: + logger.info( + " [%d/%d] comments for %s", i, len(new_posts), pid + ) + comments = self._fetch_post_comments(pid, sub_key) + for comment in comments: + c_author = comment.get("author") + c_body = comment.get("body", "") + c_id = comment.get("id", "") + c_name = f"{c_id}: {c_body[:60]}" + + entities["Comment"].append({ + "name": c_name, + "comment_id": c_id, + "post_id": pid, + "body": c_body[:1000], + "score": comment.get("score", 0), + "depth": comment.get("depth", 0), + "created_utc": comment.get("created_utc", ""), + "permalink": comment.get("permalink", ""), + "is_top_level": comment.get("depth", 0) == 0, + }) + + # Comment → Post + relationships.append({ + "type": "ON_POST", + "source_name": c_name, + "source_label": "Comment", + "target_name": f"{pid}: {title[:80]}", + "target_label": "Post", + }) + + # Author → Comment + if c_author: + _ensure_user(c_author) + relationships.append({ + "type": "AUTHORED_BY", + "source_name": c_name, + "source_label": "Comment", + "target_name": c_author, + "target_label": "Person", + }) + relationships.append({ + "type": "ACTIVE_IN", + "source_name": c_author, + "source_label": "Person", + "target_name": sub_key, + "target_label": "Subreddit", + }) + + after = listing.get("after") + if not after: + break + page += 1 + + if sub_idx < len(self._subreddits): + time.sleep(self._request_delay) + + total_entities = sum(len(v) for v in entities.values()) + logger.info( + "Reddit import complete: %d entities, %d relationships, %d documents", + total_entities, len(relationships), len(documents), + ) + return NormalizedData( + entities=entities, + relationships=relationships, + documents=documents, + ) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _fetch_post_comments(self, post_id: str, subreddit: str, max_comments: int = 50) -> list[dict]: + """Fetch top-level comments for a single post.""" + base = OAUTH_URL if self._auth_headers else BASE_URL + url = f"{base}/r/{subreddit}/comments/{post_id}.json?raw_json=1" + time.sleep(self._comment_delay) + data = _get_json(url, extra_headers=self._auth_headers) + if not data or not isinstance(data, list) or len(data) < 2: + return [] + + comments = [] + for child in data[1].get("data", {}).get("children", [])[:max_comments]: + if child.get("kind") != "t1": + continue + d = child.get("data", {}) + body = d.get("body", "") + if body in ("[deleted]", "[removed]", ""): + continue + author = d.get("author", "") + if author in ("[deleted]", ""): + author = None + comments.append({ + "id": d.get("id", ""), + "body": body, + "author": author, + "score": int(d.get("score", 0)), + "depth": int(d.get("depth", 0)), + "created_utc": _parse_timestamp(d.get("created_utc", 0)), + "permalink": f"https://www.reddit.com{d.get('permalink', '')}", + }) + return comments diff --git a/src/create_context_graph/domains/product-discovery.yaml b/src/create_context_graph/domains/product-discovery.yaml new file mode 100644 index 0000000..f2de477 --- /dev/null +++ b/src/create_context_graph/domains/product-discovery.yaml @@ -0,0 +1,591 @@ +inherits: _base + +domain: + id: product-discovery + name: Product Discovery & Community Intelligence + description: "Community-driven product discovery intelligence built from Reddit discussions — surfacing pain points, use cases, feature requests, and sentiment across any product or technology community." + tagline: "Turn community discussions into product insights" + emoji: "🔍" + +entity_types: + - label: Post + pole_type: OBJECT + subtype: COMMUNITY_POST + color: "#ff4500" + icon: file-text + properties: + - name: post_id + type: string + required: true + unique: true + - name: title + type: string + required: true + - name: body + type: string + - name: subreddit + type: string + - name: score + type: integer + - name: upvote_ratio + type: float + - name: num_comments + type: integer + - name: flair + type: string + - name: permalink + type: string + - name: created_utc + type: string + - name: is_self + type: boolean + - name: sentiment_label + type: string + enum: [positive, negative, neutral, mixed] + - name: sentiment_score + type: float + - name: summary + type: string + - name: is_question + type: boolean + - name: has_code + type: boolean + + - label: Comment + pole_type: OBJECT + subtype: COMMUNITY_COMMENT + color: "#ff6534" + icon: message-circle + properties: + - name: comment_id + type: string + required: true + unique: true + - name: post_id + type: string + - name: body + type: string + required: true + - name: score + type: integer + - name: depth + type: integer + - name: created_utc + type: string + - name: permalink + type: string + - name: is_top_level + type: boolean + - name: sentiment_label + type: string + enum: [positive, negative, neutral, mixed] + - name: sentiment_score + type: float + - name: is_question + type: boolean + - name: has_code + type: boolean + + - label: Subreddit + pole_type: ORGANIZATION + subtype: ONLINE_COMMUNITY + color: "#fb923c" + icon: users + properties: + - name: name + type: string + required: true + unique: true + - name: url + type: string + - name: description + type: string + - name: subscribers + type: integer + + - label: Technology + pole_type: OBJECT + subtype: PRODUCT + color: "#0ea5e9" + icon: cpu + properties: + - name: name + type: string + required: true + unique: true + - name: description + type: string + - name: category + type: string + enum: [product, tool, platform, framework, service, concept, other] + - name: mention_count + type: integer + + - label: Topic + pole_type: OBJECT + subtype: SUBJECT_TAG + color: "#8b5cf6" + icon: tag + properties: + - name: name + type: string + required: true + unique: true + - name: description + type: string + + - label: PainPoint + pole_type: OBJECT + subtype: PROBLEM_AREA + color: "#ef4444" + icon: alert-circle + properties: + - name: name + type: string + required: true + unique: true + - name: description + type: string + - name: frequency + type: integer + + - label: UseCase + pole_type: OBJECT + subtype: APPLICATION_PATTERN + color: "#10b981" + icon: lightbulb + properties: + - name: name + type: string + required: true + unique: true + - name: description + type: string + - name: frequency + type: integer + +relationships: + # Authorship + - type: POSTED + source: Person + target: Post + - type: AUTHORED_BY + source: Comment + target: Person + - type: ACTIVE_IN + source: Person + target: Subreddit + + # Content structure + - type: IN_SUBREDDIT + source: Post + target: Subreddit + - type: ON_POST + source: Comment + target: Post + - type: REPLY_TO + source: Comment + target: Comment + + # Knowledge links + - type: MENTIONS + source: Post + target: Technology + - type: MENTIONS + source: Comment + target: Technology + - type: TAGGED_WITH + source: Post + target: Topic + - type: TAGGED_WITH + source: Comment + target: Topic + - type: HAS_PAIN_POINT + source: Post + target: PainPoint + - type: DEMONSTRATES + source: Post + target: UseCase + + # Product graph + - type: CO_OCCURS_WITH + source: Technology + target: Technology + - type: RELATED_TO + source: Technology + target: Technology + +document_templates: + - id: community_digest + name: Weekly Community Digest + description: Summary of top discussions from target subreddits + count: 4 + prompt_template: | + Write a weekly community digest for product and engineering teams. + Cover top discussions from {{subreddit_list}} this week. + Include trending products ({{tech_list}}), common questions, pain + points, and noteworthy use cases. Format as a newsletter-style summary. + required_entities: [Post, Subreddit, Technology] + + - id: product_overview + name: Product Community Adoption Overview + description: Analysis of how a product is discussed and adopted across communities + count: 5 + prompt_template: | + Write a community adoption overview for {{technology.name}}. + Include how it is discussed in {{subreddit_list}}, common use cases, + pain points developers have reported, and how it compares to related + products ({{related_tech_list}}). Base this on real community + discussion patterns. + required_entities: [Technology, Post, PainPoint, UseCase] + + - id: use_case_writeup + name: Use Case Deep Dive + description: Detailed analysis of a community-validated use case + count: 5 + prompt_template: | + Write a technical deep dive for the use case "{{use_case.name}}". + Draw from real community discussions in {{subreddit_list}}. + Cover the problem being solved, recommended products + ({{tech_list}}), common pitfalls ({{pain_point_list}}), and + implementation approaches shared by practitioners. + required_entities: [UseCase, Technology, PainPoint, Post] + + - id: pain_point_analysis + name: Pain Point Analysis + description: Analysis of recurring problems reported by the community + count: 5 + prompt_template: | + Write an analysis of the pain point "{{pain_point.name}}" as discussed + in {{subreddit_list}}. Include how often it is raised, what products + are involved ({{tech_list}}), proposed solutions from the community, and + any unresolved issues. Reference actual post patterns. + required_entities: [PainPoint, Technology, Post] + + - id: subreddit_profile + name: Community Profile + description: Intelligence profile of a subreddit community + count: 3 + prompt_template: | + Write a community intelligence profile for r/{{subreddit.name}}. + Cover the community's focus, top products discussed ({{tech_list}}), + most active contributors, common question types, sentiment trends, and + how it relates to adjacent communities. Include insights useful for + developer relations and product teams. + required_entities: [Subreddit, Technology, Person, Topic] + +decision_traces: + - id: product_adoption + task: "Assess community readiness to adopt {{technology.name}} for {{use_case.name}}" + steps: + - thought: "Gauge community sentiment and discussion volume around the product" + action: "Query posts mentioning the product and compute sentiment distribution" + - thought: "Identify pain points that might hinder adoption" + action: "Find pain points co-occurring with the product across posts" + - thought: "Discover success stories and validated use cases" + action: "Query posts tagged with positive sentiment demonstrating this use case" + - thought: "Understand the competitive landscape from community discussions" + action: "Find products co-occurring with target and compare mention frequency" + outcome_template: "Adoption readiness: {{readiness}}. Key blockers: {{blockers}}. Champions: {{top_advocates}}" + + - id: pain_point_prioritization + task: "Prioritize pain points to address for {{technology.name}} based on community signal" + steps: + - thought: "Identify the most frequently mentioned pain points" + action: "Query pain points linked to the product ordered by frequency" + - thought: "Assess severity from post sentiment scores" + action: "Correlate pain points with negative-sentiment posts and their scores" + - thought: "Determine which pain points are already being addressed" + action: "Search for posts where pain point is mentioned alongside solution keywords" + - thought: "Map pain points to specific communities for targeted response" + action: "Break down pain point frequency by subreddit" + outcome_template: "Top pain points: {{pain_points}}. Highest community impact: {{top_issue}}" + + - id: content_strategy + task: "Determine content strategy for reaching the {{subreddit.name}} community" + steps: + - thought: "Understand what topics and formats perform best in this community" + action: "Query top-scored posts by topic and format (code, tutorial, question)" + - thought: "Identify the most engaged and influential community members" + action: "Find persons with most posts and comments in the subreddit" + - thought: "Map gaps between community questions and available resources" + action: "Find is_question=true posts with low comment counts (unanswered)" + - thought: "Assess trending products and topics for timely content" + action: "Query products with rising mention counts in recent posts" + outcome_template: "Strategy: {{strategy}}. Priority topics: {{topics}}. Format recommendation: {{format}}" + + - id: competitive_intelligence + task: "Analyse community perception of {{technology.name}} vs competitors" + steps: + - thought: "Find posts comparing the product to alternatives" + action: "Query posts mentioning both the product and related products" + - thought: "Assess sentiment differential between the product and competitors" + action: "Compare average sentiment scores for posts mentioning each product" + - thought: "Identify use cases where community prefers each option" + action: "Find use cases co-occurring with each product and their frequency" + - thought: "Determine switching intent signals from the community" + action: "Search for posts where product and 'migrate', 'switch', or 'replace' co-occur" + outcome_template: "Competitive position: {{position}}. Strengths: {{strengths}}. Risks: {{risks}}" + + - id: community_health_check + task: "Assess health and engagement of the {{subreddit.name}} community" + steps: + - thought: "Evaluate recent posting frequency and engagement trends" + action: "Query post volume and average score by month for the subreddit" + - thought: "Measure question-to-answer ratio to gauge community responsiveness" + action: "Count is_question posts and compare with those having comment replies" + - thought: "Identify top contributors and concentration risk" + action: "Find persons with the most posts and assess diversity of contributors" + - thought: "Assess topic diversity and community breadth" + action: "Count posts per topic and product to evaluate community breadth" + outcome_template: "Community health: {{health_score}}. Growth trend: {{trend}}. Key contributors: {{contributors}}" + +demo_scenarios: + - name: Product Intelligence + prompts: + - "What are people saying about {{product}} in the community this month?" + - "Show me the most discussed use cases across all subreddits" + - "Which products are most often mentioned together?" + + - name: Pain Point Discovery + prompts: + - "What are the top pain points developers report?" + - "What questions are going unanswered in the community?" + - "Show me recurring complaints from the last 30 days" + + - name: Trend Analysis + prompts: + - "What topics are trending right now?" + - "How has community sentiment changed over time?" + - "Which subreddits have the most active discussions?" + + - name: Competitive Intelligence + prompts: + - "How does the community compare products in this space?" + - "What are people migrating away from?" + - "Show posts where developers discuss switching between tools" + +agent_tools: + - name: search_posts + description: Search posts by keyword or topic + cypher: | + MATCH (p:Post) + WHERE toLower(p.title) CONTAINS toLower($query) + OR toLower(coalesce(p.body, '')) CONTAINS toLower($query) + OPTIONAL MATCH (p)-[:MENTIONS]->(t:Technology) + OPTIONAL MATCH (p)-[:TAGGED_WITH]->(topic:Topic) + RETURN p, collect(DISTINCT t.name) AS products, + collect(DISTINCT topic.name) AS topics + ORDER BY p.score DESC + LIMIT 20 + parameters: + - name: query + type: string + description: Keyword or phrase to search for in post titles and bodies + + - name: get_product_pulse + description: Get community sentiment and activity for a specific product or technology + cypher: | + MATCH (tech:Technology {name: $name}) + OPTIONAL MATCH (p:Post)-[:MENTIONS]->(tech) + OPTIONAL MATCH (p)-[:IN_SUBREDDIT]->(s:Subreddit) + WITH tech, + count(DISTINCT p) AS post_count, + avg(p.sentiment_score) AS avg_sentiment, + collect(DISTINCT s.name)[..5] AS active_subreddits, + avg(p.score) AS avg_score + RETURN tech, post_count, avg_sentiment, avg_score, active_subreddits + parameters: + - name: name + type: string + description: Product or technology name + + - name: find_pain_points + description: Find community-reported pain points, optionally filtered by product + cypher: | + MATCH (p:Post)-[:HAS_PAIN_POINT]->(pp:PainPoint) + OPTIONAL MATCH (p)-[:MENTIONS]->(tech:Technology) + WHERE $product = '' OR tech.name = $product + RETURN pp.name AS pain_point, + count(DISTINCT p) AS frequency, + collect(DISTINCT tech.name)[..5] AS products, + avg(p.score) AS avg_post_score + ORDER BY frequency DESC + LIMIT 15 + parameters: + - name: product + type: string + description: Product name to filter by (empty string for all) + + - name: find_use_cases + description: Discover validated use cases mentioned by the community + cypher: | + MATCH (p:Post)-[:DEMONSTRATES]->(uc:UseCase) + OPTIONAL MATCH (p)-[:MENTIONS]->(tech:Technology) + WHERE $product = '' OR tech.name = $product + RETURN uc.name AS use_case, + count(DISTINCT p) AS frequency, + collect(DISTINCT tech.name)[..5] AS products, + avg(p.sentiment_score) AS avg_sentiment + ORDER BY frequency DESC + LIMIT 15 + parameters: + - name: product + type: string + description: Product name to filter by (empty string for all) + + - name: product_co_occurrence + description: Find products frequently discussed together + cypher: | + MATCH (t1:Technology {name: $name})-[r:CO_OCCURS_WITH]->(t2:Technology) + RETURN t2.name AS co_product, + coalesce(r.weight, 1) AS co_occurrences + ORDER BY co_occurrences DESC + LIMIT 15 + parameters: + - name: name + type: string + description: Product name to find co-occurrences for + + - name: top_contributors + description: Find the most active contributors in a subreddit + cypher: | + MATCH (u:Person)-[:ACTIVE_IN]->(s:Subreddit) + WHERE toLower(s.name) = toLower($subreddit) OR $subreddit = '' + OPTIONAL MATCH (u)-[:POSTED]->(p:Post)-[:IN_SUBREDDIT]->(s) + RETURN u.name AS contributor, + count(DISTINCT p) AS post_count, + avg(p.score) AS avg_score, + collect(DISTINCT s.name)[..3] AS active_communities + ORDER BY post_count DESC + LIMIT 20 + parameters: + - name: subreddit + type: string + description: Subreddit name to filter by (empty string for all) + + - name: unanswered_questions + description: Find questions with few or no replies + cypher: | + MATCH (p:Post {is_question: true}) + OPTIONAL MATCH (p)-[:IN_SUBREDDIT]->(s:Subreddit) + OPTIONAL MATCH (p)-[:MENTIONS]->(tech:Technology) + WHERE p.num_comments <= $max_comments + RETURN p.title AS question, + p.permalink AS link, + p.score AS score, + p.num_comments AS comment_count, + s.name AS subreddit, + collect(DISTINCT tech.name)[..3] AS products + ORDER BY p.created_utc DESC + LIMIT 20 + parameters: + - name: max_comments + type: integer + default: 2 + description: Maximum number of comments to consider "unanswered" + + - name: sentiment_by_subreddit + description: Compare sentiment across subreddits for a product + cypher: | + MATCH (p:Post)-[:MENTIONS]->(tech:Technology {name: $product}) + MATCH (p)-[:IN_SUBREDDIT]->(s:Subreddit) + WHERE p.sentiment_label IS NOT NULL + RETURN s.name AS subreddit, + count(p) AS post_count, + avg(p.sentiment_score) AS avg_sentiment, + sum(CASE WHEN p.sentiment_label = 'positive' THEN 1 ELSE 0 END) AS positive, + sum(CASE WHEN p.sentiment_label = 'negative' THEN 1 ELSE 0 END) AS negative, + sum(CASE WHEN p.sentiment_label = 'neutral' THEN 1 ELSE 0 END) AS neutral + ORDER BY post_count DESC + parameters: + - name: product + type: string + description: Product name to analyse sentiment for + + - name: trending_topics + description: Identify trending discussion topics across communities + cypher: | + MATCH (p:Post)-[:TAGGED_WITH]->(t:Topic) + OPTIONAL MATCH (p)-[:IN_SUBREDDIT]->(s:Subreddit) + WHERE toLower(s.name) = toLower($subreddit) OR $subreddit = '' + RETURN t.name AS topic, + count(DISTINCT p) AS post_count, + avg(p.score) AS avg_score, + collect(DISTINCT s.name)[..5] AS subreddits + ORDER BY post_count DESC + LIMIT 20 + parameters: + - name: subreddit + type: string + description: Filter by subreddit name (empty for all) + + - name: list_posts + description: "List Post records ordered by score" + cypher: | + MATCH (n:Post) + RETURN n + ORDER BY n.score DESC + LIMIT toInteger($limit) + parameters: + - name: limit + type: string + description: "Maximum number of results to return (default: 10)" + + - name: get_post_by_id + description: "Get a specific Post by post_id with all connections" + cypher: | + MATCH (n:Post {post_id: $id}) + OPTIONAL MATCH (n)-[r]-(related) + RETURN n, type(r) AS relationship, + labels(related) AS related_labels, + related.name AS related_name + LIMIT 50 + parameters: + - name: id + type: string + description: "The Reddit post_id to look up" + +system_prompt: | + You are a product intelligence analyst with access to a knowledge graph + built from Reddit community discussions. + + Your knowledge graph contains posts, comments, authors, subreddits, and + extracted knowledge including products, topics, pain points, and use cases. + + Your capabilities include: + - Analysing community sentiment and trending discussions around specific products + - Surfacing pain points and unmet needs reported by practitioners + - Discovering real-world use cases validated by the community + - Mapping product co-occurrence and competitive relationships + - Identifying unanswered questions and content opportunities + - Tracking community health and contributor engagement + + Always cite specific evidence from the graph. When surfacing pain points or + sentiment trends, quantify with post counts and scores where possible. + Focus on actionable insights for product, developer relations, and community teams. + +visualization: + node_colors: + Post: "#ff4500" + Comment: "#ff6534" + Subreddit: "#fb923c" + Person: "#22c55e" + Technology: "#0ea5e9" + Topic: "#8b5cf6" + PainPoint: "#ef4444" + UseCase: "#10b981" + Organization: "#3b82f6" + node_sizes: + Technology: 35 + Subreddit: 30 + Post: 20 + Person: 20 + Topic: 15 + PainPoint: 18 + UseCase: 18 + Comment: 10 + default_cypher: | + MATCH (t:Technology)-[r:CO_OCCURS_WITH]-(t2:Technology) + RETURN t, r, t2 + LIMIT 100 diff --git a/tests/test_connectors.py b/tests/test_connectors.py index 7529086..3b0bb85 100644 --- a/tests/test_connectors.py +++ b/tests/test_connectors.py @@ -78,10 +78,10 @@ def test_merge(self): class TestConnectorRegistry: def test_all_registered(self): - assert len(CONNECTOR_REGISTRY) == 13 + assert len(CONNECTOR_REGISTRY) == 14 def test_expected_connectors(self): - expected = {"github", "notion", "jira", "slack", "gmail", "gcal", "salesforce", "linear", "google-workspace", "claude-code", "claude-ai", "chatgpt", "local-file"} + expected = {"github", "notion", "jira", "slack", "gmail", "gcal", "salesforce", "linear", "google-workspace", "claude-code", "claude-ai", "chatgpt", "reddit", "local-file"} assert set(CONNECTOR_REGISTRY.keys()) == expected def test_get_connector(self): @@ -94,9 +94,10 @@ def test_get_unknown_raises(self): def test_list_connectors(self): result = list_connectors() - assert len(result) == 13 + assert len(result) == 14 ids = {c["id"] for c in result} assert "github" in ids + assert "reddit" in ids assert "local-file" in ids def test_all_have_credential_prompts(self): @@ -106,6 +107,159 @@ def test_all_have_credential_prompts(self): assert isinstance(prompts, list) +# --------------------------------------------------------------------------- +# Reddit connector +# --------------------------------------------------------------------------- + + +class TestRedditConnector: + def _make_connector(self, credentials: dict | None = None) -> object: + from create_context_graph.connectors.reddit_connector import RedditConnector + conn = RedditConnector() + conn.authenticate(credentials or {}) + return conn + + def test_service_metadata(self): + conn = get_connector("reddit") + assert conn.service_name == "Reddit" + assert conn.requires_oauth is False + + def test_credential_prompts(self): + conn = get_connector("reddit") + prompts = conn.get_credential_prompts() + names = {p["name"] for p in prompts} + # Core scraping options + assert "subreddits" in names + assert "keywords" in names + assert "fetch_comments" in names + assert "enrich_posts" in names + # Optional OAuth fields + assert "client_id" in names + assert "client_secret" in names + assert "reddit_username" in names + assert "reddit_password" in names + # reddit_password should be marked secret + secret_names = {p["name"] for p in prompts if p.get("secret")} + assert "reddit_password" in secret_names + + def test_authenticate_defaults(self): + conn = self._make_connector({}) + assert conn._subreddits == ["neo4j"] + assert conn._keywords == ["rag", "agentic ai", "knowledge graph"] + assert conn._max_pages == 1 + assert conn._fetch_comments is True + assert conn._enrich_posts is True + # Unauthenticated by default — no auth headers, public rate limits + assert not conn._auth_headers # None or empty dict — both are falsy + assert conn._anthropic_api_key is None + assert conn._openai_api_key is None + + def test_authenticate_custom_values(self): + conn = self._make_connector({ + "subreddits": "python, datascience", + "keywords": "llm, vector search", + "max_pages": "3", + "fetch_comments": "no", + "enrich_posts": "no", + }) + assert conn._subreddits == ["python", "datascience"] + assert conn._keywords == ["llm", "vector search"] + assert conn._max_pages == 3 + assert conn._fetch_comments is False + + def test_authenticate_oauth_sets_auth_headers(self): + """When OAuth creds are provided and token fetch succeeds, auth headers are set.""" + from create_context_graph.connectors.reddit_connector import _fetch_oauth_token + + with patch( + "create_context_graph.connectors.reddit_connector._fetch_oauth_token", + return_value="test_token_abc", + ): + conn = self._make_connector({ + "client_id": "myclientid", + "client_secret": "myclientsecret", + "reddit_username": "testuser", + "reddit_password": "testpass", + }) + + assert conn._auth_headers is not None + assert "Authorization" in conn._auth_headers + assert conn._auth_headers["Authorization"] == "bearer test_token_abc" + + def test_authenticate_oauth_falls_back_on_failure(self): + """When OAuth token fetch fails, connector falls back to unauthenticated mode.""" + with patch( + "create_context_graph.connectors.reddit_connector._fetch_oauth_token", + return_value=None, + ): + conn = self._make_connector({ + "client_id": "myclientid", + "client_secret": "myclientsecret", + "reddit_username": "testuser", + "reddit_password": "testpass", + }) + + assert not conn._auth_headers # None or empty dict — both are falsy, falls back to public API + + def test_fetch_returns_normalized_data(self, monkeypatch): + """fetch() returns NormalizedData with expected entity types when scrape returns posts.""" + from create_context_graph.connectors.reddit_connector import _get_json + + sample_listing = { + "data": { + "children": [ + { + "data": { + "id": "abc123", + "title": "How do I use RAG with Neo4j?", + "selftext": "Looking for guidance on setting up a RAG pipeline.", + "author": "testuser", + "subreddit": "neo4j", + "score": 42, + "upvote_ratio": 0.95, + "num_comments": 5, + "link_flair_text": "Question", + "permalink": "/r/neo4j/comments/abc123/", + "created_utc": 1700000000, + "is_self": True, + } + } + ], + "after": None, + } + } + + def mock_get_json(url, params=None, retries=3, **kwargs): + if "search.json" in url: + return sample_listing + return None + + monkeypatch.setattr( + "create_context_graph.connectors.reddit_connector._get_json", + mock_get_json, + ) + + conn = self._make_connector({ + "fetch_comments": "no", + "enrich_posts": "no", + "subreddits": "neo4j", + "keywords": "rag", + }) + result = conn.fetch() + + assert len(result.entities["Post"]) == 1 + assert len(result.entities["Person"]) == 1 + assert len(result.entities["Subreddit"]) == 1 + assert result.entities["Post"][0]["post_id"] == "abc123" + assert result.entities["Person"][0]["name"] == "testuser" + + rel_types = {r["type"] for r in result.relationships} + assert "POSTED" in rel_types + assert "IN_SUBREDDIT" in rel_types + + assert len(result.documents) == 1 + + # --------------------------------------------------------------------------- # Merge helper # ---------------------------------------------------------------------------