Subscription and Extraction Features

radis/chats/utils/chat_client.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Iterable
+from typing import Iterable, NoReturn
 
 import openai
 from django.conf import settings
@@ -16,53 +16,171 @@ def _get_base_url() -> str:
     return base_url
 
 
-class AsyncChatClient:
+def _validate_completion_response(completion) -> str:
+    """
+    Validates that the LLM completion response contains valid content.
+
+    Args:
+        completion: The completion response from the LLM
+
+    Returns:
+        The message content as a string
+
+    Raises:
+        ValueError: If the response is empty or invalid
+    """
+    if not completion.choices:
+        logger.error("LLM returned empty choices list")
+        raise ValueError("LLM returned no response choices")
+
+    answer = completion.choices[0].message.content
+    if answer is None:
+        logger.error("LLM returned None for message content")
+        raise ValueError("LLM returned empty response content")
+
+    return answer
+
+
+def _validate_parsed_response(completion) -> BaseModel:
+    """
+    Validates that the LLM completion response contains valid parsed data.
+
+    Args:
+        completion: The completion response from the LLM
+
+    Returns:
+        The parsed BaseModel instance
+
+    Raises:
+        ValueError: If the response is empty or invalid
+    """
+    if not completion.choices:
+        logger.error("LLM returned empty choices list")
+        raise ValueError("LLM returned no response choices")
+
+    parsed = completion.choices[0].message.parsed
+    if parsed is None:
+        logger.error("LLM returned None for parsed message")
+        raise ValueError("LLM returned empty parsed response")
+
+    return parsed
+
+
+def _handle_api_error(error: openai.APIError, operation: str) -> NoReturn:
+    """
+    Logs and re-raises API errors with consistent error messages.
+
+    Args:
+        error: The API error from OpenAI
+        operation: Description of the operation that failed (e.g., "chat", "data extraction")
+
+    Raises:
+        RuntimeError: Always raises with a user-friendly error message
+    """
+    logger.error(f"OpenAI API error during {operation}: {error}")
+    raise RuntimeError(f"Failed to communicate with LLM service: {error}") from error
+
+
+class _BaseChatClient:
+    """Base class containing shared chat client logic."""
+
     def __init__(self):
-        base_url = _get_base_url()
-        api_key = settings.EXTERNAL_LLM_PROVIDER_API_KEY
-        self._client = openai.AsyncOpenAI(base_url=base_url, api_key=api_key)
+        self._base_url = _get_base_url()
+        self._api_key = settings.EXTERNAL_LLM_PROVIDER_API_KEY
         self._model_name = settings.LLM_MODEL_NAME
 
-    async def chat(
+    def _build_chat_request(
         self,
         messages: Iterable[ChatCompletionMessageParam],
         max_completion_tokens: int | None = None,
-    ) -> str:
-        logger.debug(f"Sending messages to LLM for chat:\n{messages}")
-
+    ) -> dict:
+        """Build the request dictionary for chat completion."""
         request = {
             "model": self._model_name,
             "messages": messages,
         }
         if max_completion_tokens is not None:
             request["max_completion_tokens"] = max_completion_tokens
+        return request
 
-        completion = await self._client.chat.completions.create(**request)
-        answer = completion.choices[0].message.content
-        assert answer is not None
+    def _log_request(self, messages: Iterable[ChatCompletionMessageParam]) -> None:
+        """Log the outgoing request."""
+        logger.debug(f"Sending messages to LLM for chat:\n{messages}")
+
+    def _log_response(self, answer: str) -> None:
+        """Log the response from LLM."""
         logger.debug("Received from LLM: %s", answer)
+
+
+class AsyncChatClient(_BaseChatClient):
+    def __init__(self):
+        super().__init__()
+        self._client = openai.AsyncOpenAI(base_url=self._base_url, api_key=self._api_key)
+
+    async def chat(
+        self,
+        messages: Iterable[ChatCompletionMessageParam],
+        max_completion_tokens: int | None = None,
+    ) -> str:
+        self._log_request(messages)
+        request = self._build_chat_request(messages, max_completion_tokens)
+
+        try:
+            completion = await self._client.chat.completions.create(**request)
+        except openai.APIError as e:
+            _handle_api_error(e, "chat")
+
+        answer = _validate_completion_response(completion)
+        self._log_response(answer)
         return answer
 
 
