app/study.py

"""Versioned study definitions and server-side response validation."""

from __future__ import annotations

import json
from dataclasses import dataclass


STUDY_ID = "modality_revision_v1"
CONDITIONS = {"text_only", "text_audio"}
ELIGIBLE_GATES = {"substantive_answer_attempt", "explicit_disclosure_boundary"}


@dataclass(frozen=True)
class Option:
    value: str
    label: str
    hint: str = ""


GATE_OPTIONS = [
    Option("substantive_answer_attempt", "Substantive answer attempt", "The executive attempts to answer the information request."),
    Option("clarification_repair", "Clarification / dialogue repair", "The executive asks what the analyst meant or which scope was intended."),
    Option("technical_failure", "Technical or hearing failure", "The response concerns inaudible, broken, or missing audio."),
    Option("interrupted_incomplete", "Interrupted / incomplete response", "The turn ends before a usable answer can be given."),
    Option("procedural_deferral", "Procedural deferral", "The answer is deferred to later in the call or offline."),
    Option("explicit_disclosure_boundary", "Explicit disclosure boundary", "The executive clearly states that the requested information is not disclosed."),
    Option("cannot_determine", "Cannot determine", "The available material is insufficient to classify the response opportunity."),
]
GATE_VALUES = {option.value for option in GATE_OPTIONS}

RASIAH_OPTIONS = [
    Option("direct", "Direct", "Addresses the core request explicitly and substantially."),
    Option("intermediate", "Intermediate", "Supplies some relevant information but leaves an important part unresolved."),
    Option("fully_evasive", "Fully evasive", "Does not supply the requested information."),
]
RASIAH_VALUES = {option.value for option in RASIAH_OPTIONS}

SUPPLIED_OPTIONS = [
    Option("none", "None"),
    Option("some", "Some"),
    Option("most", "Most"),
    Option("all", "All"),
]
SUPPLIED_VALUES = {option.value for option in SUPPLIED_OPTIONS}

DESCRIPTORS = [
    Option("hesitant", "Hesitant"),
    Option("uncertain", "Uncertain"),
    Option("defensive", "Defensive"),
    Option("emphatic", "Emphatic"),
    Option("cooperative", "Cooperative"),
    Option("rehearsed", "Rehearsed / polished"),
    Option("strained", "Strained"),
    Option("spontaneous", "Spontaneous"),
]
DESCRIPTOR_VALUES = {"1", "2", "3", "4", "5", "na"}

AUDIO_REVISION_VALUES = {"more_responsive", "unchanged", "less_responsive", "cannot_determine"}
AUDIO_EVENT_VALUES = {
    "response_latency",
    "pause",
    "pace_change",
    "emphasis",
    "pitch_intonation",
    "voice_quality",
    "interruption_overlap",
    "no_specific_event",
    "other",
}


def parse_int(value: str | int | None, minimum: int, maximum: int) -> int | None:
    try:
        parsed = int(value)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return None
    return parsed if minimum <= parsed <= maximum else None


def validate_submission(
    form: dict, condition: str, audio_duration_s: float | None = None
) -> tuple[dict | None, str | None]:
    """Validate and normalize a study response.

    Returns ``(clean, None)`` on success or ``(None, public_error_code)``.
    """
    gate = str(form.get("gate") or "")
    gate_confidence = parse_int(form.get("gate_confidence"), 0, 100)
    if (
        gate not in GATE_VALUES
        or gate_confidence is None
        or str(form.get("gate_confidence_touched")) != "1"
    ):
        return None, "complete_gate"

    clean: dict = {
        "gate": gate,
        "gate_confidence": gate_confidence,
        "responsiveness": None,
        "rasiah": None,
        "supplied_information": None,
        "confidence": None,
        "delivery_descriptors_json": None,
        "audio_revision": None,
        "audible_event": None,
        "audible_event_other": None,
    }

    if gate in ELIGIBLE_GATES:
        responsiveness = parse_int(form.get("responsiveness"), 0, 100)
        confidence = parse_int(form.get("confidence"), 0, 100)
        rasiah = str(form.get("rasiah") or "")
        supplied = str(form.get("supplied_information") or "")
        if (
            responsiveness is None
            or confidence is None
            or str(form.get("responsiveness_touched")) != "1"
            or str(form.get("confidence_touched")) != "1"
        ):
            return None, "complete_scales"
        if rasiah not in RASIAH_VALUES or supplied not in SUPPLIED_VALUES:
            return None, "complete_responsiveness"
        descriptors = {}
        for option in DESCRIPTORS:
            value = str(form.get(f"descriptor_{option.value}") or "")
            if value not in DESCRIPTOR_VALUES:
                return None, "complete_delivery"
            descriptors[option.value] = value
        clean.update(
            {
                "responsiveness": responsiveness,
                "confidence": confidence,
                "rasiah": rasiah,
                "supplied_information": supplied,
                "delivery_descriptors_json": json.dumps(descriptors, sort_keys=True),
            }
        )

        if condition == "text_audio":
            audio_revision = str(form.get("audio_revision") or "")
            audible_event = str(form.get("audible_event") or "")
            if audio_revision not in AUDIO_REVISION_VALUES:
                return None, "complete_audio_revision"
            if audible_event not in AUDIO_EVENT_VALUES:
                return None, "complete_audio_event"
            other = str(form.get("audible_event_other") or "").strip()[:300] or None
            if audible_event == "other" and not other:
                return None, "complete_audio_event"
            clean.update(
                {
                    "audio_revision": audio_revision,
                    "audible_event": audible_event,
                    "audible_event_other": other,
                }
            )

    rationale = str(form.get("rationale") or "").strip()
    if len(rationale) < 8:
        return None, "add_rationale"
    clean["rationale"] = rationale[:1000]
    clean["response_time_ms"] = parse_int(form.get("response_time_ms"), 0, 7_200_000)
    clean["audio_played_ms"] = parse_int(form.get("audio_played_ms"), 0, 7_200_000) or 0
    clean["audio_completed"] = str(form.get("audio_completed") or "0") == "1"
    if condition == "text_audio":
        minimum_play_ms = min(
            5_000,
            max(1_000, int((audio_duration_s or 0) * 300)),
        )
        if not clean["audio_completed"] or clean["audio_played_ms"] < minimum_play_ms:
            return None, "play_audio"
    return clean, None