penracourses/atlas/filters.py

import django_filters
from django import forms
from crispy_forms.helper import FormHelper
from crispy_forms.layout import Layout, Div, Field
from collections import OrderedDict

from .models import (
    Case,
    CaseCollection,
    Condition,
    Finding,
    PathologicalProcess,
    Presentation,
    Procedure,
    QuestionSchema,
    Series,
    Structure,
    Subspecialty,
    NormalCase,
    Resource,
    Site,
)
from django.contrib.auth.models import User
from django.db.models import Q
from django.http import QueryDict


def get_authors(request):
    if request is None:
        return User.objects.none()

    if not request.user.groups.filter(name="atlas_editor").exists():
        queryset = User.objects.filter(id=request.user.id)
    else:
        queryset = User.objects.all()

    return queryset


def get_cases(request):
    if request is None:
        return Case.objects.none()

    if not request.user.groups.filter(name="atlas_editor").exists():
        queryset = Case.objects.filter(author__id=request.user.id)
    else:
        queryset = Case.objects.all()

    return queryset


def get_series(request):
    if request is None:
        return Series.objects.none()

    if not request.user.groups.filter(name="atlas_editor").exists():
        queryset = Series.objects.filter(author__id=request.user.id)
    else:
        queryset = Series.objects.all()

    queryset = queryset.prefetch_related("examination", "plane")

    return queryset


class CaseCollectionFilter(django_filters.FilterSet):
    author = django_filters.ModelMultipleChoiceFilter(
        queryset=get_authors, null_label="No author"
    )
    cases = django_filters.ModelMultipleChoiceFilter(
        queryset=get_cases, null_label="No cases"
    )

    class Meta:
        model = CaseCollection
        fields = (
            "archive",
            "author",
        )  # "site",

    def __init__(
        self,
        data=None,
        queryset=None,
        prefix=None,
        strict=None,
        user=None,
        request=None,
    ):
        # Default the archive filter to False when not provided so collections
        # are not archived by default in listings.
        if data is None:
            data = QueryDict('', mutable=True)
            data['archive'] = 'false'
        else:
            # QueryDict instances need to be copied to be mutable
            try:
                if 'archive' not in data:
                    data = data.copy()
                    data['archive'] = 'false'
            except Exception:
                # Fallback for non-QueryDict mappings
                if isinstance(data, dict) and 'archive' not in data:
                    data = dict(data)
                    data['archive'] = 'false'
        if not request.user.groups.filter(name="atlas_editor").exists():
            queryset = queryset.prefetch_related("author").filter(
                author__id=request.user.id
            )
        super(CaseCollectionFilter, self).__init__(
            data=data, queryset=queryset, prefix=prefix, request=request
        )
        pass


class CaseFilter(django_filters.FilterSet):
    author = django_filters.ModelMultipleChoiceFilter(
        queryset=get_authors, null_label="No author"
    )
    # Show a distinct list of `Series.examination` values instead of raw Series
    # entries so examinations that differ only by plane/description appear once.
    series = django_filters.ModelMultipleChoiceFilter(
        field_name='series__examination',
        queryset=lambda request: __import__('generic.models', fromlist=['Examination']).Examination.objects.filter(atlas_series_examination__isnull=False).distinct(),
        label='Examination',
    )
    # Free-text search to help find the correct examination when there are
    # many similarly-named Series/Examination entries.
    examination_text = django_filters.CharFilter(method='filter_examination_text', label='Examination (text)')
    # Full list of conditions and a text search to find conditions
    condition = django_filters.ModelMultipleChoiceFilter(
        queryset=Condition.objects.all(),
        label='Condition',
    )
    condition_text = django_filters.CharFilter(method='filter_condition_text', label='Condition (text)')
    differential_case = django_filters.ModelMultipleChoiceFilter(
        label="Differential",
        field_name="differentialcase__condition",
        # to_field_name="differential",
        queryset=Condition.objects.all(),
    )

    class Meta:
        model = Case
        fields = (
            "created_date",
            "open_access",
            "author",
            "differential_case",
        )  # "site",

    def __init__(
        self,
        data=None,
        queryset=None,
        prefix=None,
        strict=None,
        user=None,
        request=None,
    ):
        if not request.user.groups.filter(name="atlas_editor").exists():
            queryset = queryset.prefetch_related("author").filter(
                (Q(author__id=request.user.id) | Q(open_access=True))
            )
        super(CaseFilter, self).__init__(
            data=data, queryset=queryset, prefix=prefix, request=request
        )
        pass

    def filter_examination_text(self, queryset, name, value):
        """Filter Cases by a text search on related Examination records.

        Matches against the Examination string representation and any
        common identifying fields (name/code). Uses a case-insensitive
        contains lookup and filters Cases which have a Series linked to
        a matching Examination.
        """
        if not value:
            return queryset

        from generic.models import Examination

        # Only attempt to match PK when the provided value looks like an integer
        try:
            pk_val = int(value)
        except Exception:
            pk_val = None

        if pk_val is not None:
            matches = Examination.objects.filter(
                Q(examination__icontains=value) | Q(pk=pk_val)
            ).distinct()
        else:
            matches = Examination.objects.filter(
                Q(examination__icontains=value)
            ).distinct()

        return queryset.filter(series__examination__in=matches).distinct()

    def filter_condition_text(self, queryset, name, value):
        """Filter Cases by a text search on related Condition records.

        Matches both direct `condition` M2M and `differential` entries.
        """
        if not value:
            return queryset

        # Only try PK match when numeric
        try:
            pk_val = int(value)
        except Exception:
            pk_val = None

        if pk_val is not None:
            matches = Condition.objects.filter(Q(name__icontains=value) | Q(pk=pk_val)).distinct()
        else:
            matches = Condition.objects.filter(Q(name__icontains=value)).distinct()

        # Match either direct condition M2M or differential (through Differential)
        return queryset.filter(Q(condition__in=matches) | Q(differential__in=matches)).distinct()


