Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add reliability and resilience secret skills

Browse Source
This commit is contained in:
lizerk 2026-08-26 09:50:55 +03:00
parent 9d16efadec
commit 912b0da773
10 changed files with 618 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
backend/api/v1/employees/serializers/shared.py
View File

@ -56,6 +56,7 @@ class SkillSerializer(serializers.ModelSerializer):
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()
@ -85,6 +86,7 @@ class SkillSerializer(serializers.ModelSerializer):
"progress_data",
"next_rank_requirements",
"anti_streak",
"anti_streak_label",
"anti_streak_limit",
"anti_streak_remaining",
"recovery_streak",
@ -157,6 +159,9 @@ class SkillSerializer(serializers.ModelSerializer):
def get_anti_streak(self, obj):
return self._rank(obj).get("anti_streak", 0)
def get_anti_streak_label(self, obj):
return self._rank(obj).get("anti_streak_label", "Антисерия")
def get_anti_streak_limit(self, obj):
return self._rank(obj).get("anti_streak_limit", 0)

28
backend/employees/management/commands/recalculate_secret_skills.py
View File

@ -1,6 +1,7 @@
from django.core.management.base import BaseCommand
from attendance.models import AttendanceRecord
from employees.models import EmployeeAction
from employees.services.secret_skills import recalculate_attendance_secret_skills
@ -12,20 +13,27 @@ class Command(BaseCommand):
def handle(self, *args, **options):
employee_id = options.get("employee_id")
employee_ids = (
AttendanceRecord.objects.order_by()
.values_list(
"employee_id",
flat=True,
)
.distinct()
)
if employee_id:
employee_ids = employee_ids.filter(employee_id=employee_id)
employee_ids = [employee_id]
else:
employee_ids = sorted(
set(
AttendanceRecord.objects.order_by().values_list(
"employee_id",
flat=True,
)
)
| set(
EmployeeAction.objects.order_by().values_list(
"employee_id",
flat=True,
)
)
)
processed = 0
rank_changes = 0
for current_employee_id in employee_ids.iterator():
for current_employee_id in employee_ids:
changes = recalculate_attendance_secret_skills(current_employee_id)
processed += 1
rank_changes += len(changes)

57
backend/employees/migrations/0059_reliability_and_resilience_skills.py Normal file
View File

@ -0,0 +1,57 @@
# Generated by Django 5.2.4 on 2026-08-26
from django.db import migrations, models
def create_reliability_and_resilience_skills(apps, schema_editor):
Skill = apps.get_model("employees", "Skill")
definitions = (
(
"attendance_reliability",
"Надёжность",
"attendance_reliability",
),
(
"personnel_resilience",
"Стойкость",
"personnel_resilience",
),
)
for system_code, name, progression_type in definitions:
skill, _ = Skill.objects.get_or_create(name=name)
skill.is_secret = True
skill.system_code = system_code
skill.progression_type = progression_type
skill.save(update_fields=["is_secret", "system_code", "progression_type"])
class Migration(migrations.Migration):
dependencies = [
("employees", "0058_secret_skill_progress"),
]
operations = [
migrations.AlterField(
model_name="skill",
name="progression_type",
field=models.CharField(
choices=[
("standard", "Обычный опыт"),
("attendance_punctuality", "Посещаемость: пунктуальность"),
(
"attendance_self_organization",
"Посещаемость: самоорганизация",
),
("attendance_reliability", "Посещаемость: надёжность"),
("personnel_resilience", "Кадровые события: стойкость"),
],
default="standard",
max_length=48,
verbose_name="Механика прогресса",
),
),
migrations.RunPython(
create_reliability_and_resilience_skills,
migrations.RunPython.noop,
),
]

8
backend/employees/models.py
View File

@ -702,6 +702,14 @@ class Skill(models.Model):
"attendance_self_organization",
"Посещаемость: самоорганизация",
)
ATTENDANCE_RELIABILITY = (
"attendance_reliability",
"Посещаемость: надёжность",
)
PERSONNEL_RESILIENCE = (
"personnel_resilience",
"Кадровые события: стойкость",
)
name = models.CharField(max_length=100, unique=True)
is_secret = models.BooleanField("Секретный навык", default=False)

