""" Minimal OpenRouter REST client (sync). Reads OPENROUTER_API_KEY / OPENROUTER_MODEL / OPENROUTER_BASE_URL from env. """ from __future__ import annotations import json import logging import os from typing import Any, Dict, List, Mapping, Optional import httpx _logger = logging.getLogger("shinkan.openrouter") _SKIP_ANTHROPIC_BLOCK_TYPES = frozenset( { "thinking", "redacted_thinking", "reasoning", "tool_use", "tool_calls", } ) def _shinkan_ai_debug() -> bool: return os.getenv("SHINKAN_AI_DEBUG", "").strip().lower() in ("1", "true", "yes", "full") def _coerce_nested_text(val: Any) -> str: if val is None: return "" if isinstance(val, str): return val.strip() if isinstance(val, bool) or isinstance(val, (int, float)): return str(val).strip() if isinstance(val, list): return "".join(_coerce_nested_text(x) for x in val).strip() if isinstance(val, dict): # OpenRouter/Anthropic: verschachtelte text/content-Hüllen for key in ("text", "content", "value"): if key in val: nested = _coerce_nested_text(val.get(key)) if nested: return nested return "" return str(val).strip() def _flatten_message_content(content: Any) -> str: """ Chat-Completion: `content` als String oder als Liste strukturierter Blöcke (Anthropic Claude über OpenRouter/Bedrock, teils verschachtelt). """ if content is None: return "" if isinstance(content, str): return content.strip() if isinstance(content, list): parts: List[str] = [] for block in content: if isinstance(block, str): bits = _coerce_nested_text(block) if bits: parts.append(bits) elif isinstance(block, dict): t_raw = block.get("type") ts = str(t_raw or "").strip().lower() if ts and (ts in _SKIP_ANTHROPIC_BLOCK_TYPES or ts.endswith("_thinking")): continue txt = None if ts in ("text", "output_text", ""): txt = block.get("text") if txt is None: txt = block.get("content") else: # unbekannten Typ weiter versuchen (Provider-Varianten), aber tool-use überspringen low = ts if "tool_use" in low or low.startswith("tool_"): continue txt = block.get("text") if block.get("text") is not None else block.get("content") bits = _coerce_nested_text(txt) if bits: parts.append(bits) return "".join(parts).strip() if isinstance(content, dict): return _coerce_nested_text(content) return str(content).strip() class OpenRouterError(Exception): """Upstream or transport failure.""" def openrouter_chat_completion( *, api_key: str, model: str, user_content: str, system_content: Optional[str] = None, timeout_sec: float = 120.0, temperature: float = 0.25, site_url: Optional[str] = None, app_title: Optional[str] = None, ) -> str: """ Returns assistant message content (plain string). Caller validates empty responses. """ base = (os.getenv("OPENROUTER_BASE_URL") or "").strip().rstrip("/") or "https://openrouter.ai/api/v1" url = f"{base}/chat/completions" headers: Dict[str, str] = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } referer = (site_url or os.getenv("APP_URL") or "").strip() if referer: headers["HTTP-Referer"] = referer title = (app_title or os.getenv("OPENROUTER_APP_TITLE") or "Shinkan Jinkendo").strip() if title: headers["X-Title"] = title messages: List[Dict[str, str]] = [] if system_content and str(system_content).strip(): messages.append({"role": "system", "content": str(system_content).strip()}) messages.append({"role": "user", "content": user_content}) payload: Dict[str, Any] = { "model": model, "messages": messages, "temperature": temperature, } try: with httpx.Client(timeout=timeout_sec) as client: resp = client.post(url, headers=headers, json=payload) except httpx.RequestError as e: raise OpenRouterError(str(e)) from e if resp.status_code >= 400: detail = "" try: j = resp.json() detail = ( str(j.get("error", {}).get("message")) if isinstance(j.get("error"), dict) else str(j.get("message") or j) ) except Exception: detail = (resp.text or "")[:600] raise OpenRouterError(f"HTTP {resp.status_code}: {detail}".strip()) try: data = resp.json() except json.JSONDecodeError as e: raise OpenRouterError("Ungueltige JSON-Antwort von OpenRouter") from e choices = data.get("choices") if isinstance(data, dict) else None if not choices or not isinstance(choices, list): raise OpenRouterError("OpenRouter: keine choices in Antwort") msg0 = choices[0] if choices else {} inner = msg0.get("message") if isinstance(msg0, dict) else None blobs: List[Any] = [] if isinstance(inner, dict): if inner.get("content") is not None: blobs.append(inner.get("content")) if inner.get("refusal") is not None: blobs.append(inner.get("refusal")) elif isinstance(inner, str): blobs.append(inner) if isinstance(msg0, dict) and msg0.get("content") is not None and msg0.get("content") not in blobs: blobs.append(msg0.get("content")) pieces = [_flatten_message_content(b).strip() for b in blobs if b is not None] joined = ("\n".join(p for p in pieces if p)).strip() if _shinkan_ai_debug(): fr = str(msg0.get("finish_reason") or "") if isinstance(msg0, dict) else "" fu = data.get("usage") if isinstance(data.get("usage"), dict) else {} pu = str(fu.get("prompt_tokens") or "") pc = str(fu.get("completion_tokens") or "") pt = str(fu.get("total_tokens") or "") raw_cls = type(blobs[0]).__name__ if blobs else "none" cc = str(len(joined)) _logger.warning( "[AI_DEBUG/openrouter] model=%s finish_reason=%s usage_prompt=%s usage_completion=%s usage_total=%s " "raw_content_cls=%s out_chars=%s", model, fr, pu, pc, pt, raw_cls, cc, ) try: from planning_llm_usage import record_planning_llm_call record_planning_llm_call(1) except Exception: pass return joined def normalize_openrouter_env() -> tuple[str, str]: key = (os.getenv("OPENROUTER_API_KEY") or "").strip() model = (os.getenv("OPENROUTER_MODEL") or "anthropic/claude-sonnet-4").strip() return key, model def default_openrouter_model_id() -> str: """Standard-Modell aus OPENROUTER_MODEL (ohne API-Key zu pruefen).""" _, model = normalize_openrouter_env() return model def effective_openrouter_model_for_prompt_row(row: Optional[Mapping[str, Any]]) -> str: """ Pro-Prompt-Override in ai_prompts.openrouter_model, sonst Env-Default. `row` kann eine partial Row aus load_ai_prompt_row sein (Felder slug, openrouter_model, …). """ if row: custom = str(row.get("openrouter_model") or "").strip() if custom: return custom return default_openrouter_model_id()