class SeriesFilter(django_filters.FilterSet):
    case = django_filters.ModelMultipleChoiceFilter(
        queryset=get_cases, null_label="No case"
    )
    author = django_filters.ModelMultipleChoiceFilter(
        queryset=get_authors, null_label="No author"
    )

    class Meta:
        model = Series
        fields = ("modality", "case", "examination", "author")

    def __init__(
        self,
        data=None,
        queryset=None,
        prefix=None,
        strict=None,
        user=None,
        request=None,
    ):
        if not request.user.groups.filter(name="atlas_editor").exists():
            # queryset = queryset.filter(open_access=True) | queryset.filter(author__id=request.user.id)
            queryset = queryset.filter(author__id=request.user.id)
        super(SeriesFilter, self).__init__(
            data=data, queryset=queryset, prefix=prefix, request=request
        )
        pass


class ConditionFilter(django_filters.FilterSet):
    # Replace legacy `primary` filter with `is_canonical` and provide a
    # `synonym` filter which will filter by the canonical group of the
    # selected condition.
    is_canonical = django_filters.BooleanFilter(method="filter_is_canonical", label="Canonical")
    synonym = django_filters.ModelChoiceFilter(queryset=Condition.objects.all(), method="filter_synonym", label="Synonym group")

    class Meta:
        model = Condition
        fields = {
            "name": ["icontains"],
            "subspecialty": ["exact"],
            "rcr_curriculum": ["exact"],
        }

    def __init__(
        self,
        data=None,
        queryset=None,
        prefix=None,
        strict=None,
        user=None,
        request=None,
    ):
        super(ConditionFilter, self).__init__(
            data=data, queryset=queryset, prefix=prefix, request=request
        )
        pass

    def filter_is_canonical(self, queryset, name, value):
        if value in (True, "True", "true", 1, "1"):
            return queryset.filter(canonical__isnull=True)
        if value in (False, "False", "false", 0, "0"):
            return queryset.filter(canonical__isnull=False)
        return queryset

    def filter_synonym(self, queryset, name, value):
        # value is a Condition instance selected in the filter. Return all
        # Conditions that belong to the same canonical group as `value`.
        if not value:
            return queryset
        master = value.canonical if value.canonical else value
        return queryset.filter(Q(canonical=master) | Q(pk=master.pk))


class FindingFilter(django_filters.FilterSet):
    # Replace legacy `primary`/`synonym` with canonical-based filters
    is_canonical = django_filters.BooleanFilter(method="filter_is_canonical", label="Canonical")
    synonym = django_filters.ModelChoiceFilter(queryset=Finding.objects.all(), method="filter_synonym", label="Synonym group")

    class Meta:
        model = Finding
        fields = {
            "name": ["icontains"],
        }

    def __init__(
        self,
        data=None,
        queryset=None,
        prefix=None,
        strict=None,
        user=None,
        request=None,
    ):
        super(FindingFilter, self).__init__(
            data=data, queryset=queryset, prefix=prefix, request=request
        )
        pass

    def filter_is_canonical(self, queryset, name, value):
        if value in (True, "True", "true", 1, "1"):
            return queryset.filter(canonical__isnull=True)
        if value in (False, "False", "false", 0, "0"):
            return queryset.filter(canonical__isnull=False)
        return queryset

    def filter_synonym(self, queryset, name, value):
        # value is a Finding instance selected in the filter. Return all
        # Findings that belong to the same canonical group as `value`.
        if not value:
            return queryset
        master = value.canonical if value.canonical else value
        return queryset.filter(Q(canonical=master) | Q(pk=master.pk))


