from django.contrib.auth import get_user_model from django.contrib.auth.password_validation import validate_password from django.utils.translation import gettext_lazy as _ from rest_framework import serializers from rest_framework_simplejwt.tokens import RefreshToken from rest_framework_simplejwt.serializers import TokenObtainPairSerializer from phonenumber_field.serializerfields import PhoneNumberField from drf_spectacular.utils import extend_schema_field, OpenApiTypes from apps.account.models.verification import VerificationDocument User = get_user_model() class RegisterRequestSerializer(serializers.Serializer): email = serializers.EmailField(required=True) password = serializers.CharField(write_only=True, required=True, min_length=6) first_name = serializers.CharField(required=False, allow_blank=True, max_length=150) last_name = serializers.CharField(required=False, allow_blank=True, max_length=150) fullname = serializers.CharField(required=False, allow_blank=True, max_length=255) phone_number = serializers.CharField(required=False, allow_blank=True, allow_null=True) user_type = serializers.ChoiceField( choices=User.UserType.choices, default=User.UserType.CLIENT, required=False ) languages = serializers.ListField( child=serializers.CharField(max_length=10), required=False, default=list ) skills = serializers.ListField( child=serializers.CharField(max_length=100), required=False, default=list ) country = serializers.CharField(required=False, allow_blank=True, max_length=255) city = serializers.CharField(required=False, allow_blank=True, max_length=255) def validate_email(self, value): normalized = value.strip().lower() if User.objects.filter(email__iexact=normalized).exists(): raise serializers.ValidationError("A user with this email address already exists.") return normalized def validate_password(self, value): validate_password(value) return value def validate(self, attrs): fn = attrs.get('first_name', '').strip() ln = attrs.get('last_name', '').strip() fl = attrs.get('fullname', '').strip() if fn or ln: attrs['first_name'] = fn attrs['last_name'] = ln if not fl: attrs['fullname'] = f"{fn} {ln}".strip() else: attrs['fullname'] = fl elif fl: parts = fl.split(maxsplit=1) attrs['first_name'] = parts[0] attrs['last_name'] = parts[1] if len(parts) > 1 else "" attrs['fullname'] = fl else: raise serializers.ValidationError({"first_name": "Please provide first_name and last_name (or full name)."}) return attrs def create(self, validated_data): email = validated_data.pop('email') password = validated_data.pop('password') first_name = validated_data.pop('first_name', '') last_name = validated_data.pop('last_name', '') fullname = validated_data.pop('fullname', '') user_type = validated_data.pop('user_type', User.UserType.CLIENT) user = User.objects.create_user( email=email, password=password, first_name=first_name, last_name=last_name, fullname=fullname, username=email, user_type=user_type, **validated_data ) return user class UserMeSerializer(serializers.ModelSerializer): bio = serializers.CharField(source='info', allow_blank=True, required=False, allow_null=True) role = serializers.SerializerMethodField() roles = serializers.SerializerMethodField() permissions = serializers.SerializerMethodField() permission_flags = serializers.SerializerMethodField() managed_regions = serializers.SerializerMethodField() regional_permissions = serializers.ReadOnlyField() verification_status = serializers.SerializerMethodField() is_verified = serializers.SerializerMethodField() first_name = serializers.SerializerMethodField() last_name = serializers.SerializerMethodField() name = serializers.SerializerMethodField() phone = serializers.SerializerMethodField() preferred_language = serializers.SerializerMethodField() title = serializers.SerializerMethodField() rank = serializers.SerializerMethodField() website = serializers.SerializerMethodField() address = serializers.SerializerMethodField() position = serializers.SerializerMethodField() institution = serializers.SerializerMethodField() institution_id = serializers.SerializerMethodField() two_factor_enabled = serializers.SerializerMethodField() two_factor_method = serializers.SerializerMethodField() is_2fa_forced_by_admin = serializers.SerializerMethodField() require_2fa = serializers.SerializerMethodField() has_totp_configured = serializers.SerializerMethodField() backup_codes_count = serializers.SerializerMethodField() class Meta: model = User fields = [ 'id', 'email', 'username', 'fullname', 'first_name', 'last_name', 'name', 'phone_number', 'phone', 'user_type', 'avatar', 'bio', 'languages', 'preferred_language', 'skills', 'title', 'rank', 'position', 'institution', 'institution_id', 'website', 'address', 'country', 'city', 'gender', 'birthdate', 'date_joined', 'is_active', 'is_staff', 'role', 'roles', 'permissions', 'permission_flags', 'managed_regions', 'regional_permissions', 'verification_status', 'is_verified', 'two_factor_enabled', 'two_factor_method', 'is_2fa_forced_by_admin', 'require_2fa', 'has_totp_configured', 'backup_codes_count', ] read_only_fields = [ 'id', 'email', 'username', 'first_name', 'last_name', 'name', 'phone', 'preferred_language', 'title', 'rank', 'position', 'institution', 'institution_id', 'website', 'address', 'user_type', 'date_joined', 'is_active', 'is_staff', 'roles', 'permissions', 'managed_regions', 'regional_permissions', 'verification_status', 'is_verified', 'two_factor_enabled', 'two_factor_method', 'is_2fa_forced_by_admin', 'require_2fa', 'has_totp_configured', 'backup_codes_count', ] def get_first_name(self, obj) -> str: if obj.first_name: return obj.first_name if obj.fullname: return obj.fullname.strip().split()[0] return "" def get_last_name(self, obj) -> str: if obj.last_name: return obj.last_name if obj.fullname: parts = obj.fullname.strip().split() if len(parts) > 1: return " ".join(parts[1:]) return "" def get_name(self, obj) -> str: return obj.fullname or f"{obj.first_name or ''} {obj.last_name or ''}".strip() def get_phone(self, obj) -> str: if obj.phone_number: raw = (getattr(obj.phone_number, 'as_e164', None) or str(obj.phone_number)).replace(' ', '') if raw.startswith('+98') and len(raw) == 13: return '0' + raw[3:] return str(obj.phone_number) return "" def get_preferred_language(self, obj) -> str: if obj.languages and isinstance(obj.languages, list) and len(obj.languages) > 0: return obj.languages[0] return obj.language or "fa" def get_title(self, obj) -> str: if obj.title: return obj.title membership = obj.institution_memberships.first() if hasattr(obj, 'institution_memberships') else None if membership and membership.title: return membership.title return obj.skill or "" def get_rank(self, obj) -> str: return obj.rank or "" def get_website(self, obj) -> str: return obj.website or "" def get_address(self, obj) -> str: return obj.address or "" def get_institution(self, obj) -> str: if obj.institution_name: return obj.institution_name membership = obj.institution_memberships.first() if hasattr(obj, 'institution_memberships') else None if membership and membership.institution: return membership.institution.name return "" @extend_schema_field(serializers.IntegerField(allow_null=True)) def get_institution_id(self, obj): membership = obj.institution_memberships.first() if hasattr(obj, 'institution_memberships') else None if membership and membership.institution: return membership.institution.id return None def get_position(self, obj) -> str: membership = obj.institution_memberships.first() if hasattr(obj, 'institution_memberships') else None if membership and membership.role: return membership.role return "" def get_role(self, obj): if obj.role: return { 'id': obj.role.id, 'name': obj.role.name, 'slug': obj.role.slug, 'description': obj.role.description, } return None def get_roles(self, obj) -> list: roles = [obj.user_type] if obj.user_type else [] if obj.role and obj.role.slug: roles.append(obj.role.slug) for group in obj.groups.all(): roles.append(group.name) return list(set(roles)) def get_permission_flags(self, obj) -> dict: """ Returns full boolean flags dictionary dynamically resolved from Role and custom_permissions. """ if not obj or not getattr(obj, 'is_active', False): return {} if getattr(obj, 'is_superuser', False): from apps.account.models.role import ALL_PERMISSION_FIELDS flags = {perm: True for perm in ALL_PERMISSION_FIELDS} flags['require_2fa'] = obj.is_2fa_forced_by_admin() return flags flags = obj.role.get_permissions_dict() if obj.role else {} if obj.custom_permissions and isinstance(obj.custom_permissions, dict): flags.update(obj.custom_permissions) flags['require_2fa'] = obj.is_2fa_forced_by_admin() return flags def get_permissions(self, obj) -> list: """ Returns active permission capability flags where boolean value is True, dynamically derived from the user's assigned database Role and custom overrides. """ if not obj or not getattr(obj, 'is_active', False): return [] flags = self.get_permission_flags(obj) granted = [k for k, v in flags.items() if v] if obj.is_institution_admin and 'manage_institution_profile' not in granted: granted.append('manage_institution_profile') if obj.is_super_admin and 'super_admin' not in granted: granted.append('super_admin') return granted def get_verification_status(self, obj) -> str: return obj.verification_status def get_is_verified(self, obj) -> bool: return obj.verification_status == VerificationDocument.Status.VERIFIED def get_managed_regions(self, obj) -> list: if not getattr(obj, 'is_authenticated', False): return [] return [ { 'id': r.id, 'name': r.name, 'slug': r.slug, 'code': r.code, } for r in obj.managed_regions.filter(is_active=True) ] def get_two_factor_enabled(self, obj) -> bool: return obj.is_2fa_required() def get_two_factor_method(self, obj) -> str: return obj.get_effective_2fa_method() def get_is_2fa_forced_by_admin(self, obj) -> bool: return obj.is_2fa_forced_by_admin() def get_require_2fa(self, obj) -> bool: return obj.is_2fa_forced_by_admin() def get_has_totp_configured(self, obj) -> bool: return obj.has_totp_configured() def get_backup_codes_count(self, obj) -> int: if isinstance(obj.two_factor_backup_codes, list): return len(obj.two_factor_backup_codes) return 0 class UserMeUpdateSerializer(serializers.ModelSerializer): bio = serializers.CharField(source='info', allow_blank=True, required=False, allow_null=True) first_name = serializers.CharField(required=False, allow_blank=True) last_name = serializers.CharField(required=False, allow_blank=True) name = serializers.CharField(required=False, write_only=True, allow_blank=True) phone = serializers.CharField(required=False, write_only=True, allow_blank=True) preferred_language = serializers.CharField(required=False, write_only=True, allow_blank=True) expertise = serializers.ListField(child=serializers.CharField(), required=False, write_only=True) title = serializers.CharField(required=False, allow_blank=True) rank = serializers.CharField(required=False, allow_blank=True) institution = serializers.CharField(required=False, write_only=True, allow_blank=True) institution_name = serializers.CharField(required=False, allow_blank=True) website = serializers.CharField(required=False, allow_blank=True) address = serializers.CharField(required=False, allow_blank=True) two_factor_enabled = serializers.BooleanField(required=False) two_factor_method = serializers.ChoiceField(choices=User.TwoFactorMethod.choices, required=False) class Meta: model = User fields = [ 'fullname', 'first_name', 'last_name', 'name', 'phone_number', 'phone', 'avatar', 'bio', 'languages', 'preferred_language', 'skills', 'expertise', 'title', 'rank', 'institution', 'institution_name', 'website', 'address', 'country', 'city', 'gender', 'birthdate', 'two_factor_enabled', 'two_factor_method', ] def validate(self, attrs): attrs = super().validate(attrs) instance = getattr(self, 'instance', None) if instance and 'two_factor_enabled' in attrs and attrs['two_factor_enabled'] is False: if instance.is_2fa_forced_by_admin(): from rest_framework.exceptions import PermissionDenied raise PermissionDenied( _("Two-Factor Authentication (2FA) is enforced by an administrator and cannot be disabled.") ) if instance and attrs.get('two_factor_method') == User.TwoFactorMethod.TOTP: if not instance.has_totp_configured(): raise serializers.ValidationError({ "two_factor_method": _("Please complete Authenticator App (TOTP) setup before selecting TOTP as your active method.") }) return attrs def to_internal_value(self, data): if hasattr(data, 'copy'): mutable_data = data.copy() else: mutable_data = dict(data) # Synchronize first_name, last_name, fullname, and name fn = mutable_data.get('first_name') ln = mutable_data.get('last_name') fl = mutable_data.get('fullname') or mutable_data.get('name') if fn is not None or ln is not None: if not fl: mutable_data['fullname'] = f"{fn or ''} {ln or ''}".strip() elif fl: parts = fl.strip().split(maxsplit=1) mutable_data['first_name'] = parts[0] if parts else "" mutable_data['last_name'] = parts[1] if len(parts) > 1 else "" mutable_data['fullname'] = fl.strip() # Harmonize institution -> institution_name inst = mutable_data.get('institution') or mutable_data.get('institution_name') if inst: mutable_data['institution_name'] = inst # Harmonize title if 'title' in mutable_data and mutable_data['title']: mutable_data['title'] = mutable_data['title'] # Harmonize phone / phone_number and normalize local numbers (09... -> +989...) raw_phone = mutable_data.get('phone') or mutable_data.get('phone_number') if raw_phone: raw_phone_str = str(raw_phone).strip() if raw_phone_str.startswith('09') and len(raw_phone_str) == 11 and raw_phone_str.isdigit(): mutable_data['phone_number'] = '+98' + raw_phone_str[1:] elif raw_phone_str.startswith('0098'): mutable_data['phone_number'] = '+98' + raw_phone_str[4:] elif raw_phone_str.startswith('989') and len(raw_phone_str) == 12 and raw_phone_str.isdigit(): mutable_data['phone_number'] = '+' + raw_phone_str else: mutable_data['phone_number'] = raw_phone_str if 'expertise' in mutable_data and not mutable_data.get('skills'): mutable_data['skills'] = mutable_data['expertise'] if 'preferred_language' in mutable_data and not mutable_data.get('languages'): mutable_data['languages'] = [mutable_data['preferred_language']] # Handle avatar: support File, base64 Data URI, null (remove), and ignore existing URL strings if 'avatar' in mutable_data: avatar_val = mutable_data['avatar'] if isinstance(avatar_val, str): if avatar_val.startswith('data:image/'): try: import base64, uuid from django.core.files.base import ContentFile format_part, imgstr = avatar_val.split(';base64,', 1) ext = format_part.split('/')[-1] if ext == 'jpeg': ext = 'jpg' name_id = uuid.uuid4().hex[:8] mutable_data['avatar'] = ContentFile(base64.b64decode(imgstr), name=f"avatar_{name_id}.{ext}") except Exception: mutable_data.pop('avatar', None) elif not avatar_val.strip() or avatar_val.startswith('http') or avatar_val.startswith('/'): mutable_data.pop('avatar', None) elif avatar_val is None: mutable_data['avatar'] = None return super().to_internal_value(mutable_data) class RegisterResponseSerializer(serializers.Serializer): user = UserMeSerializer() access = serializers.CharField() refresh = serializers.CharField() message = serializers.CharField() class VerificationDocumentSerializer(serializers.ModelSerializer): document_type_display = serializers.CharField(source='get_document_type_display', read_only=True) status_display = serializers.CharField(source='get_status_display', read_only=True) class Meta: model = VerificationDocument fields = [ 'id', 'user', 'document_type', 'document_type_display', 'title', 'document_file', 'status', 'status_display', 'admin_notes', 'uploaded_at', 'reviewed_at', ] read_only_fields = [ 'id', 'user', 'document_type_display', 'status', 'status_display', 'admin_notes', 'uploaded_at', 'reviewed_at', ] class VerificationDocumentUploadSerializer(serializers.Serializer): document_type = serializers.ChoiceField( choices=VerificationDocument.DocumentType.choices, default=VerificationDocument.DocumentType.NATIONAL_ID ) title = serializers.CharField(required=False, allow_blank=True, max_length=255) document_file = serializers.FileField(required=True) class PasswordRecoverRequestSerializer(serializers.Serializer): email = serializers.EmailField(required=True) class PasswordResetRequestSerializer(serializers.Serializer): email = serializers.EmailField(required=True) token = serializers.CharField(required=True) new_password = serializers.CharField(required=True, min_length=6) def validate_new_password(self, value): validate_password(value) return value def mask_email_address(email: str) -> str: if not email or "@" not in email: return email or "" local, domain = email.split("@", 1) if len(local) <= 2: masked_local = local[0] + "*" else: masked_local = local[0] + ("*" * min(len(local) - 2, 5)) + local[-1] return f"{masked_local}@{domain}" class CustomTokenObtainPairSerializer(TokenObtainPairSerializer): """ JWT token obtain serializer that accepts either 'username' or 'email' as the credential. Enforces Two-Factor Authentication (2FA) when user.is_2fa_required() is True. """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) if 'email' in self.fields: self.fields['email'].required = False if 'username' not in self.fields: self.fields['username'] = serializers.CharField(required=False, write_only=True) def to_internal_value(self, data): if isinstance(data, dict): mutable_data = data.copy() identifier = mutable_data.get('email') or mutable_data.get('username') if identifier: mutable_data['email'] = identifier mutable_data['username'] = identifier return super().to_internal_value(mutable_data) return super().to_internal_value(data) def validate(self, attrs): identifier = attrs.get('email') or attrs.get('username') if not identifier: raise serializers.ValidationError({"email": _("This field is required.")}) attrs['email'] = identifier attrs['username'] = identifier data = super().validate(attrs) if self.user and self.user.is_2fa_required(): from utils.redis import RedisManager from apps.account.tasks import dispatch_2fa_email_otp redis_mgr = RedisManager() effective_method = self.user.get_effective_2fa_method() temp_token = redis_mgr.create_2fa_challenge( user_id=self.user.id, method=effective_method, ttl=300, ) if effective_method == User.TwoFactorMethod.EMAIL and self.user.email: otp_code = redis_mgr.generate_2fa_otp_code() redis_mgr.set_2fa_email_otp(temp_token, otp_code, ttl=300) dispatch_2fa_email_otp(self.user.email, otp_code) return { "requires_2fa": True, "temp_token": temp_token, "method": effective_method, "available_methods": self.user.get_available_2fa_methods(), "email": self.user.email or "", "masked_email": mask_email_address(self.user.email or ""), "is_forced_by_admin": self.user.is_2fa_forced_by_admin(), "is_2fa_forced_by_admin": self.user.is_2fa_forced_by_admin(), "message": _("Two-Factor Authentication (2FA) is required to complete login."), } return data # ─── Two-Factor Authentication (2FA) Serializers ───────────────────────────── class TwoFactorVerifySerializer(serializers.Serializer): temp_token = serializers.CharField(required=True, help_text="Ephemeral 2FA challenge token returned by /api/v1/auth/token/.") code = serializers.CharField(required=True, min_length=6, max_length=16, help_text="6-digit OTP code (Email or TOTP) or 8-character backup recovery code.") method = serializers.ChoiceField( choices=[ ("email", "Email OTP"), ("totp", "Authenticator App (TOTP)"), ("backup", "Backup Recovery Code"), ("backup_code", "Backup Recovery Code"), ], required=False, help_text="Verification method used. Defaults to the challenge's active method.", ) class TwoFactorResendEmailSerializer(serializers.Serializer): temp_token = serializers.CharField(required=True, help_text="Ephemeral 2FA challenge token.") class TwoFactorToggleSerializer(serializers.Serializer): enabled = serializers.BooleanField(required=True, help_text="Set True to enable 2FA or False to disable 2FA.") method = serializers.ChoiceField( choices=User.TwoFactorMethod.choices, required=False, help_text="Optional preferred 2FA method ('email' or 'totp').", ) class TwoFactorSetupInitiateSerializer(serializers.Serializer): method = serializers.ChoiceField( choices=User.TwoFactorMethod.choices, default=User.TwoFactorMethod.TOTP, required=False, help_text="2FA method to set up ('totp' or 'email').", ) class TwoFactorSetupConfirmSerializer(serializers.Serializer): method = serializers.ChoiceField( choices=User.TwoFactorMethod.choices, default=User.TwoFactorMethod.TOTP, required=False, help_text="2FA method being confirmed ('totp' or 'email').", ) code = serializers.CharField(required=True, min_length=6, max_length=8, help_text="6-digit verification code from Authenticator app or Email.") class TwoFactorDisableSerializer(serializers.Serializer): password = serializers.CharField(required=False, allow_blank=True, write_only=True, help_text="Optional current password confirmation.") class TwoFactorBackupCodesRegenerateSerializer(serializers.Serializer): password = serializers.CharField(required=False, allow_blank=True, write_only=True, help_text="Optional current password confirmation.")