Игорь Надеин
e72a039744
All checks were successful
Deploy production / deploy (push) Successful in 10m51s
305 lines
11 KiB
Python
305 lines
11 KiB
Python
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,
|
||
)
|