class StructureFilter(django_filters.FilterSet):
    is_canonical = django_filters.BooleanFilter(method="filter_is_canonical", label="Canonical")
    synonym = django_filters.ModelChoiceFilter(queryset=Structure.objects.all(), method="filter_synonym", label="Synonym group")

    class Meta:
        model = Structure
        fields = {"name": ["icontains"]}

    def __init__(
        self,
        data=None,
        queryset=None,
        prefix=None,
        strict=None,
        user=None,
        request=None,
    ):
        super(StructureFilter, self).__init__(
            data=data, queryset=queryset, prefix=prefix, request=request
        )
        pass

    def filter_is_canonical(self, queryset, name, value):
        if value in (True, "True", "true", 1, "1"):
            return queryset.filter(canonical__isnull=True)
        if value in (False, "False", "false", 0, "0"):
            return queryset.filter(canonical__isnull=False)
        return queryset

    def filter_synonym(self, queryset, name, value):
        if not value:
            return queryset
        master = value.canonical if value.canonical else value
        return queryset.filter(Q(canonical=master) | Q(pk=master.pk))


class PresentationFilter(django_filters.FilterSet):
    class Meta:
        model = Presentation
        fields = ("name", "subspecialty")

    def __init__(
        self,
        data=None,
        queryset=None,
        prefix=None,
        strict=None,
        user=None,
        request=None,
    ):
        super(PresentationFilter, self).__init__(
            data=data, queryset=queryset, prefix=prefix, request=request
        )
        pass


class PathologicalProcessFilter(django_filters.FilterSet):
    class Meta:
        model = PathologicalProcess
        fields = ("name",)

    def __init__(
        self,
        data=None,
        queryset=None,
        prefix=None,
        strict=None,
        user=None,
        request=None,
    ):
        super(PathologicalProcessFilter, self).__init__(
            data=data, queryset=queryset, prefix=prefix, request=request
        )
        pass


class SubspecialtyFilter(django_filters.FilterSet):
    class Meta:
        model = Subspecialty
        fields = ("name",)

    def __init__(
        self,
        data=None,
        queryset=None,
        prefix=None,
        strict=None,
        user=None,
        request=None,
    ):
        super(SubspecialtyFilter, self).__init__(
            data=data, queryset=queryset, prefix=prefix, request=request
        )
        pass


