""" title: RollCalc Assistant author: RollCalc version: 0.1.0 description: Deterministic RollCalc conversation API adapter """ from __future__ import annotations import asyncio import os import re from typing import Any, Callable from urllib.parse import urljoin, urlsplit import httpx from pydantic import BaseModel, Field, SecretStr def _environment_flag(name: str, default: bool = False) -> bool: value = os.getenv(name) if value is None: return default return value.strip().casefold() in {"1", "true", "yes", "on"} class Pipe: """Expose RollCalc as an OpenWebUI model without an LLM tool loop.""" class Valves(BaseModel): ROLLCALC_API_BASE_URL: str = Field( default=os.getenv("ROLLCALC_API_BASE_URL", ""), description=( "RollCalc URL reachable from the OpenWebUI server/container, " "for example http://host.docker.internal:5000" ), ) ROLLCALC_PUBLIC_BASE_URL: str = Field( default=os.getenv("ROLLCALC_PUBLIC_BASE_URL", ""), description=( "RollCalc URL reachable from the user's browser; used only " "for authenticated PDF links" ), ) ROLLCALC_USERNAME: str = Field( default=os.getenv("ROLLCALC_USERNAME", ""), description="HTTP Basic Auth username used server-side", ) ROLLCALC_PASSWORD: SecretStr = Field( default=SecretStr(os.getenv("ROLLCALC_PASSWORD", "")), description="HTTP Basic Auth password used server-side", ) ROLLCALC_TIMEOUT_SECONDS: float = Field( default=float(os.getenv("ROLLCALC_OPENWEBUI_TIMEOUT_SECONDS", "90")), ge=1, le=600, description="End-to-end timeout for one RollCalc request", ) ROLLCALC_DEBUG: bool = Field( default=_environment_flag("ROLLCALC_OPENWEBUI_DEBUG"), description="Log safe adapter diagnostics to the OpenWebUI console", ) type = "pipe" name = "RollCalc Assistant" def __init__( self, client_factory: Callable[..., Any] | None = None, ) -> None: self.valves = self.Valves() self._client_factory = client_factory or httpx.AsyncClient self._conversations: dict[tuple[str, str], str] = {} self._chat_locks: dict[tuple[str, str], asyncio.Lock] = {} self._lock_guard = asyncio.Lock() async def pipe( self, body: dict[str, Any], __user__: Any = None, __metadata__: dict[str, Any] | None = None, __chat_id__: str | None = None, __task__: str | None = None, ) -> str: """Forward one browser message to one authoritative RollCalc session.""" if __task__: return self._task_response(__task__) configuration_error = self._configuration_error() if configuration_error: return configuration_error message = self._user_message(body, __metadata__) if not message: return "Bitte gib eine RollCalc-Anfrage als Text ein." session_key = self._session_key( body, __user__, __metadata__, __chat_id__, ) if session_key is None: return ( "Die RollCalc-Sitzung konnte diesem Chat nicht sicher zugeordnet " "werden. Bitte verwende einen gespeicherten OpenWebUI-Chat." ) chat_lock = await self._chat_lock(session_key) async with chat_lock: try: return await self._send_message(session_key, message) except httpx.TimeoutException: return ( "RollCalc benötigt derzeit zu lange für eine Antwort. " "Bitte versuche es erneut." ) except httpx.RequestError: return ( "RollCalc ist derzeit nicht erreichbar. " "Bitte versuche es später erneut." ) except (TypeError, ValueError): return ( "RollCalc hat eine unerwartete Antwort geliefert. " "Bitte versuche es erneut." ) except Exception as error: self._debug(f"unexpected adapter error: {error.__class__.__name__}") return ( "Die RollCalc-Anfrage konnte nicht verarbeitet werden. " "Bitte versuche es erneut." ) async def _send_message( self, session_key: tuple[str, str], message: str, ) -> str: api_base = self.valves.ROLLCALC_API_BASE_URL.rstrip("/") timeout = self.valves.ROLLCALC_TIMEOUT_SECONDS auth = httpx.BasicAuth( self.valves.ROLLCALC_USERNAME, self._password(), ) async with self._client_factory(timeout=timeout, auth=auth) as client: conversation_id = self._conversations.get(session_key) if conversation_id is None: response = await client.post(f"{api_base}/api/conversations") error = self._http_error(response) if error: return error payload = self._json_object(response) conversation_id = payload.get("conversation_id") if not isinstance(conversation_id, str) or not conversation_id: raise ValueError("missing conversation_id") self._conversations[session_key] = conversation_id response = await client.post( f"{api_base}/api/conversations/{conversation_id}/messages", json={"message": message}, ) if response.status_code == 404: self._conversations.pop(session_key, None) return ( "Die RollCalc-Sitzung ist abgelaufen. Bitte starte die " "Berechnung mit deiner vollständigen Anfrage erneut." ) error = self._http_error(response) if error: return error payload = self._json_object(response) response_message = payload.get("message") if not isinstance(response_message, str) or not response_message.strip(): raise ValueError("missing user-facing message") return self._render_response(response_message, payload.get("pdf")) async def _chat_lock(self, key: tuple[str, str]) -> asyncio.Lock: async with self._lock_guard: lock = self._chat_locks.get(key) if lock is None: lock = asyncio.Lock() self._chat_locks[key] = lock return lock def _configuration_error(self) -> str | None: api_url = self.valves.ROLLCALC_API_BASE_URL.strip() if not self._valid_base_url(api_url): return ( "RollCalc ist in OpenWebUI noch nicht vollständig konfiguriert. " "Bitte hinterlege eine gültige interne RollCalc-URL." ) if not self.valves.ROLLCALC_USERNAME.strip() or not self._password(): return ( "Die RollCalc-Authentifizierung ist in OpenWebUI noch nicht " "vollständig konfiguriert." ) return None @staticmethod def _valid_base_url(value: str) -> bool: parsed = urlsplit(value) return ( parsed.scheme in {"http", "https"} and bool(parsed.netloc) and parsed.username is None and parsed.password is None ) def _password(self) -> str: password = self.valves.ROLLCALC_PASSWORD getter = getattr(password, "get_secret_value", None) return getter() if callable(getter) else str(password) @staticmethod def _json_object(response: Any) -> dict[str, Any]: payload = response.json() if not isinstance(payload, dict): raise ValueError("JSON object required") return payload @staticmethod def _http_error(response: Any) -> str | None: if 200 <= response.status_code < 300: return None if response.status_code == 400: try: payload = response.json() except Exception: payload = None message = payload.get("message") if isinstance(payload, dict) else None if isinstance(message, str) and message.strip(): return message return "Die Anfrage enthält ungültige Eingaben. Bitte prüfe sie." if response.status_code in {401, 403}: return ( "Die RollCalc-Authentifizierung ist nicht korrekt konfiguriert. " "Bitte informiere die Administration." ) if response.status_code == 503: return ( "Die lokale Spracherkennung ist derzeit nicht erreichbar. " "Bitte versuche es später erneut." ) if response.status_code == 504: return ( "Die lokale Spracherkennung benötigt derzeit zu lange. " "Bitte versuche es erneut." ) return ( "RollCalc konnte die Anfrage nicht verarbeiten. " "Bitte versuche es erneut." ) def _render_response(self, message: str, pdf: Any) -> str: lines = message.strip().splitlines() lines = [line for line in lines if not line.strip().startswith("PDF:")] if isinstance(pdf, dict) and isinstance(pdf.get("url"), str): report_path = pdf["url"] if self._valid_report_path(report_path): public_base = self.valves.ROLLCALC_PUBLIC_BASE_URL.strip() if self._valid_base_url(public_base): report_url = urljoin( f"{public_base.rstrip('/')}/", report_path.lstrip("/"), ) lines.extend(("", f"[PDF herunterladen]({report_url})")) else: lines.extend(( "", "PDF erstellt; öffentlicher Download-Link nicht " "konfiguriert.", )) return "\n".join(lines).strip() @staticmethod def _valid_report_path(value: str) -> bool: return bool( re.fullmatch( r"/api/conversations/reports/[A-Za-z0-9_-]+\.pdf", value, ) ) @staticmethod def _user_message( body: dict[str, Any], metadata: dict[str, Any] | None, ) -> str | None: if isinstance(metadata, dict): prompt = metadata.get("user_prompt") if isinstance(prompt, str) and prompt.strip(): return prompt.strip() messages = body.get("messages", []) if isinstance(body, dict) else [] if not isinstance(messages, list): return None for item in reversed(messages): if not isinstance(item, dict) or item.get("role") != "user": continue content = item.get("content") if isinstance(content, str) and content.strip(): return content.strip() return None @staticmethod def _session_key( body: dict[str, Any], user: Any, metadata: dict[str, Any] | None, injected_chat_id: str | None, ) -> tuple[str, str] | None: user_id = None if isinstance(user, dict): user_id = user.get("id") or user.get("email") else: user_id = getattr(user, "id", None) or getattr(user, "email", None) if not user_id and isinstance(metadata, dict): user_id = metadata.get("user_id") chat_id = injected_chat_id if isinstance(metadata, dict): chat_id = ( chat_id or metadata.get("chat_id") or metadata.get("session_id") ) if isinstance(body, dict): chat_id = chat_id or body.get("chat_id") or body.get("session_id") if not user_id or not chat_id: return None return str(user_id), str(chat_id) @staticmethod def _task_response(task: str) -> str: normalized = task.casefold() if "title" in normalized: return "RollCalc" if "emoji" in normalized: return "🧮" if "tag" in normalized or "follow" in normalized: return "[]" return "" def _debug(self, message: str) -> None: if self.valves.ROLLCALC_DEBUG: print(f"[RollCalc OpenWebUI] {message}")