Source code for nlpmed_portal.nlp.api.serializers

# SPDX-FileCopyrightText: Copyright (C) 2026 Omid Jafari <omidjafari.com>
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import codecs

from drf_writable_nested import WritableNestedModelSerializer
from rest_framework import serializers

from nlpmed_portal.nlp.constants import VTE_KEYWORD_EXC_LIST
from nlpmed_portal.nlp.constants import VTE_KEYWORD_INC_LIST
from nlpmed_portal.nlp.constants import VTE_SECTION_EXC_LIST
from nlpmed_portal.nlp.constants import VTE_SECTION_INC_LIST
from nlpmed_portal.nlp.models import DuplicateChecker
from nlpmed_portal.nlp.models import EncodingFixer
from nlpmed_portal.nlp.models import Joiner
from nlpmed_portal.nlp.models import MLInference
from nlpmed_portal.nlp.models import NlpProcessJob
from nlpmed_portal.nlp.models import NLPSetting
from nlpmed_portal.nlp.models import NoteFilter
from nlpmed_portal.nlp.models import PatternReplacer
from nlpmed_portal.nlp.models import SectionFilter
from nlpmed_portal.nlp.models import SectionSplitter
from nlpmed_portal.nlp.models import SentenceExpander
from nlpmed_portal.nlp.models import SentenceFilter
from nlpmed_portal.nlp.models import SentenceSegmenter
from nlpmed_portal.nlp.models import WordMasker




class EscapedNewlineCharField(serializers.CharField):


    def get_initial(self):
        value = super().get_initial()
        return value.replace("\n", "\\n").replace(r"\s", r"\\s") if value else value




    def to_internal_value(self, data):
        return codecs.decode(data, "unicode-escape")






class EncodingFixerSerializer(serializers.ModelSerializer):


    class Meta:
        model = EncodingFixer
        fields = ["status"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
        }






class PatternReplacerSerializer(serializers.ModelSerializer):
    pattern = EscapedNewlineCharField(
        initial=r"(?:\s*\n\s*){2,}",
        trim_whitespace=False,
    )
    target = EscapedNewlineCharField(initial="\n\n", trim_whitespace=False)



    class Meta:
        model = PatternReplacer
        fields = ["status", "pattern", "target"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "pattern": {"initial": r"(?:\s*\n\s*){2,}", "trim_whitespace": False},
            "target": {"initial": "\n\n", "trim_whitespace": False},
        }






class WordMaskerSerializer(serializers.ModelSerializer):


    class Meta:
        model = WordMasker
        fields = ["status", "words_to_mask", "mask_char"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "words_to_mask": {"initial": VTE_KEYWORD_EXC_LIST},
            "mask_char": {"initial": "*"},
        }






class NoteFilterSerializer(serializers.ModelSerializer):


    class Meta:
        model = NoteFilter
        fields = ["status", "words_to_search"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "words_to_search": {"initial": VTE_KEYWORD_INC_LIST},
        }






class SectionSplitterSerializer(serializers.ModelSerializer):
    delimiter = EscapedNewlineCharField(initial="\n\n", trim_whitespace=False)



    class Meta:
        model = SectionSplitter
        fields = ["status", "delimiter"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "delimiter": {"initial": "\n\n", "trim_whitespace": False},
        }






class SectionFilterSerializer(serializers.ModelSerializer):


    class Meta:
        model = SectionFilter
        fields = ["status", "section_inc_list", "section_exc_list", "fallback"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "section_inc_list": {"initial": VTE_SECTION_INC_LIST},
            "section_exc_list": {"initial": VTE_SECTION_EXC_LIST},
            "fallback": {"initial": True},
        }






class SentenceSegmenterSerializer(serializers.ModelSerializer):


    class Meta:
        model = SentenceSegmenter
        fields = ["status", "model_name", "batch_size"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "model_name": {"initial": "en_core_sci_lg"},
            "batch_size": {"initial": 10},
        }






class DuplicateCheckerSerializer(serializers.ModelSerializer):


    class Meta:
        model = DuplicateChecker
        fields = ["status", "num_perm", "sim_threshold", "length_threshold"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "num_perm": {"initial": 256},
            "sim_threshold": {"initial": 0.9},
            "length_threshold": {"initial": 50},
        }






