import re from django.contrib.auth import get_user_model from django.contrib.auth.password_validation import validate_password from rest_framework import serializers from rest_framework_simplejwt.serializers import TokenObtainPairSerializer User = get_user_model() def _derive_username(email: str) -> str: local = email.split("@", 1)[0] candidate = re.sub(r"[^\w.@+-]", "_", local).strip("._") return candidate[:150] if candidate else "user" def _unique_username(base: str) -> str: username = base[:150] if not User.objects.filter(username=username).exists(): return username suffix = 1 while User.objects.filter(username=f"{username[:140]}_{suffix}").exists(): suffix += 1 return f"{username[:140]}_{suffix}" class RegisterSerializer(serializers.Serializer): email = serializers.EmailField() password = serializers.CharField(write_only=True, min_length=8) def validate_email(self, value: str) -> str: email = value.lower() if User.objects.filter(email__iexact=email).exists(): raise serializers.ValidationError("A user with this email already exists.") return email def validate_password(self, value: str) -> str: validate_password(value) return value def create(self, validated_data: dict) -> User: email = validated_data["email"] username = _unique_username(_derive_username(email)) return User.objects.create_user( username=username, email=email, password=validated_data["password"], ) def to_representation(self, instance: User) -> dict: return { "id": instance.id, "email": instance.email, "username": instance.username, } class EmailTokenObtainPairSerializer(TokenObtainPairSerializer): def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) self.fields.pop(self.username_field, None) self.fields["email"] = serializers.EmailField(required=True) def validate(self, attrs: dict) -> dict: email = attrs.get("email", "").lower() password = attrs.get("password") try: user = User.objects.get(email__iexact=email) except User.DoesNotExist as exc: raise serializers.ValidationError( {"detail": "No active account found with the given credentials."} ) from exc if not user.check_password(password): raise serializers.ValidationError( {"detail": "No active account found with the given credentials."} ) if not user.is_active: raise serializers.ValidationError({"detail": "User account is disabled."}) refresh = self.get_token(user) return { "refresh": str(refresh), "access": str(refresh.access_token), } @classmethod def get_token(cls, user: User) -> object: token = super().get_token(user) token["email"] = user.email return token