[PATCH v15 2/7] devtools: add common infrastructure for AI review scripts

[Stephen Hemminger <stephen@networkplumber.org>]

The AI review scripts share similar options and infrastructure.

Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
---
 devtools/ai/_common.py | 246 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 246 insertions(+)
 create mode 100644 devtools/ai/_common.py

diff --git a/devtools/ai/_common.py b/devtools/ai/_common.py
new file mode 100644
index 0000000000..69982cbda5
--- /dev/null
+++ b/devtools/ai/_common.py
@@ -0,0 +1,246 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2026 Stephen Hemminger
+
+"""Common utilities shared by the DPDK AI review scripts."""
+
+import argparse
+import json
+import subprocess
+import sys
+from dataclasses import dataclass
+from typing import Any, NoReturn
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+# Provider configurations (model defaults; override with --model).
+PROVIDERS: dict[str, dict[str, str]] = {
+    "anthropic": {
+        "name": "Claude",
+        "endpoint": "https://api.anthropic.com/v1/messages",
+        "default_model": "claude-sonnet-4-5-20250929",
+        "env_var": "ANTHROPIC_API_KEY",
+    },
+    "openai": {
+        "name": "ChatGPT",
+        "endpoint": "https://api.openai.com/v1/chat/completions",
+        "default_model": "gpt-4.1",
+        "env_var": "OPENAI_API_KEY",
+    },
+    "xai": {
+        "name": "Grok",
+        "endpoint": "https://api.x.ai/v1/chat/completions",
+        "default_model": "grok-4-1-fast-non-reasoning",
+        "env_var": "XAI_API_KEY",
+    },
+    "google": {
+        "name": "Gemini",
+        "endpoint": "https://generativelanguage.googleapis.com/v1beta/models",
+        "default_model": "gemini-3-flash-preview",
+        "env_var": "GOOGLE_API_KEY",
+    },
+}
+
+
+def error(msg: str) -> NoReturn:
+    """Print error message to stderr and exit with status 1."""
+    print(f"Error: {msg}", file=sys.stderr)
+    sys.exit(1)
+
+
+def get_git_config(key: str) -> str | None:
+    """Return the value of a git config key, or None if unset/git missing."""
+    try:
+        result = subprocess.run(
+            ["git", "config", "--get", key],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        return result.stdout.strip()
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        return None
+
+
+def list_providers() -> NoReturn:
+    """Print available providers and exit."""
+    print("Available AI Providers:\n")
+    print(f"{'Provider':<12} {'Default Model':<30} {'API Key Variable'}")
+    print(f"{'--------':<12} {'-------------':<30} {'----------------'}")
+    for name, config in PROVIDERS.items():
+        print(f"{name:<12} {config['default_model']:<30} {config['env_var']}")
+    sys.exit(0)
+
+
+@dataclass
+class TokenUsage:
+    """Accumulated token usage across API calls."""
+
+    input_tokens: int = 0
+    output_tokens: int = 0
+    cache_creation_tokens: int = 0
+    cache_read_tokens: int = 0
+    api_calls: int = 0
+
+    def add(self, other: "TokenUsage") -> None:
+        """Accumulate usage from another TokenUsage."""
+        self.input_tokens += other.input_tokens
+        self.output_tokens += other.output_tokens
+        self.cache_creation_tokens += other.cache_creation_tokens
+        self.cache_read_tokens += other.cache_read_tokens
+        self.api_calls += other.api_calls
+
+
+def format_token_summary(usage: TokenUsage, provider: str, model: str) -> str:
+    """Format a token usage summary string."""
+    provider_label = PROVIDERS.get(provider, {}).get("name", provider)
+    lines = ["=== Token Usage Summary ==="]
+    lines.append(f"Provider:      {provider_label} ({model})")
+    lines.append(f"API calls:     {usage.api_calls}")
+    lines.append(f"Input tokens:  {usage.input_tokens:,}")
+    lines.append(f"Output tokens: {usage.output_tokens:,}")
+    if usage.cache_creation_tokens:
+        lines.append(f"Cache write:   {usage.cache_creation_tokens:,}")
+    if usage.cache_read_tokens:
+        lines.append(f"Cache read:    {usage.cache_read_tokens:,}")
+    total = usage.input_tokens + usage.output_tokens
+    lines.append(f"Total tokens:  {total:,}")
+    lines.append("=" * 27)
+    return "\n".join(lines)
+
+
+def add_token_args(parser: argparse.ArgumentParser) -> None:
+    """Add the --show-tokens flag to an ArgumentParser."""
+    parser.add_argument(
+        "--show-tokens",
+        action="store_true",
+        help="Show token usage summary on stderr after the run",
+    )
+
+
+def print_token_summary(
+    usage: TokenUsage, provider: str, model: str, show: bool
+) -> None:
+    """Print token usage summary to stderr if requested and any calls were made."""
+    if not show or usage.api_calls == 0:
+        return
+    print("", file=sys.stderr)
+    print(format_token_summary(usage, provider, model), file=sys.stderr)
+
+
+def _build_request_meta(
+    provider: str, api_key: str, model: str
+) -> tuple[str, dict[str, str]]:
+    """Return (url, headers) for a provider request."""
+    config = PROVIDERS[provider]
+    if provider == "anthropic":
+        return config["endpoint"], {
+            "Content-Type": "application/json",
+            "x-api-key": api_key,
+            "anthropic-version": "2023-06-01",
+        }
+    if provider == "google":
+        url = f"{config['endpoint']}/{model}:generateContent?key={api_key}"
+        return url, {"Content-Type": "application/json"}
+    # openai, xai
+    return config["endpoint"], {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {api_key}",
+    }
+
+
+def _extract_usage(provider: str, result: dict[str, Any]) -> TokenUsage:
+    """Extract TokenUsage from a provider response."""
+    usage = TokenUsage(api_calls=1)
+    if provider == "anthropic":
+        raw = result.get("usage", {})
+        usage.input_tokens = raw.get("input_tokens", 0)
+        usage.output_tokens = raw.get("output_tokens", 0)
+        usage.cache_creation_tokens = raw.get("cache_creation_input_tokens", 0)
+        usage.cache_read_tokens = raw.get("cache_read_input_tokens", 0)
+    elif provider == "google":
+        raw = result.get("usageMetadata", {})
+        usage.input_tokens = raw.get("promptTokenCount", 0)
+        usage.output_tokens = raw.get("candidatesTokenCount", 0)
+    else:  # openai, xai
+        raw = result.get("usage", {})
+        usage.input_tokens = raw.get("prompt_tokens", 0)
+        usage.output_tokens = raw.get("completion_tokens", 0)
+        cache_details = raw.get("prompt_tokens_details", {})
+        if cache_details:
+            usage.cache_read_tokens = cache_details.get("cached_tokens", 0)
+    return usage
+
+
+def _extract_text(provider: str, result: dict[str, Any]) -> str:
+    """Extract response text from a provider response. Calls error() on failure."""
+    if "error" in result:
+        error(f"API error: {result['error'].get('message', result)}")
+    if provider == "anthropic":
+        content = result.get("content", [])
+        return "".join(
+            block.get("text", "") for block in content if block.get("type") == "text"
+        )
+    if provider == "google":
+        candidates = result.get("candidates", [])
+        if not candidates:
+            error("No response from Gemini")
+        parts = candidates[0].get("content", {}).get("parts", [])
+        return "".join(part.get("text", "") for part in parts)
+    # openai, xai
+    choices = result.get("choices", [])
+    if not choices:
+        error("No response from API")
+    return choices[0].get("message", {}).get("content", "")
+
+
+def _print_verbose_usage(usage: TokenUsage) -> None:
+    """Print per-call token details to stderr."""
+    print("=== Token Usage ===", file=sys.stderr)
+    print(f"Input tokens: {usage.input_tokens:,}", file=sys.stderr)
+    print(f"Output tokens: {usage.output_tokens:,}", file=sys.stderr)
+    if usage.cache_creation_tokens:
+        print(f"Cache creation: {usage.cache_creation_tokens:,}", file=sys.stderr)
+    if usage.cache_read_tokens:
+        print(f"Cache read: {usage.cache_read_tokens:,}", file=sys.stderr)
+    print("===================", file=sys.stderr)
+
+
+def send_request(
+    provider: str,
+    api_key: str,
+    model: str,
+    request_data: dict[str, Any],
+    *,
+    timeout: int = 120,
+    verbose: bool = False,
+) -> tuple[str, TokenUsage]:
+    """Send a prebuilt request to a provider and return (response_text, usage).
+
+    The caller assembles the provider-specific request body via its own
+    build_*_request helpers (the prompts differ per script). This function
+    handles transport, error reporting, and token-usage extraction.
+    """
+    url, headers = _build_request_meta(provider, api_key, model)
+    body = json.dumps(request_data).encode("utf-8")
+    req = Request(url, data=body, headers=headers)
+
+    try:
+        with urlopen(req, timeout=timeout) as response:
+            result = json.loads(response.read().decode("utf-8"))
+    except HTTPError as e:
+        error_body = e.read().decode("utf-8")
+        try:
+            error_data = json.loads(error_body)
+            error(f"API error: {error_data.get('error', error_body)}")
+        except json.JSONDecodeError:
+            error(f"API error ({e.code}): {error_body}")
+    except URLError as e:
+        if isinstance(e.reason, TimeoutError):
+            error(f"Request timed out after {timeout} seconds")
+        error(f"Connection error: {e.reason}")
+
+    usage = _extract_usage(provider, result)
+    if verbose:
+        _print_verbose_usage(usage)
+    return _extract_text(provider, result), usage
-- 
2.53.0