368
backend/employees/services/secret_skills.py
View File

@ -2,14 +2,20 @@
from __future__ import annotations
from collections import deque
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from typing import Iterable
from django.db import transaction
from django.utils import timezone
from employees.constants import ACTION_REMOTE
from employees.models import Employee, EmployeeSkillProgress, Skill
from employees.constants import (
ACTION_ON_DAY_OFF,
ACTION_ON_SICK_LEAVE,
ACTION_REMOTE,
)
from employees.models import Employee, EmployeeAction, EmployeeSkillProgress, Skill
from employees.services.skill_progress import SKILL_RANKS, rarity_payload
RANK_CODES = tuple(rank["code"] for rank in SKILL_RANKS)
@ -57,12 +63,37 @@ SELF_ORGANIZATION_REQUIREMENTS = {
},
}
RELIABILITY_REQUIREMENTS = {
"bronze": {"worked_days": 60},
"silver": {"worked_days": 150},
"gold": {"worked_days": 300},
"ruby": {"worked_days": 600},
"diamond": {"worked_days": 1000},
}
RESILIENCE_REQUIREMENTS = {
"bronze": {"best_resilience_streak": 60},
"silver": {"best_resilience_streak": 120},
"gold": {"best_resilience_streak": 240},
"ruby": {"best_resilience_streak": 480},
"diamond": {"best_resilience_streak": 800},
}
REQUIREMENT_LABELS = {
"best_streak": "Лучшая серия без опозданий",
"clean_days": "Чистые рабочие дни",
"best_clean_streak": "Лучшая чистая серия",
"off_schedule_days": "Выходы в нерабочий день",
"overtime_days": "Дни с переработкой от 1 часа",
"worked_days": "Полностью отработанные плановые дни",
"best_resilience_streak": "Лучшая серия между больничными",
}
ANTI_STREAK_LABELS = {
Skill.ProgressionType.ATTENDANCE_PUNCTUALITY: "Опоздания подряд",
Skill.ProgressionType.ATTENDANCE_SELF_ORGANIZATION: "Проблемные дни подряд",
Skill.ProgressionType.ATTENDANCE_RELIABILITY: "Пропуски за 30 дней",
Skill.ProgressionType.PERSONNEL_RESILIENCE: "Больничные за 120 дней",
}
@ -93,6 +124,22 @@ SECRET_SKILL_DEFINITIONS = (
anti_streak_limit=5,
recovery_streak_target=30,
),
SecretSkillDefinition(
system_code="attendance_reliability",
name="Надёжность",
progression_type=Skill.ProgressionType.ATTENDANCE_RELIABILITY,
requirements=RELIABILITY_REQUIREMENTS,
anti_streak_limit=3,
recovery_streak_target=60,
),
SecretSkillDefinition(
system_code="personnel_resilience",
name="Стойкость",
progression_type=Skill.ProgressionType.PERSONNEL_RESILIENCE,
requirements=RESILIENCE_REQUIREMENTS,
anti_streak_limit=3,
recovery_streak_target=90,
),
)
@ -107,14 +154,104 @@ def _has_work(record) -> bool:
def _rank_metrics(metrics: dict[str, object], definition: SecretSkillDefinition):
if definition.progression_type == Skill.ProgressionType.ATTENDANCE_PUNCTUALITY:
return {"best_streak": metrics["best_streak"]}
if (
definition.progression_type
== Skill.ProgressionType.ATTENDANCE_SELF_ORGANIZATION
):
return {
"clean_days": metrics["clean_days"],
"best_clean_streak": metrics["best_clean_streak"],
"off_schedule_days": metrics["off_schedule_days"],
"overtime_days": metrics["overtime_days"],
}
if definition.progression_type == Skill.ProgressionType.ATTENDANCE_RELIABILITY:
return {"worked_days": metrics["worked_days"]}
return {
"clean_days": metrics["clean_days"],
"best_clean_streak": metrics["best_clean_streak"],
"off_schedule_days": metrics["off_schedule_days"],
"overtime_days": metrics["overtime_days"],
"best_resilience_streak": metrics["best_resilience_streak"],
}
def _action_date(action: EmployeeAction) -> date | None:
value = action.date
if not value:
return None
if isinstance(value, datetime):
if timezone.is_aware(value):
return timezone.localtime(value).date()
return value.date()
return value
def _is_gamified_sick_leave(action: EmployeeAction) -> bool:
return action.action == ACTION_ON_SICK_LEAVE and not (
isinstance(action.extra, dict)
and action.extra.get("exclude_from_secret_skills") is True
)
def _prune_rolling_dates(
dates: deque[date],
current_date: date,
window_days: int,
) -> None:
minimum_date = current_date - timedelta(days=window_days - 1)
while dates and dates[0] < minimum_date:
dates.popleft()
def _register_rolling_violation(
state: dict[str, object],
definition: SecretSkillDefinition,
metrics: dict[str, object],
*,
violation_date: date,
window_days: int,
) -> None:
_refresh_secret_state(state, definition, metrics)
earned_rank_index = RANK_INDEX[str(state["earned_rank"])]
dates = state["_rolling_dates"]
if earned_rank_index == 0:
dates.clear()
state["anti_streak"] = 0
state["recovery_streak"] = 0
return
_prune_rolling_dates(dates, violation_date, window_days)
dates.append(violation_date)
state["anti_streak"] = len(dates)
state["recovery_streak"] = 0
penalty_levels = int(state["penalty_levels"])
if (
earned_rank_index > 0
and penalty_levels < earned_rank_index
and len(dates) >= definition.anti_streak_limit
):
state["penalty_levels"] = penalty_levels + 1
state["anti_streak"] = 0
state["last_demotion_date"] = violation_date.isoformat()
dates.clear()
_refresh_secret_state(state, definition, metrics)
def _advance_rolling_recovery(
state: dict[str, object],
definition: SecretSkillDefinition,
metrics: dict[str, object],
*,
record_date: date,
) -> None:
_refresh_secret_state(state, definition, metrics)
if int(state["penalty_levels"]) <= 0:
state["recovery_streak"] = 0
return
state["recovery_streak"] = int(state["recovery_streak"]) + 1
if int(state["recovery_streak"]) >= definition.recovery_streak_target:
state["penalty_levels"] = int(state["penalty_levels"]) - 1
state["recovery_streak"] = 0
state["last_recovery_date"] = record_date.isoformat()
_refresh_secret_state(state, definition, metrics)
def _refresh_secret_state(
state: dict[str, int | str],
definition: SecretSkillDefinition,
@ -175,8 +312,11 @@ def _advance_secret_state(
_refresh_secret_state(state, definition, metrics)
def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
"""Рассчитать фактические показатели по нормализованным записям."""
def calculate_attendance_secret_metrics( # noqa: C901
records: Iterable,
actions: Iterable[EmployeeAction] = (),
) -> dict[str, object]:
"""Рассчитать системные показатели по посещаемости и кадровым событиям."""
punctuality_streak = 0
best_punctuality_streak = 0
@ -185,8 +325,28 @@ def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
clean_days = 0
off_schedule_days = 0
overtime_days = 0
worked_days = 0
reliability_streak = 0
best_reliability_streak = 0
resilience_streak = 0
best_resilience_streak = 0
sick_leave_episodes = 0
last_evaluated_date = ""
skill_states: dict[str, dict[str, int | str]] = {
definitions_by_type = {
definition.progression_type: definition
for definition in SECRET_SKILL_DEFINITIONS
}
consecutive_definitions = (
definitions_by_type[Skill.ProgressionType.ATTENDANCE_PUNCTUALITY],
definitions_by_type[Skill.ProgressionType.ATTENDANCE_SELF_ORGANIZATION],
)
reliability_definition = definitions_by_type[
Skill.ProgressionType.ATTENDANCE_RELIABILITY
]
resilience_definition = definitions_by_type[
Skill.ProgressionType.PERSONNEL_RESILIENCE
]
skill_states: dict[str, dict[str, object]] = {
definition.system_code: {
"earned_rank": "none",
"effective_rank": "none",
@ -200,6 +360,8 @@ def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
}
for definition in SECRET_SKILL_DEFINITIONS
}
skill_states[reliability_definition.system_code]["_rolling_dates"] = deque()
skill_states[resilience_definition.system_code]["_rolling_dates"] = deque()
def current_metrics():
return {
@ -210,6 +372,12 @@ def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
"clean_days": clean_days,
"off_schedule_days": off_schedule_days,
"overtime_days": overtime_days,
"worked_days": worked_days,
"current_reliability_streak": reliability_streak,
"best_reliability_streak": best_reliability_streak,
"current_resilience_streak": resilience_streak,
"best_resilience_streak": best_resilience_streak,
"sick_leave_episodes": sick_leave_episodes,
"last_evaluated_date": last_evaluated_date,
}
@ -222,8 +390,44 @@ def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
metrics,
)
for record in sorted(records, key=lambda item: (item.date, item.id or 0)):
if record.technical_issues:
record_by_date = {
record.date: record
for record in sorted(records, key=lambda item: (item.date, item.id or 0))
}
today = timezone.localdate()
sick_episode_dates = {
action_date
for action in actions
if _is_gamified_sick_leave(action)
and (action_date := _action_date(action)) is not None
and action_date <= today
}
timeline_dates = sorted(set(record_by_date) | sick_episode_dates)
for timeline_date in timeline_dates:
for definition, window_days in (
(reliability_definition, 30),
(resilience_definition, 120),
):
state = skill_states[definition.system_code]
rolling_dates = state["_rolling_dates"]
_prune_rolling_dates(rolling_dates, timeline_date, window_days)
state["anti_streak"] = len(rolling_dates)
if timeline_date in sick_episode_dates:
resilience_streak = 0
sick_leave_episodes += 1
_register_rolling_violation(
skill_states[resilience_definition.system_code],
resilience_definition,
current_metrics(),
violation_date=timeline_date,
window_days=120,
)
record = record_by_date.get(timeline_date)
if record is None or record.technical_issues:
refresh_paused_states()
continue
has_work = _has_work(record)
@ -238,6 +442,9 @@ def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
and has_work
):
off_schedule_days += 1
if record.personnel_status == ACTION_ON_DAY_OFF:
reliability_streak = 0
skill_states[reliability_definition.system_code]["recovery_streak"] = 0
refresh_paused_states()
continue
@ -246,11 +453,14 @@ def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
continue
last_evaluated_date = record.date.isoformat()
if record.is_absent or not has_work:
is_unexcused_absence = bool(record.is_absent or not has_work)
if is_unexcused_absence:
punctuality_streak = 0
clean_streak = 0
reliability_streak = 0
skill_states[reliability_definition.system_code]["recovery_streak"] = 0
metrics = current_metrics()
for definition in SECRET_SKILL_DEFINITIONS:
for definition in consecutive_definitions:
_advance_secret_state(
skill_states[definition.system_code],
definition,
@ -258,6 +468,13 @@ def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
is_bad=True,
record_date=record.date,
)
_register_rolling_violation(
skill_states[reliability_definition.system_code],
reliability_definition,
metrics,
violation_date=record.date,
window_days=30,
)
continue
if record.is_late:
@ -284,8 +501,22 @@ def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
else:
clean_streak = 0
is_fully_worked = not record.is_underwork and not record.is_early_leave
if is_fully_worked:
worked_days += 1
reliability_streak += 1
best_reliability_streak = max(
best_reliability_streak,
reliability_streak,
)
else:
reliability_streak = 0
skill_states[reliability_definition.system_code]["recovery_streak"] = 0
resilience_streak += 1
best_resilience_streak = max(best_resilience_streak, resilience_streak)
metrics = current_metrics()
for definition in SECRET_SKILL_DEFINITIONS:
for definition in consecutive_definitions:
is_bad = (
bool(record.is_late)
if definition.progression_type
@ -299,16 +530,41 @@ def calculate_attendance_secret_metrics(records: Iterable) -> dict[str, object]:
is_bad=is_bad,
record_date=record.date,
)
if is_fully_worked:
_advance_rolling_recovery(
skill_states[reliability_definition.system_code],
reliability_definition,
metrics,
record_date=record.date,
)
else:
_refresh_secret_state(
skill_states[reliability_definition.system_code],
reliability_definition,
metrics,
)
_advance_rolling_recovery(
skill_states[resilience_definition.system_code],
resilience_definition,
metrics,
record_date=record.date,
)
reference_date = timeline_dates[-1] if timeline_dates else today
for definition, window_days in (
(reliability_definition, 30),
(resilience_definition, 120),
):
state = skill_states[definition.system_code]
rolling_dates = state.pop("_rolling_dates")
_prune_rolling_dates(rolling_dates, reference_date, window_days)
state["anti_streak"] = len(rolling_dates)
state["recent_violation_dates"] = [
violation_date.isoformat() for violation_date in rolling_dates
]
return {
"current_streak": punctuality_streak,
"best_streak": best_punctuality_streak,
"current_clean_streak": clean_streak,
"best_clean_streak": best_clean_streak,
"clean_days": clean_days,
"off_schedule_days": off_schedule_days,
"overtime_days": overtime_days,
"last_evaluated_date": last_evaluated_date,
**current_metrics(),
"skill_states": skill_states,
}
@ -353,19 +609,38 @@ def _metrics_for_definition(
"last_evaluated_date": metrics["last_evaluated_date"],
**skill_state,
}
if (
definition.progression_type
== Skill.ProgressionType.ATTENDANCE_SELF_ORGANIZATION
):
return {
"current_clean_streak": metrics["current_clean_streak"],
"best_clean_streak": metrics["best_clean_streak"],
"clean_days": metrics["clean_days"],
"off_schedule_days": metrics["off_schedule_days"],
"overtime_days": metrics["overtime_days"],
"last_evaluated_date": metrics["last_evaluated_date"],
**skill_state,
}
if definition.progression_type == Skill.ProgressionType.ATTENDANCE_RELIABILITY:
return {
"worked_days": metrics["worked_days"],
"current_reliability_streak": metrics["current_reliability_streak"],
"best_reliability_streak": metrics["best_reliability_streak"],
"last_evaluated_date": metrics["last_evaluated_date"],
**skill_state,
}
return {
"current_clean_streak": metrics["current_clean_streak"],
"best_clean_streak": metrics["best_clean_streak"],
"clean_days": metrics["clean_days"],
"off_schedule_days": metrics["off_schedule_days"],
"overtime_days": metrics["overtime_days"],
"current_resilience_streak": metrics["current_resilience_streak"],
"best_resilience_streak": metrics["best_resilience_streak"],
"sick_leave_episodes": metrics["sick_leave_episodes"],
"last_evaluated_date": metrics["last_evaluated_date"],
**skill_state,
}
@transaction.atomic
def recalculate_attendance_secret_skills(employee_id: int) -> list[str]:
def recalculate_attendance_secret_skills(employee_id: int) -> list[str]: # noqa: C901
"""Пересчитать и при необходимости открыть системные навыки сотрудника."""
from attendance.models import AttendanceRecord
@ -377,7 +652,10 @@ def recalculate_attendance_secret_skills(employee_id: int) -> list[str]:
records = list(
AttendanceRecord.objects.filter(employee_id=employee_id).order_by("date", "id")
)
metrics = calculate_attendance_secret_metrics(records)
actions = list(
EmployeeAction.objects.filter(employee_id=employee_id).order_by("date", "id")
)
metrics = calculate_attendance_secret_metrics(records, actions)
changed_skills: list[str] = []
now = timezone.now()
@ -539,25 +817,42 @@ def secret_skill_progress_payload(
else:
progress_percent = 100
recovery_label = (
f"Восстановление ранга: {metrics.get('recovery_streak', 0)}/"
f"{definition.recovery_streak_target} дн."
)
if definition.progression_type == Skill.ProgressionType.ATTENDANCE_PUNCTUALITY:
progress_value = int(metrics.get("best_streak", 0) or 0)
progress_label = (
f"Восстановление ранга: {metrics.get('recovery_streak', 0)}/"
f"{definition.recovery_streak_target} дн."
if penalty_levels
else f"Лучшая серия: {progress_value} дн."
recovery_label if penalty_levels else f"Лучшая серия: {progress_value} дн."
)
else:
elif (
definition.progression_type
== Skill.ProgressionType.ATTENDANCE_SELF_ORGANIZATION
):
progress_value = int(metrics.get("clean_days", 0) or 0)
completed = sum(item["complete"] for item in requirement_payload)
progress_label = (
f"Восстановление ранга: {metrics.get('recovery_streak', 0)}/"
f"{definition.recovery_streak_target} дн."
recovery_label
if penalty_levels
else f"До следующего ранга: {completed}/{len(requirement_payload)} условий"
if requirement_payload
else "Все условия выполнены"
)
elif definition.progression_type == Skill.ProgressionType.ATTENDANCE_RELIABILITY:
progress_value = int(metrics.get("worked_days", 0) or 0)
progress_label = (
recovery_label
if penalty_levels
else f"Полностью отработано: {progress_value} дн."
)
else:
progress_value = int(metrics.get("best_resilience_streak", 0) or 0)
progress_label = (
recovery_label
if penalty_levels
else f"Лучшая серия без больничного: {progress_value} дн."
)
primary_target = next(iter(next_requirements.values()), None)
primary_current = (
@ -583,6 +878,7 @@ def secret_skill_progress_payload(
"progress_data": metrics,
"next_rank_requirements": requirement_payload,
"anti_streak": int(metrics.get("anti_streak", 0) or 0),
"anti_streak_label": ANTI_STREAK_LABELS[definition.progression_type],
"anti_streak_limit": definition.anti_streak_limit,
"anti_streak_remaining": max(
definition.anti_streak_limit - int(metrics.get("anti_streak", 0) or 0),

4
backend/employees/signals/__init__.py
View File

@ -13,7 +13,7 @@ from importlib import import_module
from django.conf import settings
from . import birthday, common, skills
from . import birthday, common, secret_skills, skills
from .common import (
IMMEDIATE_ACTION_MAPPING,
SCHEDULED_ACTION_MAPPING,
@ -31,6 +31,7 @@ if getattr(settings, "LDAP_ENABLED", False):
"common",
"birthday",
"skills",
"secret_skills",
"IMMEDIATE_ACTION_MAPPING",
"SCHEDULED_ACTION_MAPPING",
"_apply_action_effects",
@ -41,6 +42,7 @@ else:
"common",
"birthday",
"skills",
"secret_skills",
"IMMEDIATE_ACTION_MAPPING",
"SCHEDULED_ACTION_MAPPING",
"_apply_action_effects",

62
backend/employees/signals/secret_skills.py Normal file
View File

@ -0,0 +1,62 @@
"""Пересчёт системных навыков после изменения кадровых событий."""
import logging
from django.db import transaction
from django.db.models.signals import post_delete, post_save, pre_save
from django.dispatch import receiver
from employees.models import EmployeeAction
logger = logging.getLogger(__name__)
def _schedule_recalculation(employee_ids) -> None:
normalized_ids = {employee_id for employee_id in employee_ids if employee_id}
if not normalized_ids:
return
def recalculate():
from employees.services.secret_skills import (
recalculate_attendance_secret_skills,
)
for employee_id in normalized_ids:
try:
recalculate_attendance_secret_skills(employee_id)
except Exception:
logger.exception(
"Secret skill recalculation after personnel action failed",
extra={
"event": "employee.secret_skills.recalculation_failed",
"employee_id": employee_id,
},
)
transaction.on_commit(recalculate)
@receiver(pre_save, sender=EmployeeAction)
def remember_previous_action_employee(sender, instance, **kwargs):
if not instance.pk:
return
instance._secret_skills_previous_employee_id = (
sender.objects.filter(pk=instance.pk)
.values_list("employee_id", flat=True)
.first()
)
@receiver(post_save, sender=EmployeeAction)
def recalculate_secret_skills_after_action_save(sender, instance, **kwargs):
_schedule_recalculation(
{
instance.employee_id,
getattr(instance, "_secret_skills_previous_employee_id", None),
}
)
@receiver(post_delete, sender=EmployeeAction)
def recalculate_secret_skills_after_action_delete(sender, instance, **kwargs):
_schedule_recalculation({instance.employee_id})

117
backend/tests/employees/test_secret_skills.py
View File

@ -1,10 +1,12 @@
from datetime import date, timedelta
from datetime import date, datetime, time, timedelta
from unittest.mock import patch
import pytest
from django.utils import timezone
from attendance.models import AttendanceAnalysisRun, AttendanceRecord
from employees.constants import ACTION_REMOTE
from employees.models import EmployeeSkillProgress, Skill
from employees.constants import ACTION_ON_SICK_LEAVE, ACTION_REMOTE
from employees.models import EmployeeAction, EmployeeSkillProgress, Skill
from employees.services.secret_skills import recalculate_attendance_secret_skills
pytestmark = pytest.mark.django_db
@ -38,6 +40,18 @@ def _record(run, employee, record_date, **overrides):
)
def _action(employee, action, action_date, **overrides):
values = {
"date": timezone.make_aware(datetime.combine(action_date, time.min)),
}
values.update(overrides)
return EmployeeAction.objects.create(
employee=employee,
action=action,
**values,
)
def test_punctuality_uses_best_streak_and_pauses_unknown_days(user_factory):
employee = user_factory()
start = date(2026, 1, 1)
@ -260,3 +274,100 @@ def test_self_organization_anti_streak_revokes_and_recovers(user_factory):
assert progress.special_rank == "bronze"
assert progress.is_active is True
assert progress.progress_data["penalty_levels"] == 0
def test_reliability_counts_worked_days_and_rolling_absences(user_factory):
employee = user_factory()
start = date(2025, 1, 1)
run = _run(employee, start, start + timedelta(days=129))
for offset in range(60):
_record(run, employee, start + timedelta(days=offset))
recalculate_attendance_secret_skills(employee.id)
skill = Skill.objects.get(system_code="attendance_reliability")
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
assert progress.special_rank == "bronze"
assert progress.progress_data["worked_days"] == 60
for offset in (60, 65, 70):
_record(
run,
employee,
start + timedelta(days=offset),
arrival_time=None,
departure_time=None,
work_hours=0,
is_absent=True,
)
recalculate_attendance_secret_skills(employee.id)
progress.refresh_from_db()
assert progress.special_rank == "none"
assert progress.is_active is False
assert progress.progress_data["penalty_levels"] == 1
for offset in range(71, 131):
_record(run, employee, start + timedelta(days=offset))
recalculate_attendance_secret_skills(employee.id)
progress.refresh_from_db()
assert progress.special_rank == "bronze"
assert progress.is_active is True
assert progress.progress_data["worked_days"] == 120
assert progress.progress_data["penalty_levels"] == 0
def test_resilience_counts_sick_episodes_and_recovers_rank(user_factory):
employee = user_factory()
start = date(2024, 1, 1)
run = _run(employee, start, start + timedelta(days=210))
for offset in range(60):
_record(run, employee, start + timedelta(days=offset))
recalculate_attendance_secret_skills(employee.id)
skill = Skill.objects.get(system_code="personnel_resilience")
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
assert progress.special_rank == "bronze"
assert progress.progress_data["best_resilience_streak"] == 60
_action(
employee,
ACTION_ON_SICK_LEAVE,
start + timedelta(days=60),
date_to=start + timedelta(days=79),
)
_action(employee, ACTION_ON_SICK_LEAVE, start + timedelta(days=90))
_action(employee, ACTION_ON_SICK_LEAVE, start + timedelta(days=120))
recalculate_attendance_secret_skills(employee.id)
progress.refresh_from_db()
assert progress.special_rank == "none"
assert progress.is_active is False
assert progress.progress_data["sick_leave_episodes"] == 3
assert progress.progress_data["penalty_levels"] == 1
for offset in range(121, 211):
_record(run, employee, start + timedelta(days=offset))
recalculate_attendance_secret_skills(employee.id)
progress.refresh_from_db()
assert progress.special_rank == "bronze"
assert progress.is_active is True
assert progress.progress_data["current_resilience_streak"] == 90
assert progress.progress_data["penalty_levels"] == 0
def test_personnel_action_schedules_secret_skill_recalculation(
user_factory,
django_capture_on_commit_callbacks,
):
employee = user_factory()
with patch(
"employees.services.secret_skills.recalculate_attendance_secret_skills"
) as recalculate:
with django_capture_on_commit_callbacks(execute=True):
_action(employee, ACTION_ON_SICK_LEAVE, date(2026, 8, 1))
recalculate.assert_called_once_with(employee.id)

9
frontend/src/components/users/ProfileSections.tsx
View File

@ -909,7 +909,7 @@ export function ProfileSkillsInline({
{(skill.anti_streak || 0) > 0 ? (
<span
className="rounded-full bg-red-500/15 px-1.5 py-0.5 text-[10px] font-semibold leading-none text-red-200"
title={`Антисерия ${skill.anti_streak}/${skill.anti_streak_limit}`}
title={`${skill.anti_streak_label || "Антисерия"}: ${skill.anti_streak}/${skill.anti_streak_limit}`}
>
! {skill.anti_streak}/{skill.anti_streak_limit}
</span>
@ -1168,12 +1168,13 @@ export function ProfileSkillsInline({
{progress.is_secret && (progress.anti_streak || 0) > 0 ? (
<div className="mt-3 rounded-xl border border-red-400/40 bg-red-500/10 px-3 py-2.5 text-xs text-red-200">
<p className="font-semibold">
Антисерия: {progress.anti_streak}/{progress.anti_streak_limit}
{progress.anti_streak_label || "Антисерия"}: {progress.anti_streak}/
{progress.anti_streak_limit}
</p>
<p className="mt-1 text-red-200/80">
{progress.anti_streak_remaining === 1
? "Следующее нарушение понизит ранг."
: `До понижения ранга: ${progress.anti_streak_remaining} наруш.`}
? "Следующий случай понизит ранг."
: `До понижения ранга: ${progress.anti_streak_remaining}.`}
</p>
</div>
) : null}

16
frontend/src/types/api.ts
View File

@ -3,7 +3,12 @@ export interface Skill {
id: number;
name: string;
is_secret?: boolean;
progression_type?: "standard" | "attendance_punctuality" | "attendance_self_organization";
progression_type?:
| "standard"
| "attendance_punctuality"
| "attendance_self_organization"
| "attendance_reliability"
| "personnel_resilience";
description?: string;
experience?: number;
rank?: SkillRank;
@ -21,6 +26,7 @@ export interface Skill {
progress_data?: Record<string, unknown>;
next_rank_requirements?: SkillRankRequirement[];
anti_streak?: number;
anti_streak_label?: string;
anti_streak_limit?: number;
anti_streak_remaining?: number;
recovery_streak?: number;
@ -57,7 +63,12 @@ export interface EmployeeSkillProgress {
skill_id: number;
skill_name: string;
is_secret: boolean;
progression_type: "standard" | "attendance_punctuality" | "attendance_self_organization";
progression_type:
| "standard"
| "attendance_punctuality"
| "attendance_self_organization"
| "attendance_reliability"
| "personnel_resilience";
experience: number;
rank: SkillRank;
rank_display: string;
@ -74,6 +85,7 @@ export interface EmployeeSkillProgress {
progress_data?: Record<string, unknown>;
next_rank_requirements?: SkillRankRequirement[];
anti_streak?: number;
anti_streak_label?: string;
anti_streak_limit?: number;
anti_streak_remaining?: number;
recovery_streak?: number;