-class ChatClient:
+class ChatClient(_BaseChatClient):
     def __init__(self) -> None:
-        base_url = _get_base_url()
-        api_key = settings.EXTERNAL_LLM_PROVIDER_API_KEY
+        super().__init__()
+        self._client = openai.OpenAI(base_url=self._base_url, api_key=self._api_key)
+
+    def chat(
+        self,
+        messages: Iterable[ChatCompletionMessageParam],
+        max_completion_tokens: int | None = None,
+    ) -> str:
+        """
+        Send messages to LLM and return the response text.
 
-        self._client = openai.OpenAI(base_url=base_url, api_key=api_key)
-        self._llm_model_name = settings.LLM_MODEL_NAME
+        Args:
+            messages: List of message dictionaries with 'role' and 'content'
+            max_completion_tokens: Optional maximum tokens to generate
+
+        Returns:
+            The LLM's response as a string
+        """
+        self._log_request(messages)
+        request = self._build_chat_request(messages, max_completion_tokens)
+
+        try:
+            completion = self._client.chat.completions.create(**request)
+        except openai.APIError as e:
+            _handle_api_error(e, "chat")
+
+        answer = _validate_completion_response(completion)
+        self._log_response(answer)
+        return answer
 
     def extract_data(self, prompt: str, schema: type[BaseModel]) -> BaseModel:
         logger.debug("Sending prompt and schema to LLM to extract data.")
         logger.debug("Prompt:\n%s", prompt)
         logger.debug("Schema:\n%s", schema.model_json_schema())
 
-        completion = self._client.beta.chat.completions.parse(
-            model=self._llm_model_name,
-            messages=[{"role": "system", "content": prompt}],
-            response_format=schema,
-        )
-        event = completion.choices[0].message.parsed
-        assert event
+        try:
+            completion = self._client.beta.chat.completions.parse(
+                model=self._model_name,
+                messages=[{"role": "system", "content": prompt}],
+                response_format=schema,
+            )
+        except openai.APIError as e:
+            _handle_api_error(e, "data extraction")
+
+        event = _validate_parsed_response(completion)
         logger.debug("Received from LLM: %s", event)
         return event

radis/core/constants.py

@@ -2,3 +2,7 @@
     "de": "German",
     "en": "English",
 }
+
+MIN_AGE = 0
+MAX_AGE = 120
+AGE_STEP = 10

radis/core/form_fields.py

