devpulse.py

#!/usr/bin/env python3
"""DevPulse: Developer community research tool for LayerLens.

Searches Reddit, Hacker News, and GitHub Discussions for what developers
are saying about any topic in the last 30 days.

Usage:
    python3 devpulse.py "AI evaluation tools"
    python3 devpulse.py "LLM benchmarking" --depth deep
    python3 devpulse.py "LayerLens" --days 14
    python3 devpulse.py "agent testing" --sources reddit,hackernews
    python3 devpulse.py "model evaluation" --output json
"""

import argparse
import json
import os
import sys
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

# Add lib to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from lib.reddit import search_reddit
from lib.hackernews import search_hackernews
from lib.github_discussions import search_issues_and_discussions


def get_config() -> Dict[str, Optional[str]]:
    """Load config from environment or .env file."""
    config = {
        "SCRAPECREATORS_API_KEY": os.environ.get("SCRAPECREATORS_API_KEY"),
        "GITHUB_TOKEN": os.environ.get("GITHUB_TOKEN"),
        "REDDIT_CLIENT_ID": os.environ.get("REDDIT_CLIENT_ID"),
        "REDDIT_CLIENT_SECRET": os.environ.get("REDDIT_CLIENT_SECRET"),
    }

    # Check for .env file
    env_paths = [
        os.path.expanduser("~/.config/devpulse/.env"),
        os.path.join(os.path.dirname(__file__), ".env"),
    ]
    for path in env_paths:
        if os.path.exists(path):
            with open(path) as f:
                for line in f:
                    line = line.strip()
                    if not line or line.startswith("#"):
                        continue
                    if "=" in line:
                        key, _, value = line.partition("=")
                        key = key.strip()
                        value = value.strip().strip("'\"")
                        if key in config and not config[key] and value:
                            config[key] = value
            break

    return config


def run_research(
    topic: str,
    days: int = 30,
    depth: str = "default",
    sources: Optional[List[str]] = None,
    output_format: str = "text",
) -> Dict[str, Any]:
    """Run multi-source research on a topic.

    Args:
        topic: What to search for
        days: How many days back to look (default 30)
        depth: quick, default, or deep
        sources: Which sources to use (reddit, hackernews, github). Default: all.
        output_format: text or json

    Returns:
        Dict with results from each source
    """
    config = get_config()
    sources = sources or ["reddit", "hackernews", "github"]

    to_date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    from_date = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d")

    results = {
        "topic": topic,
        "from_date": from_date,
        "to_date": to_date,
        "depth": depth,
        "sources_used": [],
        "reddit": [],
        "hackernews": [],
        "github": [],
    }

    # Reddit
    if "reddit" in sources:
        sc_token = config.get("SCRAPECREATORS_API_KEY")
        reddit_client_id = config.get("REDDIT_CLIENT_ID")
        reddit_client_secret = config.get("REDDIT_CLIENT_SECRET")
        if sc_token:
            backend = "ScrapeCreators"
        elif reddit_client_id and reddit_client_secret:
            backend = "OAuth (authenticated)"
        else:
            backend = "free JSON"
        sys.stderr.write(f"\n{'='*60}\n")
        sys.stderr.write(f"REDDIT (backend: {backend})\n")
        sys.stderr.write(f"{'='*60}\n")
        reddit_results = search_reddit(
            topic, from_date, to_date, depth=depth, sc_token=sc_token,
            reddit_client_id=reddit_client_id,
            reddit_client_secret=reddit_client_secret,
        )
        results["reddit"] = reddit_results
        results["sources_used"].append(f"reddit ({backend})")

    # Hacker News
    if "hackernews" in sources:
        sys.stderr.write(f"\n{'='*60}\n")
        sys.stderr.write(f"HACKER NEWS (Algolia API, free)\n")
        sys.stderr.write(f"{'='*60}\n")
        hn_results = search_hackernews(topic, from_date, to_date, depth=depth)
        results["hackernews"] = hn_results
        results["sources_used"].append("hackernews (Algolia)")

    # GitHub
    if "github" in sources:
        gh_token = config.get("GITHUB_TOKEN")
        sys.stderr.write(f"\n{'='*60}\n")
        sys.stderr.write(f"GITHUB ({'authenticated' if gh_token else 'unauthenticated'})\n")
        sys.stderr.write(f"{'='*60}\n")
        gh_results = search_issues_and_discussions(
            topic, from_date, to_date, depth=depth, gh_token=gh_token
        )
        results["github"] = gh_results
        results["sources_used"].append("github")

    return results


