[Eng-32] feat :create otp based reset for password

care/emr/api/otp_viewsets/login.py

@@ -12,20 +12,57 @@
 from rest_framework.response import Response
 
 from care.emr.api.viewsets.base import EMRBaseViewSet
-from care.facility.models.patient import PatientMobileOTP
+from care.facility.models.patient import MobileOTP
 from care.utils import sms
 from care.utils.models.validators import mobile_validator
-from care.utils.sms.utils import get_sms_content
 from config.patient_otp_token import PatientToken
 
 logger = logging.getLogger(__name__)
 
 
+class OTPType:
+    def render_content(self, otp: str) -> str:
+        pass
+
+
+class LoginOTP(OTPType):
+    def render_content(self, otp: str) -> str:
+        return settings.OTP_SMS_LOGIN_CONTENT.format(otp=otp)
+
+
 def rand_pass(size):
     return "".join(secrets.choice(string.digits) for _ in range(size))
 
 
-class OTPLoginRequestSpec(BaseModel):
+def send_otp(phone_number, otp_type: OTPType):
+    sent_otps = MobileOTP.objects.filter(
+        created_date__gte=(timezone.now() - timedelta(settings.OTP_REPEAT_WINDOW)),
+        is_used=False,
+        phone_number=phone_number,
+    )
+    if sent_otps.count() >= settings.OTP_MAX_REPEATS_WINDOW:
+        raise ValueError("Max Retries has exceeded")
+
+    random_otp = ""
+    if settings.USE_SMS:
+        random_otp = rand_pass(settings.OTP_LENGTH)
+        try:
+            content = otp_type.render_content(random_otp)
+            sms.send_text_message(
+                content=content,
+                recipients=[phone_number],
+            )
+        except Exception as e:
+            raise Exception("Error while sending OTP. Contact admin.") from e
+    elif settings.IS_PRODUCTION:
+        random_otp = rand_pass(settings.OTP_LENGTH)
+    else:
+        random_otp = "45612"
+
+    MobileOTP.objects.create(phone_number=phone_number, otp=random_otp)
+
+
+class OTPRequestBaseSpec(BaseModel):
     phone_number: str
 
     @field_validator("phone_number")
@@ -39,7 +76,7 @@
         return value
 
 
-class OTPLoginSpec(OTPLoginRequestSpec):
+class OTPLoginSpec(OTPRequestBaseSpec):
     otp: str = Field(min_length=settings.OTP_LENGTH, max_length=settings.OTP_LENGTH)
 
 
@@ -48,41 +85,17 @@
     permission_classes = []
 
     @extend_schema(
-        request=OTPLoginRequestSpec,
+        request=OTPRequestBaseSpec,
     )
     @action(detail=False, methods=["POST"])
     def send(self, request):
