500 lines
17 KiB
Python
500 lines
17 KiB
Python
import base64
|
||
import binascii
|
||
from collections import defaultdict
|
||
from io import BytesIO
|
||
|
||
from django.contrib.auth import get_user_model
|
||
from django.contrib.auth.models import Group, Permission
|
||
from django.core.files.base import ContentFile
|
||
from employees.models import (
|
||
Department,
|
||
DepartmentRole,
|
||
EmployeeAction,
|
||
EmployeeDepartment,
|
||
Position,
|
||
Skill,
|
||
)
|
||
from PIL import Image
|
||
from rest_framework import serializers
|
||
|
||
from employees.constants import ACTION_CHOICES
|
||
|
||
|
||
Employee = get_user_model()
|
||
|
||
|
||
class Base64ImageField(serializers.ImageField):
|
||
"""
|
||
Двухстороннее поле:
|
||
- input: принимает строку base64 или data URI (data:image/...;base64,xxxx)
|
||
- output: возвращает data URI (data:image/...;base64,xxxx)
|
||
Совместимо с Python 3.13 (без imghdr).
|
||
"""
|
||
|
||
default_ext = "jpg"
|
||
|
||
def _detect_ext_mime(self, content: bytes) -> tuple[str, str]:
|
||
"""Определяем расширение и MIME через Pillow."""
|
||
try:
|
||
im = Image.open(BytesIO(content))
|
||
fmt = (im.format or "").upper()
|
||
ext_map = {
|
||
"JPEG": "jpg",
|
||
"JPG": "jpg",
|
||
"PNG": "png",
|
||
"WEBP": "webp",
|
||
"GIF": "gif",
|
||
"BMP": "bmp",
|
||
"TIFF": "tiff",
|
||
"ICO": "ico",
|
||
}
|
||
ext = ext_map.get(fmt, self.default_ext)
|
||
mime = Image.MIME.get(fmt, "image/jpeg")
|
||
return ext, mime
|
||
except Exception:
|
||
return self.default_ext, "image/jpeg"
|
||
|
||
def to_internal_value(self, data):
|
||
# Принимаем None/пустое
|
||
if data in (None, "", b""):
|
||
return super().to_internal_value(data)
|
||
|
||
if isinstance(data, str):
|
||
# Срезаем префикс data:image/...;base64, если есть
|
||
if data.startswith("data:image"):
|
||
try:
|
||
_, data = data.split(";base64,", 1)
|
||
except ValueError:
|
||
raise serializers.ValidationError("Некорректный формат data URI.")
|
||
try:
|
||
decoded = base64.b64decode(data)
|
||
except (TypeError, ValueError, binascii.Error):
|
||
raise serializers.ValidationError("Невалидный base64.")
|
||
|
||
ext, _ = self._detect_ext_mime(decoded)
|
||
file = ContentFile(decoded, name=f"upload.{ext}")
|
||
return super().to_internal_value(file)
|
||
|
||
return super().to_internal_value(data)
|
||
|
||
def to_representation(self, value):
|
||
if not value:
|
||
return ""
|
||
# Получаем байты
|
||
try:
|
||
file_obj = getattr(value, "file", None) or value
|
||
content = file_obj.read()
|
||
if hasattr(file_obj, "seek"):
|
||
file_obj.seek(0)
|
||
except Exception:
|
||
try:
|
||
with open(value.path, "rb") as f:
|
||
content = f.read()
|
||
except Exception:
|
||
return ""
|
||
|
||
b64 = base64.b64encode(content).decode("ascii")
|
||
_, mime = self._detect_ext_mime(content)
|
||
return f"data:{mime};base64,{b64}"
|
||
|
||
|
||
class EmailSerializer(serializers.Serializer):
|
||
email = serializers.EmailField()
|
||
|
||
|
||
class EmailVerifySerializer(serializers.Serializer):
|
||
email = serializers.EmailField()
|
||
code = serializers.CharField(max_length=6)
|
||
|
||
|
||
class RegisterSerializer(serializers.Serializer):
|
||
first_name = serializers.CharField(max_length=150)
|
||
last_name = serializers.CharField(max_length=150)
|
||
phone_number = serializers.CharField(max_length=100)
|
||
email = serializers.EmailField()
|
||
password = serializers.CharField(min_length=6)
|
||
birth_date = serializers.DateField()
|
||
|
||
telegram = serializers.CharField(required=False, allow_blank=True, default="")
|
||
whatsapp = serializers.CharField(required=False, allow_blank=True, default="")
|
||
wechat = serializers.CharField(required=False, allow_blank=True, default="")
|
||
|
||
# Доп. поля модели можно тоже принять, но они опциональны:
|
||
avatar = Base64ImageField(required=False, allow_null=True)
|
||
patronymic = serializers.CharField(required=False, allow_blank=True, default="")
|
||
gender = serializers.IntegerField(required=False, allow_null=True)
|
||
position = serializers.IntegerField(required=False, allow_null=True)
|
||
skills = serializers.ListField(child=serializers.IntegerField(), required=False)
|
||
|
||
|
||
class SkillSerializer(serializers.ModelSerializer):
|
||
class Meta:
|
||
model = Skill
|
||
fields = ["id", "name"]
|
||
|
||
|
||
class PositionBriefSerializer(serializers.ModelSerializer):
|
||
class Meta:
|
||
model = Position
|
||
fields = ("id", "name")
|
||
|
||
|
||
class EmployeeActionSerializer(serializers.ModelSerializer):
|
||
action_display = serializers.SerializerMethodField()
|
||
date_display = serializers.SerializerMethodField()
|
||
history = serializers.SerializerMethodField()
|
||
|
||
class Meta:
|
||
model = EmployeeAction
|
||
fields = [
|
||
"id",
|
||
"employee",
|
||
"action",
|
||
"action_display",
|
||
"date",
|
||
"date_display",
|
||
"comment",
|
||
"extra",
|
||
"history", # ← добавили
|
||
]
|
||
|
||
def get_action_display(self, obj):
|
||
return obj.get_action_display()
|
||
|
||
def get_date_display(self, obj):
|
||
return obj.date.strftime("%d.%m.%Y %H:%M") if obj.date else None
|
||
|
||
def get_history(self, obj):
|
||
if not self.context.get("include_action_history", False):
|
||
return []
|
||
|
||
items = list(
|
||
obj.history.select_related("history_user").order_by(
|
||
"-history_date", "-history_id"
|
||
)
|
||
)
|
||
|
||
action_labels = dict(ACTION_CHOICES)
|
||
history_type_labels = {"+": "Создано", "~": "Изменено", "-": "Удалено"}
|
||
diff_fields = ["action", "comment", "extra"]
|
||
|
||
out = []
|
||
for i, cur in enumerate(items):
|
||
prev = items[i + 1] if i + 1 < len(items) else None
|
||
changes = {}
|
||
|
||
for name in diff_fields:
|
||
new = getattr(cur, name, None)
|
||
old = getattr(prev, name, None) if prev else None
|
||
|
||
if name == "date":
|
||
new_fmt = new.isoformat() if new else None
|
||
old_fmt = old.isoformat() if old else None
|
||
if new_fmt != old_fmt:
|
||
changes[name] = {"old": old_fmt, "new": new_fmt}
|
||
elif name == "action":
|
||
if old != new:
|
||
changes[name] = {
|
||
"old": old,
|
||
"old_display": action_labels.get(old),
|
||
"new": new,
|
||
"new_display": action_labels.get(new),
|
||
}
|
||
else:
|
||
if old != new:
|
||
changes[name] = {"old": old, "new": new}
|
||
|
||
out.append(
|
||
{
|
||
"history_id": cur.history_id,
|
||
"history_type": cur.history_type,
|
||
"history_type_display": history_type_labels.get(cur.history_type),
|
||
"history_date": cur.history_date.isoformat(),
|
||
"history_date_display": cur.history_date.strftime("%d.%m.%Y %H:%M"),
|
||
"history_user": getattr(cur.history_user, "email", None),
|
||
# текущее значение действия в этом снимке + «человеческий» перевод
|
||
"action": getattr(cur, "action", None),
|
||
"action_display": action_labels.get(getattr(cur, "action", None)),
|
||
"changes": changes,
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
class EmployeeSerializer(serializers.ModelSerializer):
|
||
"""Полная версия сотрудника для собственных эндпоинтов /employees/."""
|
||
|
||
avatar = Base64ImageField(required=False, allow_null=True)
|
||
actions = EmployeeActionSerializer(many=True, read_only=True)
|
||
|
||
# навыки: читаем красиво, пишем ids
|
||
skills = SkillSerializer(many=True, read_only=True)
|
||
skills_ids = serializers.PrimaryKeyRelatedField(
|
||
queryset=Skill.objects.all(), many=True, write_only=True, required=False
|
||
)
|
||
|
||
# должность: читаем объектом, пишем через position_id
|
||
position = PositionBriefSerializer(read_only=True)
|
||
position_id = serializers.PrimaryKeyRelatedField(
|
||
source="position",
|
||
queryset=Position.objects.all(),
|
||
write_only=True,
|
||
required=False,
|
||
allow_null=True,
|
||
)
|
||
|
||
departments = serializers.SerializerMethodField()
|
||
auth = serializers.SerializerMethodField(read_only=True)
|
||
|
||
class Meta:
|
||
model = Employee
|
||
fields = (
|
||
"avatar",
|
||
"actions",
|
||
"id",
|
||
"email",
|
||
"last_name",
|
||
"first_name",
|
||
"patronymic",
|
||
"departments",
|
||
"gender",
|
||
"phone_number",
|
||
"birth_date",
|
||
"position",
|
||
"position_id",
|
||
"telegram",
|
||
"whatsapp",
|
||
"wechat",
|
||
"skills",
|
||
"skills_ids",
|
||
"is_active",
|
||
"email_verified",
|
||
"created_at",
|
||
"updated_at",
|
||
"last_login",
|
||
"date_joined",
|
||
"auth",
|
||
|
||
)
|
||
read_only_fields = (
|
||
"is_active",
|
||
"email_verified",
|
||
"created_at",
|
||
"updated_at",
|
||
"last_login",
|
||
"date_joined",
|
||
"auth",
|
||
)
|
||
extra_kwargs = {
|
||
# чтобы точно не принимали/не отдавали пароль и коды
|
||
"password": {"write_only": True, "required": False},
|
||
}
|
||
|
||
def to_representation(self, instance):
|
||
data = super().to_representation(instance)
|
||
return data
|
||
|
||
def update(self, instance, validated_data):
|
||
skills = validated_data.pop("skills_ids", None)
|
||
instance = super().update(instance, validated_data)
|
||
if skills is not None:
|
||
instance.skills.set(skills)
|
||
return instance
|
||
|
||
def get_departments(self, obj):
|
||
links = getattr(obj, "dept_links", None)
|
||
if links is None:
|
||
links = EmployeeDepartment.objects.select_related(
|
||
"department", "role"
|
||
).filter(employee=obj, is_active=True)
|
||
out = []
|
||
for link in links:
|
||
dept = link.department
|
||
out.append(
|
||
{
|
||
"id": link.department_id,
|
||
"name": dept.name if link.department_id else None,
|
||
"role_id": link.role_id,
|
||
"role_name": link.role.name if link.role_id else None,
|
||
"is_head": (dept.head_id == obj.id),
|
||
}
|
||
)
|
||
return out
|
||
|
||
def get_fields(self):
|
||
fields = super().get_fields()
|
||
|
||
# actions — как и было
|
||
if self.context.get("include_actions") is not True:
|
||
view = self.context.get("view")
|
||
action = getattr(view, "action", None) if view else None
|
||
if action not in {"retrieve", "me"}:
|
||
fields.pop("actions", None)
|
||
|
||
# ⬇️ Новое: auth показываем только в /me или при include_auth=True
|
||
include_auth = self.context.get("include_auth") is True
|
||
view = self.context.get("view")
|
||
action = getattr(view, "action", None) if view else None
|
||
if not include_auth and action != "me":
|
||
fields.pop("auth", None)
|
||
|
||
return fields
|
||
|
||
def get_auth(self, obj):
|
||
"""
|
||
Возвращает только ГЛОБАЛЬНЫЕ права текущего (request.user),
|
||
уже с учётом должности (PositionRoleBackend).
|
||
"""
|
||
req = self.context.get("request")
|
||
if not req or not getattr(req.user, "is_authenticated", False):
|
||
return None
|
||
|
||
# Не светим auth другого сотрудника (кроме явного include_auth=True)
|
||
include_auth = self.context.get("include_auth") is True
|
||
if not include_auth and req.user.pk != getattr(obj, "pk", None):
|
||
return None
|
||
|
||
u = req.user
|
||
perms = sorted(u.get_all_permissions()) # уже включает Position-права
|
||
|
||
by_app = defaultdict(list)
|
||
for p in perms:
|
||
app, code = p.split(".", 1)
|
||
by_app[app].append(code)
|
||
perms_by_app = {app: sorted(codes) for app, codes in by_app.items()}
|
||
|
||
return {
|
||
"id": u.id,
|
||
"email": u.email or "",
|
||
"is_staff": u.is_staff,
|
||
"is_superuser": u.is_superuser,
|
||
"groups": list(u.groups.values_list("name", flat=True)),
|
||
"permissions": perms,
|
||
"permissions_by_app": perms_by_app,
|
||
}
|
||
|
||
|
||
class DepartmentSerializer(serializers.ModelSerializer):
|
||
head = EmployeeSerializer(read_only=True)
|
||
head_id = serializers.PrimaryKeyRelatedField(
|
||
source="head",
|
||
queryset=Employee.objects.all(),
|
||
write_only=True,
|
||
required=False,
|
||
allow_null=True,
|
||
)
|
||
employees_count = serializers.IntegerField(read_only=True)
|
||
|
||
class Meta:
|
||
model = Department
|
||
fields = (
|
||
"id",
|
||
"name",
|
||
"description",
|
||
"head",
|
||
"head_id",
|
||
"head_appointed_at",
|
||
"created_at",
|
||
"employees_count",
|
||
)
|
||
|
||
# head_appointed_at и created_at только для чтения
|
||
read_only_fields = ("head_appointed_at", "created_at")
|
||
|
||
def validate_head_id(self, value: Employee | None):
|
||
if value and not value.is_actually_active:
|
||
raise serializers.ValidationError(
|
||
"Назначать можно только реально активного сотрудника"
|
||
)
|
||
return value
|
||
|
||
|
||
class EmployeeListSerializer(serializers.ModelSerializer):
|
||
email = serializers.EmailField(read_only=True)
|
||
phone_number = serializers.CharField(read_only=True)
|
||
avatar = Base64ImageField(read_only=True)
|
||
position = PositionBriefSerializer(read_only=True)
|
||
departments = serializers.SerializerMethodField()
|
||
full_name = serializers.SerializerMethodField()
|
||
skills = SkillSerializer(many=True, read_only=True)
|
||
|
||
class Meta:
|
||
model = Employee
|
||
fields = (
|
||
"id",
|
||
"email",
|
||
"phone_number",
|
||
"last_name",
|
||
"first_name",
|
||
"patronymic",
|
||
"full_name",
|
||
"avatar",
|
||
"position",
|
||
"departments",
|
||
"skills",
|
||
"created_at",
|
||
)
|
||
|
||
def get_departments(self, obj):
|
||
links = getattr(obj, "dept_links", None)
|
||
if links is None:
|
||
links = EmployeeDepartment.objects.select_related(
|
||
"department", "role"
|
||
).filter(employee=obj, is_active=True)
|
||
out = []
|
||
for link in links:
|
||
dept = link.department
|
||
out.append(
|
||
{
|
||
"id": link.department_id,
|
||
"name": dept.name if link.department_id else None,
|
||
"role_id": link.role_id,
|
||
"role_name": link.role.name if link.role_id else None,
|
||
"is_head": (dept.head_id == obj.id), # ✅ правильная проверка
|
||
}
|
||
)
|
||
return out
|
||
|
||
def get_full_name(self, obj):
|
||
fn = (obj.first_name or "").strip()
|
||
ln = (obj.last_name or "").strip()
|
||
pt = (obj.patronymic or "").strip()
|
||
nm = f"{ln} {fn} {pt}".strip()
|
||
return nm
|
||
|
||
|
||
class PositionSerializer(serializers.ModelSerializer):
|
||
# для записи — массив id групп, для чтения — подробная инфа
|
||
groups = serializers.PrimaryKeyRelatedField(
|
||
queryset=Group.objects.all(), many=True, required=False
|
||
)
|
||
groups_verbose = serializers.SerializerMethodField(read_only=True)
|
||
|
||
class Meta:
|
||
model = Position
|
||
fields = ["id", "name", "description", "groups", "groups_verbose"]
|
||
|
||
def get_groups_verbose(self, obj):
|
||
return [{"id": g.id, "name": g.name} for g in obj.groups.all()]
|
||
|
||
|
||
class DepartmentRoleSerializer(serializers.ModelSerializer):
|
||
permissions = serializers.PrimaryKeyRelatedField(
|
||
queryset=Permission.objects.all(), many=True, required=False
|
||
)
|
||
permissions_verbose = serializers.SerializerMethodField(read_only=True)
|
||
|
||
class Meta:
|
||
model = DepartmentRole
|
||
fields = ["id", "department", "name", "permissions", "permissions_verbose"]
|
||
|
||
def get_permissions_verbose(self, obj):
|
||
return [
|
||
{
|
||
"id": p.id,
|
||
"codename": f"{p.content_type.app_label}.{p.codename}",
|
||
"name": p.name,
|
||
}
|
||
for p in obj.permissions.select_related("content_type").all()
|
||
]
|