shinkan-jinkendo/backend/openrouter_chat.py
Lars e4451e1362
Some checks failed
Test Suite / playwright-tests (push) Waiting to run
Deploy Development / deploy (push) Successful in 43s
Test Suite / pytest-backend (push) Failing after 1s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 13s
Test Suite / k6 /health Baseline (push) Has been cancelled
Enhance Exercise Management and AI Integration
- Updated the exercise form to include a tabbed navigation structure, improving user experience with sections for Stammdaten, Anleitung, Einordnung, Varianten, and Medien & Mehr.
- Introduced the concept of **Freigabelevel** (visibility level) in the UI, replacing previous terminology for clarity and consistency across components.
- Implemented new AI endpoints for exercise suggestions and regeneration, allowing for dynamic content generation without direct database writes.
- Removed the legacy `is_primary` flag from exercise skills in the UI, ensuring that intensity levels (`niedrig`, `mittel`, `hoch`) are the primary focus for skill management.
- Enhanced the variant management process with improved saving mechanisms and UI updates to reflect changes more intuitively.
2026-05-22 07:52:31 +02:00

101 lines
3.1 KiB
Python

"""
Minimal OpenRouter REST client (sync). Reads OPENROUTER_API_KEY / OPENROUTER_MODEL / OPENROUTER_BASE_URL from env.
"""
from __future__ import annotations
import json
import os
from typing import Any, Dict, List, Optional
import httpx
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
content = ""
if isinstance(inner, dict):
content = str(inner.get("content") or "")
elif isinstance(inner, str):
content = inner
elif isinstance(msg0.get("content"), str):
content = msg0.get("content") or ""
return content.strip()
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