-        data = OTPLoginRequestSpec(**request.data)
-        sent_otps = PatientMobileOTP.objects.filter(
-            created_date__gte=(timezone.now() - timedelta(settings.OTP_REPEAT_WINDOW)),
-            is_used=False,
-            phone_number=data.phone_number,
-        )
-        if sent_otps.count() >= settings.OTP_MAX_REPEATS_WINDOW:
-            raise ValidationError({"phone_number": "Max Retries has exceeded"})
-        random_otp = ""
-        if settings.USE_SMS:
-            random_otp = rand_pass(settings.OTP_LENGTH)
-            try:
-                content = get_sms_content(
-                    settings.OTP_SMS_TEMPLATE_PATH, {"random_otp": random_otp}
-                )
-                sms.send_text_message(
-                    content=content,
-                    recipients=[data.phone_number],
-                )
-            except Exception as e:
-                logger.error(e)
-                return Response(
-                    {"error": "Error while sending OTP. Contact admin."}, status=400
-                )
-        elif settings.IS_PRODUCTION:
-            random_otp = rand_pass(settings.OTP_LENGTH)
-        else:
-            random_otp = "45612"
-
-        otp_obj = PatientMobileOTP(phone_number=data.phone_number, otp=random_otp)
-        otp_obj.save()
+        data = OTPRequestBaseSpec(**request.data)
+        try:
+            send_otp(data.phone_number, otp_type=LoginOTP())
+        except ValueError as e:
+            raise ValidationError({"phone_number": str(e)}) from e
+        except Exception as e:
+            return Response({"error": str(e)}, status=400)
         return Response({"otp": "generated"})
 
     @extend_schema(
@@ -91,7 +104,7 @@
     @action(detail=False, methods=["POST"])
     def login(self, request):
         data = OTPLoginSpec(**request.data)
-        otp_object = PatientMobileOTP.objects.filter(
+        otp_object = MobileOTP.objects.filter(
             phone_number=data.phone_number, otp=data.otp, is_used=False
         ).first()
         if not otp_object:

care/facility/migrations/0485_rename_patientmobileotp_mobileotp.py

@@ -0,0 +1,17 @@
+# Generated by Django 6.0 on 2026-05-05 17:41
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('facility', '0484_remove_facility_discount_codes_and_more'),
+    ]
+
+    operations = [
+        migrations.RenameModel(
+            old_name='PatientMobileOTP',
+            new_name='MobileOTP',
+        ),
+    ]

care/facility/models/patient.py

@@ -4,7 +4,7 @@
 from care.utils.models.validators import mobile_or_landline_number_validator
 
 
-class PatientMobileOTP(BaseModel):
+class MobileOTP(BaseModel):
     is_used = models.BooleanField(default=False)
     phone_number = models.CharField(
         max_length=14, validators=[mobile_or_landline_number_validator]

care/users/api/otp_viewset/reset_password.py

@@ -0,0 +1,105 @@
+from datetime import timedelta
+
+from django.conf import settings
+from django.contrib.auth.password_validation import (
+    get_password_validators,
+    validate_password,
+)
+from django.utils import timezone
+from drf_spectacular.utils import extend_schema
+from pydantic import Field
+from rest_framework.decorators import action
+from rest_framework.exceptions import ValidationError
+from rest_framework.response import Response
+
+from care.emr.api.otp_viewsets.login import OTPRequestBaseSpec, OTPType, send_otp
+from care.emr.api.viewsets.base import EMRBaseViewSet
+from care.facility.models.patient import MobileOTP
+from care.users.models import User
+
+
+class OTPResetSendSpec(OTPRequestBaseSpec):
+    pass
+
+
+class ResetPasswordOTP(OTPType):
+    def render_content(self, otp: str) -> str:
+        return settings.OTP_SMS_RESET_PASSWORD_CONTENT.format(otp=otp)
+
+
+class OTPResetConfirmSpec(OTPRequestBaseSpec):
+    otp: str = Field(min_length=settings.OTP_LENGTH, max_length=settings.OTP_LENGTH)
+    password: str = Field(min_length=8)
+    username: str | None = None
+
+
+class OTPResetPasswordView(EMRBaseViewSet):
+    authentication_classes = []
+    permission_classes = []
+
+    @action(detail=False, methods=["POST"])
+    @extend_schema(request=OTPResetSendSpec)
+    def send(self, request):
+        data = OTPResetSendSpec(**request.data)
+
+        if not User.objects.filter(phone_number=data.phone_number).exists():
+            return Response({"otp": "generated"})
+
+        try:
+            send_otp(data.phone_number, otp_type=ResetPasswordOTP())
+        except ValueError as e:
+            raise ValidationError({"phone_number": str(e)}) from e
+        except Exception as e:
+            return Response({"error": str(e)}, status=400)
+        return Response({"otp": "generated"})
+
+    @action(detail=False, methods=["POST"])
+    @extend_schema(request=OTPResetConfirmSpec)
+    def confirm(self, request):
+        data = OTPResetConfirmSpec(**request.data)
+        otp_obj = (
+            MobileOTP.objects.filter(
+                phone_number=data.phone_number,
+                is_used=False,
+                created_date__gte=(
+                    timezone.now() - timedelta(hours=settings.OTP_REPEAT_WINDOW)
+                ),
+            )
+            .order_by("-created_date")
+            .first()
+        )
+        if not otp_obj or otp_obj.otp != data.otp:
+            raise ValidationError({"otp": "Invalid OTP"})
+
+        users = User.objects.filter(phone_number=data.phone_number)
+        if not users.exists():
+            raise ValidationError({"error": "No User linked to this phone number"})
+        if users.count() > 1:
+            if data.username:
+                users = users.filter(username=data.username)
+                if not users.exists():
+                    raise ValidationError(
+                        {
+                            "error": "No User with this username linked to this phone number"
+                        }
+                    )
+            else:
+                return Response(
+                    {"error": "Multiple users linked to this phone number"},
+                    status=409,
+                )
+        user = users.first()
+
+        validate_password(
+            data.password,
+            user=user,
+            password_validators=get_password_validators(
+                settings.AUTH_PASSWORD_VALIDATORS
+            ),
+        )
+        user.set_password(data.password)
+        user.save()
+        MobileOTP.objects.filter(
+            phone_number=data.phone_number,
+        ).delete()
+        return Response({"message": "Password reset successful"})

config/api_router.py

@@ -109,6 +109,7 @@
 from care.emr.api.viewsets.valueset import ValueSetViewSet
 from care.security.api.viewsets.permissions import PermissionViewSet
 from care.security.api.viewsets.roles import RoleViewSet
+from care.users.api.otp_viewset.reset_password import OTPResetPasswordView
 from care.users.api.viewsets.plug_config import PlugConfigViewset
 
 router = DefaultRouter() if settings.DEBUG else SimpleRouter()
@@ -123,6 +124,9 @@
 router.register("meta_artifacts", MetaArtifactViewSet, basename="meta_artifacts")
 
 router.register("otp", OTPLoginView, basename="otp-login")
+router.register(
+    "otp/password_reset", OTPResetPasswordView, basename="otp-password-reset"
+)
 
 router.register("otp/patient", PatientOTPView, basename="otp-patient")
 

config/settings/base.py

@@ -690,7 +690,15 @@
 
 SMS_BACKEND = "care.utils.sms.backend.sns.SnsBackend"
 
-OTP_SMS_TEMPLATE_PATH = env("OTP_SMS_TEMPLATE", default="sms/otp_sms.txt")
+OTP_SMS_LOGIN_CONTENT = env(
+    "OTP_SMS_LOGIN_CONTENT",
+    default="Care OTP for login is {otp}. Please do not share this with anyone.",
+)
+
+OTP_SMS_RESET_PASSWORD_CONTENT = env(
+    "OTP_SMS_RESET_PASSWORD_CONTENT",
+    default="Care OTP for password reset is {otp}. Please do not share this with anyone.",
+)
 
 USER_CREATE_PASSWORD_EMAIL_TEMPLATE_PATH = env(
     "USER_CREATE_PASSWORD_TEMPLATE_PATH", default="email/user_create_password.html"