class NormalCaseFilter(django_filters.FilterSet):
    """FilterSet for the NormalCase model allowing filtering by
    examination, modality and nearest-year age ranges.
    """
    examination = django_filters.ModelChoiceFilter(queryset=Series.objects.none())
    modality = django_filters.ModelChoiceFilter(queryset=Series.objects.none())
    procedures = django_filters.ModelMultipleChoiceFilter(field_name="case__procedures", queryset=Procedure.objects.all(), label="Procedure")
    # Accept min/max as years in the UI but filter against the canonical
    # `age_days` field by converting years -> days.
    # Numeric min/max values entered by the user (interpreted via `age_unit`)
    min_value = django_filters.NumberFilter(label="Min", method="filter_min_value")
    max_value = django_filters.NumberFilter(label="Max", method="filter_max_value")
    AGE_UNIT_CHOICES = (
        ("days", "Days"),
        ("weeks", "Weeks"),
        ("months", "Months"),
        ("years", "Years"),
    )
    age_unit = django_filters.ChoiceFilter(label="Unit", choices=AGE_UNIT_CHOICES, method="filter_age_unit")

    class Meta:
        model = NormalCase
        fields = ("examination", "modality", "procedures", "age_days")

    def __init__(self, data=None, queryset=None, prefix=None, strict=None, user=None, request=None):
        # Use the global Examination and Modality querysets from the Series model
        from generic.models import Examination, Modality

        # We'll set the actual querysets for the filters after applying
        # any NormalCase queryset restrictions for the current user so
        # that the choices only include examinations/modalities that
        # actually have NormalCase records available.

        # If not an atlas editor restrict to cases authored by the user or open access cases
        if request is not None and not request.user.groups.filter(name="atlas_editor").exists():
            from django.db.models import Q

            if queryset is None:
                queryset = NormalCase.objects.select_related("case")
            queryset = queryset.filter((Q(case__author__id=request.user.id) | Q(case__open_access=True)))

        # Determine the NormalCase queryset we'll use to build choices
        nc_qs = queryset if queryset is not None else NormalCase.objects.all()

        # Limit examination and modality choices to those present on the
        # available NormalCase records (respecting user restrictions).
        try:
            exam_ids = nc_qs.values_list("examination_id", flat=True)
            mod_ids = nc_qs.values_list("modality_id", flat=True)
            proc_ids = nc_qs.values_list("case__procedures", flat=True)
            self.base_filters.get("examination").queryset = Examination.objects.filter(id__in=exam_ids).distinct()
            self.base_filters.get("modality").queryset = Modality.objects.filter(id__in=mod_ids).distinct()
            self.base_filters.get("procedures").queryset = Procedure.objects.filter(id__in=proc_ids).distinct()
        except Exception:
            # Fallback to all if something goes wrong
            self.base_filters.get("examination").queryset = Examination.objects.all()
            self.base_filters.get("modality").queryset = Modality.objects.all()

        super(NormalCaseFilter, self).__init__(data=data, queryset=queryset, prefix=prefix, request=request)

        # Attach a Crispy FormHelper to render the filter form as a
        # horizontal Bootstrap grid. We set `form_tag=False` because the
        # template already provides the <form> element.
        try:
            helper = FormHelper()
            helper.form_method = "get"
            helper.form_tag = False
            helper.form_class = "row g-2"
            helper.layout = Layout(
                Div(
                    Div("examination", css_class="col-md-3"),
                    Div("modality", css_class="col-md-3"),
                    Div("procedures", css_class="col-md-3"),
                    css_class="row g-2",
                ),
                Div(
                    Div(
                        Field("min_value", wrapper_class="mb-0 d-flex align-items-center"),
                        Field("max_value", wrapper_class="mb-0 d-flex align-items-center"),
                        Field("age_unit", wrapper_class="mb-0 d-flex align-items-center"),
                        css_class="col-12 d-flex gap-2 align-items-center",
                    ),
                    css_class="row g-2",
                ),
            )
            # Assign helper to the generated form instance and tweak widgets
            if hasattr(self, 'form') and self.form is not None:
                # Wider numeric inputs and default unit
                try:
                    form = self.form
                    if "min_value" in form.fields:
                        form.fields["min_value"].widget = forms.NumberInput(attrs={"class": "form-control w-auto", "min": "0", "step": "1", "style": "min-width:140px;"})
                    if "max_value" in form.fields:
                        form.fields["max_value"].widget = forms.NumberInput(attrs={"class": "form-control w-auto", "min": "0", "step": "1", "style": "min-width:140px;"})
                    if "age_unit" in form.fields:
                        form.fields["age_unit"].initial = "days"
                        form.fields["age_unit"].widget.attrs.update({"class": "form-select w-auto", "style": "min-width:220px;"})
                    form.helper = helper
                except Exception:
                    # don't crash if form structure differs
                    self.form.helper = helper
        except Exception:
            pass

    def _convert_to_days(self, value, unit: str) -> int:
        try:
            v = float(value)
        except Exception:
            raise

        if unit == "days":
            return int(v)
        if unit == "weeks":
            return int(v * 7)
        if unit == "months":
            # approximate month length
            return int(v * (365.25 / 12.0))
        # default: years
        return int(v * 365.25)

    def filter_min_value(self, queryset, name, value):
        try:
            unit = getattr(self.form, "cleaned_data", {}).get("age_unit", "days")
        except Exception:
            unit = "days"
        try:
            days = self._convert_to_days(value, unit)
        except Exception:
            return queryset
        return queryset.filter(age_days__gte=days)

    def filter_max_value(self, queryset, name, value):
        try:
            unit = getattr(self.form, "cleaned_data", {}).get("age_unit", "days")
        except Exception:
            unit = "days"
        try:
            days = self._convert_to_days(value, unit)
        except Exception:
            return queryset
        return queryset.filter(age_days__lte=days)

    def filter_age_unit(self, queryset, name, value):
        # `age_unit` is a helper input used by min/max conversion and does not
        # map to a database field on NormalCase.
        return queryset


class ResourceFilter(django_filters.FilterSet):
    name = django_filters.CharFilter(field_name='name', lookup_expr='icontains')
    site = django_filters.ModelMultipleChoiceFilter(field_name='sites', queryset=lambda request: Site.objects.all() if True else Site.objects.none())
    subspecialty = django_filters.ModelMultipleChoiceFilter(field_name='subspecialty', queryset=Subspecialty.objects.all())
    author = django_filters.ModelMultipleChoiceFilter(queryset=get_authors, null_label='No author')
    has_file = django_filters.BooleanFilter(method='filter_has_file', label='Has file')

    class Meta:
        model = Resource
        fields = ('name', 'site', 'subspecialty', 'author')

    def filter_has_file(self, queryset, name, value):
        if value is True:
            return queryset.exclude(file__isnull=True).exclude(file__exact='')
        elif value is False:
            return queryset.filter(file__isnull=True) | queryset.filter(file__exact='')
        return queryset

class QuestionSchemaFilter(django_filters.FilterSet):
    class Meta:
        model = QuestionSchema
        fields = ("name", "description")