Upload services/openrouter_client.py with huggingface_hub

services/openrouter_client.py

@@ -0,0 +1,127 @@
+"""Minimal OpenRouter chat-completions client used by the AI service."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import urllib.error
+import urllib.request
+from typing import Any
+
+
+class OpenRouterClient:
+    """Small helper for calling the OpenRouter OpenAI-compatible API."""
+
+    def __init__(self, api_key: str, model: str):
+        api_key = (api_key or "").strip()
+        if not api_key:
+            raise ValueError("OPENROUTER_API_KEY is required")
+
+        self.api_key = api_key
+        self.model = model
+        self.base_url = os.getenv(
+            "OPENROUTER_BASE_URL",
+            "https://openrouter.ai/api/v1/chat/completions",
+        )
+        self.app_name = os.getenv("OPENROUTER_APP_NAME", "Debatra AI Service")
+        self.site_url = os.getenv("OPENROUTER_SITE_URL", "http://localhost")
+
+    async def chat_completion(
+        self,
+        messages: list[dict[str, str]],
+        *,
+        temperature: float = 0.4,
+        max_tokens: int = 1024,
+    ) -> str:
+        payload = {
+            "model": self.model,
+            "messages": messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+        }
+
+        data = await asyncio.to_thread(self._post_json, payload)
+        return self._extract_content(data)
+
+    async def chat_completion_json(
+        self,
+        messages: list[dict[str, str]],
+        *,
+        temperature: float = 0.4,
+        max_tokens: int = 1024,
+    ) -> dict[str, Any]:
+        text = await self.chat_completion(
+            messages,
+            temperature=temperature,
+            max_tokens=max_tokens,
+        )
+        text = self._strip_code_fences(text)
+
+        if not text:
+            raise ValueError("OpenRouter returned an empty response")
+
+        try:
+            return json.loads(text)
+        except json.JSONDecodeError as exc:
+            raise ValueError("OpenRouter response was not valid JSON") from exc
+
+    def _post_json(self, payload: dict[str, Any]) -> dict[str, Any]:
+        request = urllib.request.Request(
+            self.base_url,
+            data=json.dumps(payload).encode("utf-8"),
+            headers={
+                "Authorization": f"Bearer {self.api_key}",
+                "Content-Type": "application/json",
+                "HTTP-Referer": self.site_url,
+                "X-Title": self.app_name,
+            },
+            method="POST",
+        )
+
+        try:
+            with urllib.request.urlopen(request, timeout=60) as response:
+                body = response.read().decode("utf-8")
+        except urllib.error.HTTPError as error:
+            error_body = error.read().decode("utf-8", errors="replace") if error.fp else ""
+            raise RuntimeError(
+                f"OpenRouter API error {error.code}: {error_body or error.reason}"
+            ) from error
+        except urllib.error.URLError as error:
+            raise RuntimeError(f"OpenRouter request failed: {error.reason}") from error
+
+        try:
+            return json.loads(body)
+        except json.JSONDecodeError as exc:
+            raise RuntimeError("OpenRouter returned invalid JSON") from exc
+
+    @staticmethod
+    def _extract_content(response: dict[str, Any]) -> str:
+        choices = response.get("choices") or []
+        if not choices:
+            raise ValueError("OpenRouter returned no choices")
+
+        message = choices[0].get("message") or {}
+        content = message.get("content") or ""
+
+        if isinstance(content, list):
+            fragments: list[str] = []
+            for fragment in content:
+                if isinstance(fragment, dict):
+                    fragments.append(str(fragment.get("text", "")))
+                else:
+                    fragments.append(str(fragment))
+            content = "".join(fragments)
+
+        return str(content).strip()
+
+    @staticmethod
+    def _strip_code_fences(text: str) -> str:
+        text = text.strip()
+        if text.startswith("```"):
+            parts = text.split("```", 2)
+            if len(parts) >= 3:
+                text = parts[1]
+                if text.startswith("json"):
+                    text = text[4:]
+        return text.strip()