@@ -0,0 +1,179 @@
+"""
+Reusable form field factories for RADIS forms.
+
+This module provides factory functions for commonly used form fields
+to reduce duplication across the codebase.
+"""
+
+from typing import Literal, overload
+
+from django import forms
+
+from radis.core.constants import AGE_STEP, LANGUAGE_LABELS, MAX_AGE, MIN_AGE
+from radis.reports.models import Language, Modality
+
+
+@overload
+def create_language_field(
+    required: bool = False,
+    empty_label: str | None = None,
+    use_pk: Literal[True] = True,
+) -> forms.ModelChoiceField: ...
+
+
+@overload
+def create_language_field(
+    required: bool = False,
+    empty_label: str | None = None,
+    use_pk: Literal[False] = False,
+) -> forms.ChoiceField: ...
+
+
+def create_language_field(
+    required: bool = False,
+    empty_label: str | None = None,
+    use_pk: bool = True,
+) -> forms.ModelChoiceField | forms.ChoiceField:
+    """
+    Create a language choice field with consistent configuration.
+
+    Args:
+        required: Whether the field is required
+        empty_label: Label for empty option (None = no empty option)
+        use_pk: If True, returns ModelChoiceField with Language objects;
+                if False, returns ChoiceField with code strings
+
+    Returns:
+        ModelChoiceField (if use_pk=True) or ChoiceField (if use_pk=False)
+
+    Example:
+        # For extraction forms (uses ModelChoiceField, returns Language objects)
+        self.fields["language"] = create_language_field()
+
+        # For subscription forms (uses ModelChoiceField, allows "All")
+        self.fields["language"] = create_language_field(empty_label="All")
+
+        # For search forms (uses ChoiceField with codes)
+        self.fields["language"] = create_language_field(use_pk=False)
+    """
+    languages = Language.objects.order_by("code")
+
+    if use_pk:
+        # Return ModelChoiceField - cleaned_data will contain Language objects
+        field = forms.ModelChoiceField(
+            queryset=languages,
+            required=required,
+            empty_label=empty_label,
+        )
+        field.label_from_instance = (  # type: ignore[method-assign]
+            lambda obj: LANGUAGE_LABELS.get(obj.code, obj.code)
+        )
+        return field
+    else:
+        # Return ChoiceField - cleaned_data will contain code strings
+        choices = [(lang.code, LANGUAGE_LABELS.get(lang.code, lang.code)) for lang in languages]
+        if empty_label is not None:
+            choices.insert(0, ("", empty_label))
+        field = forms.ChoiceField(required=required, choices=choices)
+        return field
+
+
+@overload
+def create_modality_field(
+    required: bool = False,
+    widget_size: int = 6,
+    use_pk: Literal[True] = True,
+) -> forms.ModelMultipleChoiceField: ...
+
+
+@overload
+def create_modality_field(
+    required: bool = False,
+    widget_size: int = 6,
+    use_pk: Literal[False] = False,
+) -> forms.MultipleChoiceField: ...
+
+
+def create_modality_field(
+    required: bool = False,
+    widget_size: int = 6,
+    use_pk: bool = True,
+) -> forms.ModelMultipleChoiceField | forms.MultipleChoiceField:
+    """
+    Create a modality multiple choice field with consistent configuration.
+
+    Args:
+        required: Whether the field is required
+        widget_size: Height of the select widget
+        use_pk: If True, returns ModelMultipleChoiceField with Modality objects;
+                if False, returns MultipleChoiceField with code strings
+
+    Returns:
+        ModelMultipleChoiceField (if use_pk=True) or MultipleChoiceField (if use_pk=False)
+
+    Example:
+        # For extraction forms (uses ModelMultipleChoiceField, returns Modality objects)
+        self.fields["modalities"] = create_modality_field()
+
+        # For search forms (uses MultipleChoiceField with codes)
+        self.fields["modalities"] = create_modality_field(use_pk=False)
+    """
+    modalities = Modality.objects.filter(filterable=True).order_by("code")
+
+    if use_pk:
+        # Return ModelMultipleChoiceField - cleaned_data will contain Modality objects
+        field = forms.ModelMultipleChoiceField(
+            queryset=modalities,
+            required=required,
+        )
+        # Display just the code for each modality
+        field.label_from_instance = lambda obj: obj.code
+        field.widget.attrs["size"] = widget_size
+        return field
+    else:
+        # Return MultipleChoiceField - cleaned_data will contain code strings
+        field = forms.MultipleChoiceField(required=required)
+        field.choices = [(mod.code, mod.code) for mod in modalities]
+        field.widget.attrs["size"] = widget_size
+        return field
+
+
+def create_age_range_fields() -> tuple[forms.IntegerField, forms.IntegerField]:
+    """
+    Create age_from and age_till fields with consistent configuration.
+
+    Returns:
+        Tuple of (age_from_field, age_till_field)
+
+    Example:
+        age_from, age_till = create_age_range_fields()
+        self.fields["age_from"] = age_from
+        self.fields["age_till"] = age_till
+    """
+    age_from = forms.IntegerField(
+        required=False,
+        min_value=MIN_AGE,
+        max_value=MAX_AGE,
+        widget=forms.NumberInput(
+            attrs={
+                "type": "range",
+                "step": AGE_STEP,
+                "value": MIN_AGE,
+            }
+        ),
+    )
+
+    age_till = forms.IntegerField(
+        required=False,
+        min_value=MIN_AGE,
+        max_value=MAX_AGE,
+        widget=forms.NumberInput(
+            attrs={
+                "type": "range",
+                "step": AGE_STEP,
+                "value": MAX_AGE,
+            }
+        ),
+    )
+
+    return age_from, age_till