from __future__ import annotations import logging from django.core.exceptions import SuspiciousFileOperation from django.core.files.base import ContentFile from django.db import transaction from django.utils import timezone from django.utils.crypto import get_random_string from django.utils.decorators import method_decorator from django.views.decorators.cache import never_cache from drf_spectacular.utils import ( OpenApiResponse, extend_schema, inline_serializer, ) from rest_framework import serializers, status from rest_framework.parsers import FormParser, JSONParser, MultiPartParser from rest_framework.response import Response from api.v1.employees.serializers import ( RegisterSerializer, ) from api.v1.employees.views._helpers import Employee from api.v1.employees.views.mixins import LdapUserCreationMixin from common.emails import send_templated_mail from common.ip_restrictions import RegistrationFromLAN from employees.models import Position, Skill from employees.utils import _normalize_phone from ..base import AnonymousAPIView logger = logging.getLogger(__name__) @method_decorator(never_cache, name="dispatch") class RegistrationAvailabilityAPIView(AnonymousAPIView): @extend_schema( tags=["Auth"], summary="Проверить доступность регистрации для текущего посетителя", responses={ 200: inline_serializer( "RegistrationAvailabilityResponse", {"registration_allowed": serializers.BooleanField()}, ) }, ) def get(self, request): allowed = RegistrationFromLAN().has_permission(request, self) return Response({"registration_allowed": allowed}) class RegisterAPIView(LdapUserCreationMixin, AnonymousAPIView): permission_classes = [RegistrationFromLAN] throttle_scope = "registration" parser_classes = (JSONParser, FormParser, MultiPartParser) def is_ldap_enabled(self) -> bool: # Сохраняем старую точку patch'а тестов: # api.v1.employees.views._is_ldap_enabled from api.v1.employees import views as employee_views return employee_views._is_ldap_enabled() @staticmethod def _invalid_phone_response() -> Response: logger.error( "Registration phone normalization failed", extra={ "event": "registration.denied", "reason": "invalid_phone", }, ) return Response({"ok": False, "error": "invalid_phone"}, status=400) @staticmethod def _phone_taken_response(existing_user: Employee) -> Response: logger.warning( "Registration denied", extra={ "event": "registration.denied", "reason": "phone_taken", "existing_user_id": existing_user.id, }, ) return Response( { "ok": False, "error": "phone_taken", "detail": ( "Номер телефона уже зарегистрирован. " "Войдите в существующий аккаунт " "или используйте другой номер." ), "phone_number": [ "Этот номер телефона уже привязан к другому аккаунту." ], }, status=400, ) @staticmethod def _existing_email_response(existing_user: Employee) -> Response: logger.warning( "Registration matched existing user", extra={ "event": "registration.existing_user", "existing_user_id": existing_user.id, "email_verified": existing_user.email_verified, }, ) if existing_user.email_verified: logger.warning( "Registration denied", extra={ "event": "registration.denied", "reason": "email_taken", "existing_user_id": existing_user.id, }, ) return Response({"ok": False, "error": "email_taken"}, status=400) return Response( { "ok": True, "pending_verification": True, "user_id": existing_user.id, }, status=200, ) @staticmethod def _registration_conflict(email: str, phone: str) -> Response | None: existing_phone = Employee.objects.filter(phone_number=phone).first() if existing_phone: return RegisterAPIView._phone_taken_response(existing_phone) existing_email = Employee.objects.filter(email__iexact=email).first() if existing_email: return RegisterAPIView._existing_email_response(existing_email) return None @staticmethod def _read_avatar(avatar_file) -> tuple[bytes | None, str | None]: if not avatar_file: return None, None try: avatar_bytes = avatar_file.read() if hasattr(avatar_file, "read") else None avatar_name = getattr(avatar_file, "name", None) or "avatar.jpg" return avatar_bytes, avatar_name except (OSError, ValueError): logger.exception( "Registration avatar read failed", extra={"event": "registration.avatar_read.failed"}, ) return None, None @staticmethod def _save_avatar(employee: Employee, avatar_bytes, avatar_name) -> None: if not avatar_bytes: return try: employee.avatar.save( avatar_name, ContentFile(avatar_bytes), save=False, ) except (OSError, ValueError, SuspiciousFileOperation): logger.exception( "Registration avatar save failed", extra={ "event": "registration.avatar_save.failed", "employee_id": employee.pk, }, ) @staticmethod def _apply_profile(employee: Employee, validated_data) -> None: employee.telegram = validated_data.get("telegram", "") employee.whatsapp = validated_data.get("whatsapp", "") employee.wechat = validated_data.get("wechat", "") employee.birth_date = validated_data["birth_date"] employee.gender = validated_data.get("gender") if validated_data.get("patronymic"): employee.patronymic = validated_data["patronymic"] position_id = validated_data.get("position") if position_id and Position.objects.filter(pk=position_id).exists(): employee.position_id = position_id employee.save() skill_ids = validated_data.get("skills") or [] if skill_ids: employee.skills.set(Skill.objects.filter(pk__in=skill_ids)) @staticmethod def _send_verification(employee: Employee) -> None: employee.email_activation_code = get_random_string(6, "0123456789") employee.email_activation_sent_at = timezone.now() employee.save( update_fields=[ "email_activation_code", "email_activation_sent_at", ] ) send_templated_mail( subject="Подтверждение регистрации", to=[employee.email], template_base="emails/registration_verify_code", context={"code": employee.email_activation_code, "user": employee}, ) @extend_schema( tags=["Auth"], summary="Зарегистрировать нового пользователя", request=RegisterSerializer, responses={ 201: inline_serializer( "RegisterResponse", { "id": serializers.IntegerField(), "email": serializers.EmailField(), "email_verified": serializers.BooleanField(), "is_active": serializers.BooleanField(), }, ), 200: inline_serializer( "RegisterPendingVerificationResponse", { "ok": serializers.BooleanField(), "pending_verification": serializers.BooleanField(), "user_id": serializers.IntegerField(), }, ), 400: OpenApiResponse( description="Ошибка валидации или занятый email/телефон." ), 403: OpenApiResponse( description="Регистрация разрешена только из локальной сети компании." ), }, ) @transaction.atomic def post(self, request): logger.info( "Registration attempted", extra={ "event": "registration.attempted", "content_type": request.content_type, }, ) ser = RegisterSerializer(data=request.data) if not ser.is_valid(): logger.warning( "Registration validation failed", extra={ "event": "registration.validation_failed", "validation_fields": sorted(ser.errors.keys()), }, ) ser.is_valid(raise_exception=True) validated_data = ser.validated_data email = validated_data["email"].strip().lower() phone_norm = _normalize_phone( validated_data.get("phone_number") or request.data.get("phone") ) if not phone_norm: return self._invalid_phone_response() conflict_response = self._registration_conflict(email, phone_norm) if conflict_response: return conflict_response avatar_file = validated_data.get("avatar") or getattr( request, "FILES", {}, ).get("avatar") avatar_bytes, avatar_name = self._read_avatar(avatar_file) emp, error_response = self.create_user( first_name=validated_data["first_name"], last_name=validated_data["last_name"], email=email, phone=phone_norm, password=validated_data["password"], avatar_bytes=avatar_bytes, is_active=False, ) if error_response: return error_response self._save_avatar(emp, avatar_bytes, avatar_name) self._apply_profile(emp, validated_data) self._send_verification(emp) return Response( { "id": emp.id, "email": emp.email, "email_verified": emp.email_verified, "is_active": emp.is_active, }, status=status.HTTP_201_CREATED, )