# 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 NlpProcessJobSerializer
from nlpmed_portal.nlp.api.serializers import NLPSerializer
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)