pryscraper/graphql_discovery.py

"""Pry — GraphQL Auto-Discovery.
Detects GraphQL endpoints, runs introspection queries, generates optimized queries.
Many modern sites (Shopify, GitHub, Twitter/X, etc.) have GraphQL APIs that are
10-100x more efficient than scraping HTML."""

# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Rug Munch Media LLC
#
# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
# Licensed under MIT. See LICENSE.

import json
import logging
import re
from typing import Any, ClassVar

logger = logging.getLogger(__name__)


class GraphQLDiscovery:
    """Auto-discover and query GraphQL endpoints."""

    # Common paths where GraphQL endpoints live
    COMMON_PATHS: ClassVar[list[str]] = [
        "/graphql", "/api/graphql", "/api/v1/graphql", "/v1/graphql", "/v2/graphql",
        "/graphql/v1", "/graphql/v2", "/gql", "/api/gql", "/query", "/api/query",
        "/__graphql", "/altair", "/playground",
    ]

    # Patterns in JS bundles that indicate GraphQL endpoints
    ENDPOINT_PATTERNS: ClassVar[list[str]] = [
        r'["\']([/][\w/]*graphql[\w/]*)["\']',
        r'["\'](https?://[^/"\']*[/][\w/]*graphql[\w/]*)["\']',
        r'apolloClient\s*\.\s*link\s*\(\s*["\']([^"\']+)["\']',
        r'createHttpLink\s*\(\s*{\s*uri:\s*["\']([^"\']+)["\']',
        r'endpoint["\']:\s*["\']([^"\']+graphql[^"\']*)["\']',
        r'uri["\']:\s*["\']([^"\']*graphql[^"\']*)["\']',
    ]

    INTROSPECTION_QUERY = """
    # Standard graphql introspection query
    query IntrospectionQuery {
        __schema {
            queryType { name }
            mutationType { name }
            subscriptionType { name }
            types {
                kind name
                fields { name type { name kind ofType { name kind ofType { name } } } }
            }
        }
    }
    """

    def __init__(self) -> None:
        self.discovered: dict[str, dict[str, Any]] = {}

    async def discover(self, base_url: str) -> list[dict[str, Any]]:
        """Discover GraphQL endpoints for a given base URL."""
        from client import get_client

        client = await get_client()
        found: list[dict[str, Any]] = []
        for path in self.COMMON_PATHS:
            url = base_url.rstrip("/") + path
            try:
                resp = await client.post(
                    url, json={"query": "{ __typename }"}, timeout=10
                )
                if resp.is_success:
                    try:
                        data = resp.json()
                    except (json.JSONDecodeError, ValueError):
                        continue
                    if "data" in data or "errors" in data:
                        found.append({"url": url, "method": "path_probe"})
                        self.discovered[url] = data
            except (httpx.HTTPError, httpx.RequestError) as e:
                logger.debug("graphql_probe_failed", extra={"url": url, "err": str(e)[:100]})
        return found

    async def introspect(self, endpoint: str) -> dict[str, Any]:
        """Run GraphQL introspection to get the full schema."""
        from client import get_client

        client = await get_client()
        try:
            resp = await client.post(
                endpoint, json={"query": self.INTROSPECTION_QUERY}, timeout=30
            )
            if resp.is_success:
                return resp.json()
        except (httpx.HTTPError, httpx.RequestError) as e:
            return {"error": str(e)[:300]}
        return {}

    async def query(
        self, endpoint: str, query: str, variables: dict[str, Any] | None = None
    ) -> dict[str, Any]:
        """Execute a GraphQL query."""
        from client import get_client

        client = await get_client()
        try:
            resp = await client.post(
                endpoint,
                json={"query": query, "variables": variables or {}},
                timeout=30,
            )
            if resp.is_success:
                return resp.json()
        except (httpx.HTTPError, httpx.RequestError) as e:
            return {"error": str(e)[:300]}
        return {}

    def extract_endpoints_from_js(self, js_content: str) -> list[str]:
        """Scan a JS bundle for embedded GraphQL endpoint strings.

        Returns a de-duplicated list of potential endpoint URLs/paths.
        """
        candidates: list[str] = []
        for pattern in self.ENDPOINT_PATTERNS:
            for match in re.finditer(pattern, js_content):
                if match.lastindex is None:
                    continue
                value = match.group(1).strip()
                if value:
                    candidates.append(value)
        # De-duplicate while preserving order
        seen: set[str] = set()
        unique: list[str] = []
        for c in candidates:
            if c not in seen:
                seen.add(c)
                unique.append(c)
        return unique