# 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 httpx
from django.conf import settings
from rest_framework import status
from rest_framework.permissions import BasePermission
from rest_framework.permissions import IsAuthenticated
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework.viewsets import ModelViewSet
from rest_framework.viewsets import ViewSet
from nlpmed_portal.nlp import task_dispatch
from nlpmed_portal.nlp.api.serializers import DemoSerializer
from nlpmed_portal.nlp.api.serializers import NlpProcessJobSerializer
from nlpmed_portal.nlp.api.serializers import NLPSerializer
from nlpmed_portal.nlp.constants import BLEEDING_KEYWORD_EXC_LIST
from nlpmed_portal.nlp.constants import BLEEDING_KEYWORD_INC_LIST
from nlpmed_portal.nlp.constants import BLEEDING_SECTION_EXC_LIST
from nlpmed_portal.nlp.constants import BLEEDING_SECTION_INC_LIST
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 NlpProcessJob
[docs]
class MethodBasePermission(BasePermission):
"""Enforce HTTP-method-specific permissions declared by a view.
Views using this permission class must define ``required_perms``, mapping
HTTP method names to Django permission codenames.
"""
[docs]
def has_permission(self, request: Request, view: APIView) -> bool:
required_perms = getattr(view, "required_perms", {})
if not required_perms:
return False
required_perm = required_perms.get(request.method)
if not required_perm:
return False
return request.user.has_perm(required_perm)
[docs]
class NLPViewSet(ViewSet):
serializer_class = NLPSerializer
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "nlp.view_nlp",
"HEAD": "nlp.view_nlp",
"OPTIONS": "nlp.view_nlp",
"POST": "nlp.add_nlp",
}
[docs]
def create(self, request):
serializer = self.serializer_class(data=request.data)
if serializer.is_valid(raise_exception=True):
validated = serializer.validated_data
input_text = validated.get("input_text")
debug = validated.get("debug")
components = {}
for key in [
"encoding_fixer",
"pattern_replacer",
"word_masker",
"note_filter",
"section_splitter",
"section_filter",
"sentence_segmenter",
"duplicate_checker",
"sentence_filter",
"sentence_expander",
"joiner",
"ml_inference",
]:
component = validated.get(key)
if component:
components[key] = component
try:
response = httpx.post(
f"{settings.NLP_API_URL}/process_text",
json={
"input_data": {"text": input_text},
"config": {**components, "debug": debug},
},
timeout=20,
verify=False, # ruff: ignore[request-with-no-cert-validation]
)
return Response(response.json(), status=response.status_code)
except httpx.ConnectError:
return Response(
{
"error": "Cannot connect to the NLP API. \
Make sure the API is running.",
},
status=status.HTTP_503_SERVICE_UNAVAILABLE,
)
except Exception as e: # ruff: ignore[blind-except]
return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
return Response({"errors": serializer.errors}, status=status.HTTP_400_BAD_REQUEST)
[docs]
class NlpProcessJobViewSet(ModelViewSet):
queryset = NlpProcessJob.objects.select_related("user", "project").all()
serializer_class = NlpProcessJobSerializer
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "nlp.view_nlpprocessjob",
"HEAD": "nlp.view_nlpprocessjob",
"OPTIONS": "nlp.view_nlpprocessjob",
"POST": "nlp.add_nlpprocessjob",
}
[docs]
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
job: NlpProcessJob = serializer.save(user=request.user)
task_dispatch.dispatch_nlp_job(job.id)
return Response(serializer.data, status=status.HTTP_200_OK)
[docs]
class VteBertDemoViewSet(ViewSet):
serializer_class = DemoSerializer
permission_classes = []
[docs]
def create(self, request):
serializer = self.serializer_class(data=request.data)
if serializer.is_valid(raise_exception=True):
validated = serializer.validated_data
input_text = validated.get("input_text")
config = {
"encoding_fixer": {
"status": "enabled",
},
"pattern_replacer": {
"status": "enabled",
"pattern": r"(?:\s*\n\s*){2,}",
"target": "\n\n",
},
"word_masker": {
"status": "enabled",
"words_to_mask": VTE_KEYWORD_EXC_LIST,
"mask_char": "*",
},
"note_filter": {
"status": "enabled",
"words_to_search": VTE_KEYWORD_INC_LIST,
},
"section_splitter": {
"status": "enabled",
"delimiter": "\n\n",
},
"section_filter": {
"status": "enabled",
"section_inc_list": VTE_SECTION_INC_LIST,
"section_exc_list": VTE_SECTION_EXC_LIST,
"fallback": True,
},
"sentence_segmenter": {
"status": "enabled",
"model_name": "en_core_sci_lg",
"batch_size": 10,
},
"duplicate_checker": {
"status": "disabled",
},
"sentence_filter": {
"status": "enabled",
"words_to_search": VTE_KEYWORD_INC_LIST,
},
"sentence_expander": {
"status": "enabled",
"length_threshold": 50,
},
"joiner": {
"status": "enabled",
"sentence_delimiter": "\n",
"section_delimiter": "\n\n",
},
"ml_inference": {
"status": "enabled",
"model_name": "VTE_MULTICLASS",
"use_preped_text": True,
},
"debug": True,
}
try:
response = httpx.post(
f"{settings.NLP_API_URL}/process_text",
json={
"input_data": {"text": input_text},
"config": config,
},
timeout=20,
verify=False, # ruff: ignore[request-with-no-cert-validation]
)
return Response(
response.json().get("predicted_label", "") or "none",
status=response.status_code,
)
except httpx.ConnectError:
return Response(
{
"error": "Cannot connect to the NLP API. \
Make sure the API is running.",
},
status=status.HTTP_503_SERVICE_UNAVAILABLE,
)
except Exception as e: # ruff: ignore[blind-except]
return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
return Response({"errors": serializer.errors}, status=status.HTTP_400_BAD_REQUEST)
[docs]
class BleedBertDemoViewSet(ViewSet):
serializer_class = DemoSerializer
permission_classes = []
[docs]
def create(self, request):
serializer = self.serializer_class(data=request.data)
if serializer.is_valid(raise_exception=True):
validated = serializer.validated_data
input_text = validated.get("input_text")
config = {
"encoding_fixer": {
"status": "enabled",
},
"pattern_replacer": {
"status": "enabled",
"pattern": r"(?:\s*\n\s*){2,}",
"target": "\n\n",
},
"word_masker": {
"status": "enabled",
"words_to_mask": BLEEDING_KEYWORD_EXC_LIST,
"mask_char": "*",
},
"note_filter": {
"status": "enabled",
"words_to_search": BLEEDING_KEYWORD_INC_LIST,
},
"section_splitter": {
"status": "enabled",
"delimiter": "\n\n",
},
"section_filter": {
"status": "enabled",
"section_inc_list": BLEEDING_SECTION_INC_LIST,
"section_exc_list": BLEEDING_SECTION_EXC_LIST,
"fallback": True,
},
"sentence_segmenter": {
"status": "enabled",
"model_name": "en_core_sci_lg",
"batch_size": 10,
},
"duplicate_checker": {
"status": "disabled",
},
"sentence_filter": {
"status": "enabled",
"words_to_search": BLEEDING_KEYWORD_INC_LIST,
},
"sentence_expander": {
"status": "enabled",
"length_threshold": 50,
},
"joiner": {
"status": "enabled",
"sentence_delimiter": "\n",
"section_delimiter": "\n\n",
},
"ml_inference": {
"status": "enabled",
"model_name": "BLEED_BINARY",
"use_preped_text": True,
},
"debug": True,
}
label_map = {
"0": "none",
"1": "bleed",
}
try:
response = httpx.post(
f"{settings.NLP_API_URL}/process_text",
json={
"input_data": {"text": input_text},
"config": config,
},
timeout=20,
verify=False, # ruff: ignore[request-with-no-cert-validation]
)
predicted_label = label_map.get(
str(response.json().get("predicted_label")),
"none",
)
return Response(
predicted_label,
status=response.status_code,
)
except httpx.ConnectError:
return Response(
{
"error": "Cannot connect to the NLP API. \
Make sure the API is running.",
},
status=status.HTTP_503_SERVICE_UNAVAILABLE,
)
except Exception as e: # ruff: ignore[blind-except]
return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
return Response({"errors": serializer.errors}, status=status.HTTP_400_BAD_REQUEST)