backend/tools/github_search.py

"""
Real GitHub repository + code search via GitHub REST API.
No auth required for public repos (60 req/hr).
Set GH_TOKEN or GITHUB_TOKEN env var for 5000 req/hr.
"""
from __future__ import annotations

import logging
import os
import time
from typing import Any

import httpx

logger = logging.getLogger("dolor3v.tools.github_search")

_GITHUB_API = "https://api.github.com"
_HEADERS_BASE = {
    "Accept": "application/vnd.github+json",
    "X-GitHub-Api-Version": "2022-11-28",
    "User-Agent": "DOLOR3V-TravelerDev/1.0",
}


def _auth_headers() -> dict[str, str]:
    token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN", "")
    if token:
        return {**_HEADERS_BASE, "Authorization": f"Bearer {token}"}
    return dict(_HEADERS_BASE)


async def search_repositories(
    query: str,
    sort: str = "stars",
    order: str = "desc",
    per_page: int = 10,
) -> dict[str, Any]:
    """Search GitHub repositories. Returns real API data."""
    url = f"{_GITHUB_API}/search/repositories"
    params = {"q": query, "sort": sort, "order": order, "per_page": per_page}

    t0 = time.monotonic()
    async with httpx.AsyncClient(timeout=15.0) as client:
        resp = await client.get(url, params=params, headers=_auth_headers())

    latency_ms = round((time.monotonic() - t0) * 1000)

    if resp.status_code == 403:
        remaining = resp.headers.get("X-RateLimit-Remaining", "?")
        reset = resp.headers.get("X-RateLimit-Reset", "?")
        return {
            "error": "GitHub rate limit exceeded",
            "remaining": remaining,
            "reset_at": reset,
            "tip": "Set GH_TOKEN or GITHUB_TOKEN env var for 5000 req/hr",
        }

    if resp.status_code != 200:
        return {
            "error": f"GitHub API returned {resp.status_code}",
            "body": resp.text[:500],
        }

    data = resp.json()
    items = data.get("items", [])

    results = []
    for item in items:
        results.append({
            "name": item.get("full_name"),
            "description": item.get("description"),
            "stars": item.get("stargazers_count"),
            "forks": item.get("forks_count"),
            "language": item.get("language"),
            "url": item.get("html_url"),
            "clone_url": item.get("clone_url"),
            "topics": item.get("topics", []),
            "updated_at": item.get("updated_at"),
            "open_issues": item.get("open_issues_count"),
        })

    return {
        "query": query,
        "total_count": data.get("total_count", 0),
        "returned": len(results),
        "latency_ms": latency_ms,
        "results": results,
        "rate_limit_remaining": resp.headers.get("X-RateLimit-Remaining", "unknown"),
    }


async def search_code(
    query: str,
    per_page: int = 10,
) -> dict[str, Any]:
    """Search GitHub code. Requires GH_TOKEN for reliable access."""
    url = f"{_GITHUB_API}/search/code"
    params = {"q": query, "per_page": per_page}

    token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN", "")
    if not token:
        return {
            "error": "Code search requires authentication",
            "tip": "Set GH_TOKEN env var",
        }

    t0 = time.monotonic()
    async with httpx.AsyncClient(timeout=15.0) as client:
        resp = await client.get(url, params=params, headers=_auth_headers())

    latency_ms = round((time.monotonic() - t0) * 1000)

    if resp.status_code != 200:
        return {"error": f"GitHub code search returned {resp.status_code}", "body": resp.text[:500]}

    data = resp.json()
    items = data.get("items", [])

    return {
        "query": query,
        "total_count": data.get("total_count", 0),
        "latency_ms": latency_ms,
        "results": [
            {
                "name": i.get("name"),
                "path": i.get("path"),
                "repo": i.get("repository", {}).get("full_name"),
                "url": i.get("html_url"),
                "sha": i.get("sha"),
            }
            for i in items
        ],
    }