class SentenceFilterSerializer(serializers.ModelSerializer):


    class Meta:
        model = SentenceFilter
        fields = ["status", "words_to_search"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "words_to_search": {"initial": VTE_KEYWORD_INC_LIST},
        }






class SentenceExpanderSerializer(serializers.ModelSerializer):


    class Meta:
        model = SentenceExpander
        fields = ["status", "length_threshold"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "length_threshold": {"initial": 50},
        }






class JoinerSerializer(serializers.ModelSerializer):
    sentence_delimiter = EscapedNewlineCharField(initial="\n", trim_whitespace=False)
    section_delimiter = EscapedNewlineCharField(initial="\n\n", trim_whitespace=False)



    class Meta:
        model = Joiner
        fields = ["status", "sentence_delimiter", "section_delimiter"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "sentence_delimiter": {"initial": "\n", "trim_whitespace": False},
            "section_delimiter": {"initial": "\n\n", "trim_whitespace": False},
        }






class MLInferenceSerializer(serializers.ModelSerializer):


    class Meta:
        model = MLInference
        fields = ["status", "model_name", "use_preped_text"]
        extra_kwargs = {
            "status": {"initial": "enabled"},
            "model_name": {"initial": "VTE_MULTICLASS"},
            "use_preped_text": {"initial": True},
        }




COMPONENTS = {
    "encoding_fixer": EncodingFixer,
    "pattern_replacer": PatternReplacer,
    "word_masker": WordMasker,
    "note_filter": NoteFilter,
    "section_splitter": SectionSplitter,
    "section_filter": SectionFilter,
    "sentence_segmenter": SentenceSegmenter,
    "duplicate_checker": DuplicateChecker,
    "sentence_filter": SentenceFilter,
    "sentence_expander": SentenceExpander,
    "joiner": Joiner,
    "ml_inference": MLInference,
}




class NLPSettingSerializer(WritableNestedModelSerializer):
    encoding_fixer = EncodingFixerSerializer(required=False)
    pattern_replacer = PatternReplacerSerializer(required=False)
    word_masker = WordMaskerSerializer(required=False)
    note_filter = NoteFilterSerializer(required=False)
    section_splitter = SectionSplitterSerializer(required=False)
    section_filter = SectionFilterSerializer(required=False)
    sentence_segmenter = SentenceSegmenterSerializer(required=False)
    duplicate_checker = DuplicateCheckerSerializer(required=False)
    sentence_filter = SentenceFilterSerializer(required=False)
    sentence_expander = SentenceExpanderSerializer(required=False)
    joiner = JoinerSerializer(required=False)
    ml_inference = MLInferenceSerializer(required=False)



    class Meta:
        model = NLPSetting
        fields = [
            "debug",
            "encoding_fixer",
            "pattern_replacer",
            "word_masker",
            "note_filter",
            "section_splitter",
            "section_filter",
            "sentence_segmenter",
            "duplicate_checker",
            "sentence_filter",
            "sentence_expander",
            "joiner",
            "ml_inference",
        ]
        extra_kwargs = {
            "debug": {"initial": True},
        }






class NLPSerializer(NLPSettingSerializer):
    input_text = serializers.CharField(required=True)



    class Meta(NLPSettingSerializer.Meta):
        fields = ["input_text", *NLPSettingSerializer.Meta.fields]




    def to_internal_value(self, data):
        data = super().to_internal_value(data)

        if data.get("input_text"):
            data["input_text"] = codecs.decode(data["input_text"], "unicode_escape")

        return data






class NlpProcessJobSerializer(serializers.ModelSerializer):
    job_progress = serializers.ReadOnlyField()
    username = serializers.CharField(source="user.username", read_only=True)



    class Meta:
        model = NlpProcessJob
        fields = "__all__"
        read_only_fields = [
            "job_status",
            "error_message",
            "created_at",
            "updated_at",
            "total_count",
            "completed_count",
            "user",
        ]






class DemoSerializer(serializers.Serializer):
    input_text = serializers.CharField(required=True)



    def to_internal_value(self, data):
        data = super().to_internal_value(data)

        if data.get("input_text"):
            data["input_text"] = codecs.decode(data["input_text"], "unicode_escape")

        return data