375 lines
13 KiB
Python
375 lines
13 KiB
Python
"""Общие (shared) сериализаторы: Skill, Position, EmployeeAction."""
|
|
|
|
from django.contrib.auth.models import Group
|
|
from drf_spectacular.utils import extend_schema_field
|
|
from rest_framework import serializers
|
|
|
|
from employees.constants import ACTION_CHOICES, TEMPORARY_START_ACTIONS
|
|
from employees.models import (
|
|
EmployeeAction,
|
|
EmployeeSkillProgress,
|
|
Position,
|
|
Skill,
|
|
SkillExperienceEvent,
|
|
)
|
|
from employees.services.secret_skills import (
|
|
secret_skill_progress_payload,
|
|
skill_rarity_payload,
|
|
)
|
|
from employees.services.skill_progress import (
|
|
confirmed_holder_count,
|
|
rank_payload,
|
|
)
|
|
|
|
|
|
class GroupVerboseSerializer(serializers.Serializer):
|
|
id = serializers.IntegerField()
|
|
name = serializers.CharField()
|
|
|
|
|
|
class EmployeeActionHistorySerializer(serializers.Serializer):
|
|
history_id = serializers.IntegerField()
|
|
history_type = serializers.CharField()
|
|
history_type_display = serializers.CharField(allow_null=True)
|
|
history_date = serializers.DateTimeField()
|
|
history_date_display = serializers.CharField()
|
|
history_user = serializers.CharField(allow_null=True)
|
|
action = serializers.CharField(allow_null=True)
|
|
action_display = serializers.CharField(allow_null=True)
|
|
changes = serializers.DictField()
|
|
|
|
|
|
class SkillSerializer(serializers.ModelSerializer):
|
|
experience = serializers.SerializerMethodField()
|
|
rank = serializers.SerializerMethodField()
|
|
rank_display = serializers.SerializerMethodField()
|
|
next_rank = serializers.SerializerMethodField()
|
|
next_rank_display = serializers.SerializerMethodField()
|
|
next_rank_experience = serializers.SerializerMethodField()
|
|
experience_to_next_rank = serializers.SerializerMethodField()
|
|
rank_progress_percent = serializers.SerializerMethodField()
|
|
confirmed_holders = serializers.SerializerMethodField()
|
|
rarity = serializers.SerializerMethodField()
|
|
rarity_display = serializers.SerializerMethodField()
|
|
progress_value = serializers.SerializerMethodField()
|
|
progress_label = serializers.SerializerMethodField()
|
|
progress_data = serializers.SerializerMethodField()
|
|
next_rank_requirements = serializers.SerializerMethodField()
|
|
anti_streak = serializers.SerializerMethodField()
|
|
anti_streak_label = serializers.SerializerMethodField()
|
|
anti_streak_limit = serializers.SerializerMethodField()
|
|
anti_streak_remaining = serializers.SerializerMethodField()
|
|
recovery_streak = serializers.SerializerMethodField()
|
|
recovery_streak_target = serializers.SerializerMethodField()
|
|
penalty_levels = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = Skill
|
|
fields = [
|
|
"id",
|
|
"name",
|
|
"is_secret",
|
|
"progression_type",
|
|
"experience",
|
|
"rank",
|
|
"rank_display",
|
|
"next_rank",
|
|
"next_rank_display",
|
|
"next_rank_experience",
|
|
"experience_to_next_rank",
|
|
"rank_progress_percent",
|
|
"confirmed_holders",
|
|
"rarity",
|
|
"rarity_display",
|
|
"progress_value",
|
|
"progress_label",
|
|
"progress_data",
|
|
"next_rank_requirements",
|
|
"anti_streak",
|
|
"anti_streak_label",
|
|
"anti_streak_limit",
|
|
"anti_streak_remaining",
|
|
"recovery_streak",
|
|
"recovery_streak_target",
|
|
"penalty_levels",
|
|
]
|
|
read_only_fields = ["is_secret", "progression_type"]
|
|
|
|
def _progress(self, obj):
|
|
progress_by_skill = self.context.get("skill_progress_by_skill", {})
|
|
return progress_by_skill.get(obj.id)
|
|
|
|
def _rank(self, obj):
|
|
progress = self._progress(obj)
|
|
if obj.is_secret:
|
|
return secret_skill_progress_payload(obj, progress)
|
|
return rank_payload(getattr(progress, "experience", 0))
|
|
|
|
def _rarity(self, obj):
|
|
count = getattr(obj, "confirmed_holders", None)
|
|
if count is None:
|
|
count = confirmed_holder_count(obj)
|
|
return skill_rarity_payload(obj, count)
|
|
|
|
def get_experience(self, obj) -> int:
|
|
return getattr(self._progress(obj), "experience", 0)
|
|
|
|
def get_rank(self, obj) -> str:
|
|
return self._rank(obj)["rank"]
|
|
|
|
def get_rank_display(self, obj) -> str:
|
|
return self._rank(obj)["rank_display"]
|
|
|
|
def get_next_rank(self, obj) -> str | None:
|
|
return self._rank(obj)["next_rank"]
|
|
|
|
def get_next_rank_display(self, obj) -> str | None:
|
|
return self._rank(obj)["next_rank_display"]
|
|
|
|
def get_next_rank_experience(self, obj) -> int | None:
|
|
return self._rank(obj)["next_rank_experience"]
|
|
|
|
def get_experience_to_next_rank(self, obj) -> int:
|
|
return self._rank(obj)["experience_to_next_rank"]
|
|
|
|
def get_rank_progress_percent(self, obj) -> int:
|
|
return self._rank(obj)["rank_progress_percent"]
|
|
|
|
def get_confirmed_holders(self, obj) -> int:
|
|
return self._rarity(obj)["confirmed_holders"]
|
|
|
|
def get_rarity(self, obj) -> str:
|
|
return self._rarity(obj)["rarity"]
|
|
|
|
def get_rarity_display(self, obj) -> str:
|
|
return self._rarity(obj)["rarity_display"]
|
|
|
|
def get_progress_value(self, obj) -> int | None:
|
|
return self._rank(obj).get("progress_value")
|
|
|
|
def get_progress_label(self, obj) -> str:
|
|
return self._rank(obj).get("progress_label", "")
|
|
|
|
def get_progress_data(self, obj) -> dict:
|
|
return self._rank(obj).get("progress_data", {})
|
|
|
|
def get_next_rank_requirements(self, obj) -> list[dict]:
|
|
return self._rank(obj).get("next_rank_requirements", [])
|
|
|
|
def get_anti_streak(self, obj) -> int:
|
|
return self._rank(obj).get("anti_streak", 0)
|
|
|
|
def get_anti_streak_label(self, obj) -> str:
|
|
return self._rank(obj).get("anti_streak_label", "Антисерия")
|
|
|
|
def get_anti_streak_limit(self, obj) -> int:
|
|
return self._rank(obj).get("anti_streak_limit", 0)
|
|
|
|
def get_anti_streak_remaining(self, obj) -> int:
|
|
return self._rank(obj).get("anti_streak_remaining", 0)
|
|
|
|
def get_recovery_streak(self, obj) -> int:
|
|
return self._rank(obj).get("recovery_streak", 0)
|
|
|
|
def get_recovery_streak_target(self, obj) -> int:
|
|
return self._rank(obj).get("recovery_streak_target", 0)
|
|
|
|
def get_penalty_levels(self, obj) -> int:
|
|
return self._rank(obj).get("penalty_levels", 0)
|
|
|
|
|
|
class SkillExperienceEventSerializer(serializers.ModelSerializer):
|
|
source_display = serializers.CharField(
|
|
source="get_source_type_display",
|
|
read_only=True,
|
|
)
|
|
actor_name = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = SkillExperienceEvent
|
|
fields = [
|
|
"id",
|
|
"points",
|
|
"source_type",
|
|
"source_display",
|
|
"source_id",
|
|
"actor",
|
|
"actor_name",
|
|
"reason",
|
|
"metadata",
|
|
"revoked_at",
|
|
"created_at",
|
|
]
|
|
|
|
def get_actor_name(self, obj) -> str | None:
|
|
if obj.actor is None:
|
|
return None
|
|
return obj.actor.get_full_name() or obj.actor.email
|
|
|
|
|
|
def serialize_employee_skills(employee, *, context=None):
|
|
progress_items = getattr(employee, "_prefetched_skill_progresses", None)
|
|
if progress_items is None:
|
|
progress_items = EmployeeSkillProgress.objects.filter(
|
|
employee=employee,
|
|
is_active=True,
|
|
)
|
|
progress_by_skill = {
|
|
progress.skill_id: progress for progress in progress_items if progress.is_active
|
|
}
|
|
serializer_context = dict(context or {})
|
|
serializer_context["skill_progress_by_skill"] = progress_by_skill
|
|
return SkillSerializer(
|
|
employee.skills.all(),
|
|
many=True,
|
|
context=serializer_context,
|
|
).data
|
|
|
|
|
|
class PositionBriefSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = Position
|
|
fields = ("id", "name")
|
|
|
|
|
|
class PositionSerializer(serializers.ModelSerializer):
|
|
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"]
|
|
|
|
@extend_schema_field(GroupVerboseSerializer(many=True))
|
|
def get_groups_verbose(self, obj):
|
|
return [{"id": g.id, "name": g.name} for g in obj.groups.all()]
|
|
|
|
|
|
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_to",
|
|
"date_display",
|
|
"comment",
|
|
"extra",
|
|
"history",
|
|
]
|
|
|
|
def validate(self, attrs):
|
|
action = attrs.get(
|
|
"action",
|
|
getattr(self.instance, "action", None),
|
|
)
|
|
date = attrs.get("date", getattr(self.instance, "date", None))
|
|
date_to = attrs.get("date_to", getattr(self.instance, "date_to", None))
|
|
|
|
if date_to is not None and action not in TEMPORARY_START_ACTIONS:
|
|
raise serializers.ValidationError(
|
|
{
|
|
"date_to": (
|
|
"Дата окончания доступна только для временных кадровых событий."
|
|
)
|
|
}
|
|
)
|
|
|
|
if (
|
|
action in TEMPORARY_START_ACTIONS
|
|
and date_to is None
|
|
and (
|
|
self.instance is None
|
|
or "action" in attrs
|
|
or "date" in attrs
|
|
or "date_to" in attrs
|
|
)
|
|
):
|
|
raise serializers.ValidationError(
|
|
{
|
|
"date_to": (
|
|
"Дата окончания обязательна для временных кадровых событий."
|
|
)
|
|
}
|
|
)
|
|
|
|
if date and date_to and date_to < date.date():
|
|
raise serializers.ValidationError(
|
|
{"date_to": "Дата окончания не может быть раньше даты начала."}
|
|
)
|
|
|
|
return attrs
|
|
|
|
@extend_schema_field(serializers.CharField())
|
|
def get_action_display(self, obj):
|
|
return obj.get_action_display()
|
|
|
|
@extend_schema_field(serializers.CharField(allow_null=True))
|
|
def get_date_display(self, obj):
|
|
return obj.date.strftime("%d.%m.%Y %H:%M") if obj.date else None
|
|
|
|
@extend_schema_field(EmployeeActionHistorySerializer(many=True))
|
|
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", "date", "date_to", "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 in {"date", "date_to"}:
|
|
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
|