silo-calc/pythonpath/silo_calc/ai_client.py

"""OpenRouter AI client for the Silo Calc extension.

Provides AI-powered text generation via the OpenRouter API
(https://openrouter.ai/api/v1/chat/completions).  Uses stdlib urllib
only -- no external dependencies.

The core ``chat_completion()`` function is generic and reusable for
future features (price analysis, sourcing assistance).  Domain helpers
like ``generate_description()`` build on top of it.
"""

import json
import os
import ssl
import urllib.error
import urllib.request
from typing import Any, Dict, List, Optional

from . import settings as _settings

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

OPENROUTER_API_URL = "https://openrouter.ai/api/v1/chat/completions"

DEFAULT_MODEL = "openai/gpt-4.1-nano"

DEFAULT_INSTRUCTIONS = (
    "You are a parts librarian for an engineering company. "
    "Given a seller's product description, produce a concise, standardized "
    "part description suitable for a Bill of Materials. Rules:\n"
    "- Maximum 60 characters\n"
    "- Use title case\n"
    "- Start with the component type (e.g., Bolt, Resistor, Bearing)\n"
    "- Include key specifications (size, rating, material) in order of importance\n"
    "- Omit brand names, marketing language, and redundant words\n"
    "- Use standard engineering abbreviations (SS, Al, M3, 1/4-20)\n"
    "- Output ONLY the description, no quotes or explanation"
)

# ---------------------------------------------------------------------------
# SSL helper (same pattern as client.py)
# ---------------------------------------------------------------------------


def _get_ssl_context() -> ssl.SSLContext:
    """Build an SSL context for OpenRouter API calls."""
    ctx = ssl.create_default_context()
    for ca_path in (
        "/etc/ssl/certs/ca-certificates.crt",
        "/etc/pki/tls/certs/ca-bundle.crt",
    ):
        if os.path.isfile(ca_path):
            try:
                ctx.load_verify_locations(ca_path)
            except Exception:
                pass
            break
    return ctx


# ---------------------------------------------------------------------------
# Settings resolution helpers
# ---------------------------------------------------------------------------


def _get_api_key() -> str:
    """Resolve the OpenRouter API key from settings or environment."""
    cfg = _settings.load()
    key = cfg.get("openrouter_api_key", "")
    if not key:
        key = os.environ.get("OPENROUTER_API_KEY", "")
    return key


def _get_model() -> str:
    """Resolve the model slug from settings or default."""
    cfg = _settings.load()
    return cfg.get("openrouter_model", "") or DEFAULT_MODEL


def _get_instructions() -> str:
    """Resolve the system instructions from settings or default."""
    cfg = _settings.load()
    return cfg.get("openrouter_instructions", "") or DEFAULT_INSTRUCTIONS


# ---------------------------------------------------------------------------
# Core API function
# ---------------------------------------------------------------------------


def chat_completion(
    messages: List[Dict[str, str]],
    model: Optional[str] = None,
    temperature: float = 0.3,
    max_tokens: int = 200,
) -> str:
    """Send a chat completion request to OpenRouter.

    Parameters
    ----------
    messages : list of {"role": str, "content": str}
    model : model slug (default: from settings or DEFAULT_MODEL)
    temperature : sampling temperature
    max_tokens : maximum response tokens

    Returns
    -------
    str : the assistant's response text

    Raises
    ------
    RuntimeError : on missing API key, HTTP errors, network errors,
                   or unexpected response format.
    """
    api_key = _get_api_key()
    if not api_key:
        raise RuntimeError(
            "OpenRouter API key not configured. "
            "Set it in Settings or the OPENROUTER_API_KEY environment variable."
        )

    model = model or _get_model()

    payload = {
        "model": model,
        "messages": messages,
        "temperature": temperature,
        "max_tokens": max_tokens,
    }

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "HTTP-Referer": "https://github.com/kindredsystems/silo",
        "X-Title": "Silo Calc Extension",
    }

    body = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        OPENROUTER_API_URL, data=body, headers=headers, method="POST"
    )

    try:
        with urllib.request.urlopen(
            req, context=_get_ssl_context(), timeout=30
        ) as resp:
            result = json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        error_body = e.read().decode("utf-8", errors="replace")
        if e.code == 401:
            raise RuntimeError("OpenRouter API key is invalid or expired.")
        if e.code == 402:
            raise RuntimeError("OpenRouter account has insufficient credits.")
        if e.code == 429:
            raise RuntimeError("OpenRouter rate limit exceeded. Try again shortly.")
        raise RuntimeError(f"OpenRouter API error {e.code}: {error_body}")
    except urllib.error.URLError as e:
        raise RuntimeError(f"Network error contacting OpenRouter: {e.reason}")

    choices = result.get("choices", [])
    if not choices:
        raise RuntimeError("OpenRouter returned an empty response.")

    return choices[0].get("message", {}).get("content", "").strip()


# ---------------------------------------------------------------------------
# Domain helpers
# ---------------------------------------------------------------------------


def generate_description(
    seller_description: str,
    category: str = "",
    existing_description: str = "",
    part_number: str = "",
) -> str:
    """Generate a standardized part description from a seller description.

    Parameters
    ----------
    seller_description : the raw seller/vendor description text
    category : category code (e.g. "F01") for context
    existing_description : current description in col E, if any
    part_number : the part number, for context

    Returns
    -------
    str : the AI-generated standardized description
    """
    system_prompt = _get_instructions()

    user_parts = []
    if category:
        user_parts.append(f"Category: {category}")
    if part_number:
        user_parts.append(f"Part Number: {part_number}")
    if existing_description:
        user_parts.append(f"Current Description: {existing_description}")
    user_parts.append(f"Seller Description: {seller_description}")

    user_prompt = "\n".join(user_parts)

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt},
    ]

    return chat_completion(messages)


def is_configured() -> bool:
    """Return True if the OpenRouter API key is available."""
    return bool(_get_api_key())