Source code for bhf_smarthealth.validate.qualtrics_schema

import logging

from pydantic import (
    BaseModel,
    model_validator,
    Field,
    RootModel,
    ConfigDict,
    BeforeValidator,
)

from datetime import date, datetime
from typing import Optional, Self, Annotated, Any


logger = logging.getLogger(__name__)


# parse datetimes with slashes and irregular spacing


def parse_custom_datetime(value: str) -> datetime:
    if isinstance(value, str):
        return datetime.strptime(value.strip(), "%Y-%m-%d %H:%M:%S")
    return value





def parse_custom_date(value: str) -> date:
    if isinstance(value, str):
        parsed_dt = datetime.strptime(value.strip(), "%d/%m/%Y")
        return parsed_dt.date()
    return value



# create custom types
CustomDate = Annotated[date, BeforeValidator(parse_custom_date)]
CustomDatetime = Annotated[datetime, BeforeValidator(parse_custom_datetime)]


# Validates one row of the base questions csv


class BaseQuestionsRow(BaseModel):
    model_config = ConfigDict(extra="forbid")

    duration_secs: int = Field(validation_alias="Duration (in seconds)")
    record_date: CustomDatetime = Field(validation_alias="RecordedDate")
    sheffield_id: str = Field(validation_alias="ResponseId")
    sig_date: CustomDate = Field(validation_alias="Sig_date")
    sex: str = Field(validation_alias="Sex")
    gender: str = Field(validation_alias="Gender")
    gender_free_text: Optional[str] = Field(validation_alias="Gender_4_TEXT")
    recruited_from_NHS: Optional[str] = Field(validation_alias="site1")
    recruited_from_NHS_site: Optional[str] = Field(validation_alias="site2")
    recruited_from_NHS_site_free_text: Optional[str] = Field(
        validation_alias="site2_9_TEXT"
    )
    first_heard_choice: Optional[str] = Field(validation_alias="site3")
    first_heard_other_study_free_text: Optional[str] = Field(
        validation_alias="site3_9_TEXT"
    )
    first_heard_free_text: Optional[str] = Field(validation_alias="site3_10_TEXT")
    comm_groups_choice: Optional[str] = Field(validation_alias="community")
    com_groups_free_text: Optional[str] = Field(validation_alias="community_7_TEXT")
    has_wearable: str = Field(validation_alias="Wearable1")
    wearable_list: Optional[str] | Optional[list[str]] = Field(
        validation_alias="Fitbit"
    )
    dom_hand: Optional[str] = Field(validation_alias="Dom_hand")
    wrist: Optional[str] = Field(validation_alias="Wrist")
    gps_consent: bool = Field(validation_alias="GPS")
    home_lsoa: str = Field(validation_alias="lsoa")
    qr_source: Optional[str] = Field(validation_alias="Source")
    age: int = Field(validation_alias="Age")



# Validates one row of the bridge file csv


class BridgeFileRow(BaseModel):
    model_config = ConfigDict(extra="forbid")

    sheffield_id: str = Field(validation_alias="ResponseId")
    patient_id_smarthealth: str





class BridgeFileTable(RootModel):
    root: list[BridgeFileRow]





class BaseQuestionsTable(RootModel):
    root: list[BaseQuestionsRow]



# Need to validate that the files always come together and hold the same IDs


class QualtricsStreamSchema(BaseModel):
    model_config = ConfigDict(extra="forbid")

    base_questions: BaseQuestionsTable
    bridge_file: BridgeFileTable



    @model_validator(mode="after")
    def check_participant_ids_match(self) -> Self:
        # Collect errors so all ID checks run, rather than fail-fast
        error_messages = []

        # !---- Check for IDs not in both files ----!
        bridge_ids = set(root.sheffield_id for root in self.bridge_file.root)
        base_ids = set(root.sheffield_id for root in self.base_questions.root)

        missing_ids = base_ids.symmetric_difference(bridge_ids)
        if missing_ids:
            error_messages.append(
                f"Some participant IDs are not in both the bridge AND base questions file: {', '.join(missing_ids)}"
            )

        # !---- Check for non-unique IDs ----!
        validation_checks = [
            ("Bridge file", self.bridge_file.root, "sheffield_id"),
            ("Bridge file", self.bridge_file.root, "patient_id_smarthealth"),
            ("Base questions file", self.base_questions.root, "sheffield_id"),
        ]

        for file_name, table, field_name in validation_checks:
            try:
                self._report_duplicates(
                    file_name=file_name, table=table, field_name=field_name
                )
            except ValueError as e:
                error_messages.append(str(e))

        if error_messages:
            raise ValueError(
                f"Qualtric validation checks failed for the following reasons:\n{('\n'.join(error_messages))}"
            )

        return self


    def _report_duplicates(
        self, file_name: str, table: list[Any], field_name: str
    ) -> None:
        """Helper function to check named field_name is unique, else return list of duplicated values"""
        seen = set()
        duplicates = set()

        for row in table:
            val = getattr(row, field_name)
            if val in seen:
                duplicates.add(val)
            else:
                seen.add(val)

        if duplicates:
            formatted_dups = ", ".join(duplicates)
            raise ValueError(
                f"Error in file {file_name}. Field {field_name} must be unique. Found duplicates: {formatted_dups}"
            )