def format_text_report(results: Dict[str, Any]) -> str:
    """Format results as a readable text report."""
    lines = []
    topic = results["topic"]
    from_d = results["from_date"]
    to_d = results["to_date"]

    lines.append(f"# DevPulse Research: {topic}")
    lines.append(f"Period: {from_d} to {to_d}")
    lines.append(f"Sources: {', '.join(results['sources_used'])}")
    lines.append("")

    # Reddit
    reddit = results.get("reddit", [])
    if reddit:
        lines.append(f"## Reddit ({len(reddit)} posts)")
        lines.append("")
        for item in reddit[:15]:
            score = item.get("score", 0)
            comments = item.get("num_comments", 0)
            sub = item.get("subreddit", "")
            title = item.get("title", "")
            url = item.get("url", "")
            date = item.get("date", "")

            lines.append(f"### [{title}]({url})")
            lines.append(f"r/{sub} | {score} upvotes | {comments} comments | {date}")

            selftext = item.get("selftext", "").strip()
            if selftext:
                lines.append(f"> {selftext[:200]}...")

            top_comments = item.get("top_comments", [])
            if top_comments:
                lines.append("")
                lines.append("**Top comments:**")
                for c in top_comments[:3]:
                    author = c.get("author", "")
                    cscore = c.get("score", 0)
                    excerpt = c.get("excerpt", c.get("text", ""))[:200]
                    lines.append(f"- u/{author} ({cscore} pts): {excerpt}")
            lines.append("")

    # Hacker News
    hn = results.get("hackernews", [])
    if hn:
        lines.append(f"## Hacker News ({len(hn)} stories)")
        lines.append("")
        for item in hn[:15]:
            points = item.get("score", 0)
            comments = item.get("num_comments", 0)
            title = item.get("title", "")
            hn_url = item.get("hn_url", "")
            article_url = item.get("url", "")
            date = item.get("date", "")

            lines.append(f"### [{title}]({hn_url})")
            lines.append(f"{points} points | {comments} comments | {date}")
            if article_url:
                lines.append(f"Article: {article_url}")

            top_comments = item.get("top_comments", [])
            if top_comments:
                lines.append("")
                lines.append("**Top comments:**")
                for c in top_comments[:3]:
                    author = c.get("author", "")
                    text = c.get("text", "")[:200]
                    lines.append(f"- {author}: {text}")
            lines.append("")

    # GitHub
    gh = results.get("github", [])
    if gh:
        lines.append(f"## GitHub ({len(gh)} results)")
        lines.append("")

        # Split by type
        repos = [i for i in gh if i.get("type") == "repo"]
        issues = [i for i in gh if i.get("type") in ("issue", "pr")]

        if repos:
            lines.append("### Trending Repos")
            for item in repos[:10]:
                title = item.get("title", "").replace("[Repo] ", "")
                url = item.get("url", "")
                stars = item.get("stars", item.get("score", 0))
                desc = item.get("body_preview", "")
                lines.append(f"- **[{title}]({url})** ({stars} stars): {desc[:150]}")
            lines.append("")

        if issues:
            lines.append("### Issues & Discussions")
            for item in issues[:10]:
                title = item.get("title", "")
                url = item.get("url", "")
                repo = item.get("repo", "")
                score = item.get("score", 0)
                comments = item.get("num_comments", 0)
                date = item.get("date", "")
                labels = ", ".join(item.get("labels", [])[:3])
                label_str = f" [{labels}]" if labels else ""

                lines.append(f"- **[{title}]({url})**")
                lines.append(f"  {repo} | {score} reactions | {comments} comments | {date}{label_str}")
            lines.append("")

    # Summary stats
    total = len(reddit) + len(hn) + len(gh)
    lines.append(f"---")
    lines.append(f"Total results: {total} (Reddit: {len(reddit)}, HN: {len(hn)}, GitHub: {len(gh)})")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="DevPulse: Search developer communities for recent discussions"
    )
    parser.add_argument("topic", help="Topic to research")
    parser.add_argument("--days", type=int, default=30, help="Days to look back (default: 30)")
    parser.add_argument("--depth", choices=["quick", "default", "deep"], default="default",
                        help="Research depth (default: default)")
    parser.add_argument("--sources", default="reddit,hackernews,github",
                        help="Comma-separated sources (default: reddit,hackernews,github)")
    parser.add_argument("--output", choices=["text", "json"], default="text",
                        help="Output format (default: text)")

    args = parser.parse_args()
    sources = [s.strip() for s in args.sources.split(",")]

    results = run_research(
        topic=args.topic,
        days=args.days,
        depth=args.depth,
        sources=sources,
        output_format=args.output,
    )

    if args.output == "json":
        print(json.dumps(results, indent=2, default=str))
    else:
        print(format_text_report(results))


if __name__ == "__main__":
    main()