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

fix: derive personnel skills from employee actions

Browse Source
This commit is contained in:
lizerk 2026-08-26 10:03:42 +03:00
parent 912b0da773
commit 5c896ad589
3 changed files with 305 additions and 186 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,12 +1,11 @@
from django.core.management.base import BaseCommand
from attendance.models import AttendanceRecord
from employees.models import EmployeeAction
from employees.models import Employee
from employees.services.secret_skills import recalculate_attendance_secret_skills
class Command(BaseCommand):
help = "Пересчитать секретные навыки по сохранённой посещаемости"
help = "Пересчитать системные секретные навыки сотрудников"
def add_arguments(self, parser):
parser.add_argument("--employee-id", type=int)
@ -16,19 +15,8 @@ class Command(BaseCommand):
if 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,
)
)
employee_ids = list(
Employee.objects.order_by("id").values_list("id", flat=True)
)
processed = 0

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

@ -1,4 +1,4 @@
"""Автоматический прогресс секретных навыков по данным посещаемости."""
"""Автоматический прогресс секретных навыков по системным данным."""
from __future__ import annotations
@ -11,8 +11,13 @@ from django.db import transaction
from django.utils import timezone
from employees.constants import (
ACTION_DISMISSED,
ACTION_HIRED,
ACTION_ON_DAY_OFF,
ACTION_ON_LEAVE,
ACTION_ON_MATERNITY,
ACTION_ON_SICK_LEAVE,
ACTION_REHIRED,
ACTION_REMOTE,
)
from employees.models import Employee, EmployeeAction, EmployeeSkillProgress, Skill
@ -85,14 +90,14 @@ REQUIREMENT_LABELS = {
"best_clean_streak": "Лучшая чистая серия",
"off_schedule_days": "Выходы в нерабочий день",
"overtime_days": "Дни с переработкой от 1 часа",
"worked_days": "Полностью отработанные плановые дни",
"worked_days": "Дни без отпуска и отгула",
"best_resilience_streak": "Лучшая серия между больничными",
}
ANTI_STREAK_LABELS = {
Skill.ProgressionType.ATTENDANCE_PUNCTUALITY: "Опоздания подряд",
Skill.ProgressionType.ATTENDANCE_SELF_ORGANIZATION: "Проблемные дни подряд",
Skill.ProgressionType.ATTENDANCE_RELIABILITY: "Пропуски за 30 дней",
Skill.ProgressionType.ATTENDANCE_RELIABILITY: "Отпуска и отгулы за 120 дней",
Skill.ProgressionType.PERSONNEL_RESILIENCE: "Больничные за 120 дней",
}
@ -142,6 +147,9 @@ SECRET_SKILL_DEFINITIONS = (
),
)
ATTENDANCE_SECRET_SKILL_DEFINITIONS = SECRET_SKILL_DEFINITIONS[:2]
PERSONNEL_SECRET_SKILL_DEFINITIONS = SECRET_SKILL_DEFINITIONS[2:]
def _has_work(record) -> bool:
return bool(
@ -182,13 +190,34 @@ def _action_date(action: EmployeeAction) -> date | None:
return value
def _is_gamified_sick_leave(action: EmployeeAction) -> bool:
return action.action == ACTION_ON_SICK_LEAVE and not (
def _is_gamified_action(action: EmployeeAction) -> bool:
return not (
isinstance(action.extra, dict)
and action.extra.get("exclude_from_secret_skills") is True
)
def _initial_secret_state(
definition: SecretSkillDefinition,
*,
rolling: bool = False,
) -> dict[str, object]:
state: dict[str, object] = {
"earned_rank": "none",
"effective_rank": "none",
"anti_streak": 0,
"anti_streak_limit": definition.anti_streak_limit,
"recovery_streak": 0,
"recovery_streak_target": definition.recovery_streak_target,
"penalty_levels": 0,
"last_demotion_date": "",
"last_recovery_date": "",
}
if rolling:
state["_rolling_dates"] = deque()
return state
def _prune_rolling_dates(
dates: deque[date],
current_date: date,
@ -314,9 +343,8 @@ def _advance_secret_state(
def calculate_attendance_secret_metrics( # noqa: C901
records: Iterable,
actions: Iterable[EmployeeAction] = (),
) -> dict[str, object]:
"""Рассчитать системные показатели по посещаемости и кадровым событиям."""
"""Рассчитать показатели навыков, основанных только на посещаемости."""
punctuality_streak = 0
best_punctuality_streak = 0
@ -325,43 +353,11 @@ def calculate_attendance_secret_metrics( # noqa: C901
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 = ""
definitions_by_type = {
definition.progression_type: definition
for definition in SECRET_SKILL_DEFINITIONS
skill_states = {
definition.system_code: _initial_secret_state(definition)
for definition in ATTENDANCE_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",
"anti_streak": 0,
"anti_streak_limit": definition.anti_streak_limit,
"recovery_streak": 0,
"recovery_streak_target": definition.recovery_streak_target,
"penalty_levels": 0,
"last_demotion_date": "",
"last_recovery_date": "",
}
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 {
@ -372,62 +368,20 @@ def calculate_attendance_secret_metrics( # noqa: C901
"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,
}
def refresh_paused_states():
metrics = current_metrics()
for definition in SECRET_SKILL_DEFINITIONS:
for definition in ATTENDANCE_SECRET_SKILL_DEFINITIONS:
_refresh_secret_state(
skill_states[definition.system_code],
definition,
metrics,
)
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()
for record in sorted(records, key=lambda item: (item.date, item.id or 0)):
if record.technical_issues:
continue
has_work = _has_work(record)
@ -442,9 +396,6 @@ def calculate_attendance_secret_metrics( # noqa: C901
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
@ -453,14 +404,11 @@ def calculate_attendance_secret_metrics( # noqa: C901
continue
last_evaluated_date = record.date.isoformat()
is_unexcused_absence = bool(record.is_absent or not has_work)
if is_unexcused_absence:
if record.is_absent or not has_work:
punctuality_streak = 0
clean_streak = 0
reliability_streak = 0
skill_states[reliability_definition.system_code]["recovery_streak"] = 0
metrics = current_metrics()
for definition in consecutive_definitions:
for definition in ATTENDANCE_SECRET_SKILL_DEFINITIONS:
_advance_secret_state(
skill_states[definition.system_code],
definition,
@ -468,13 +416,6 @@ def calculate_attendance_secret_metrics( # noqa: C901
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:
@ -501,22 +442,8 @@ def calculate_attendance_secret_metrics( # noqa: C901
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 consecutive_definitions:
for definition in ATTENDANCE_SECRET_SKILL_DEFINITIONS:
is_bad = (
bool(record.is_late)
if definition.progression_type
@ -530,43 +457,222 @@ def calculate_attendance_secret_metrics( # noqa: C901
is_bad=is_bad,
record_date=record.date,
)
if is_fully_worked:
return {**current_metrics(), "skill_states": skill_states}
def _employment_period(
employee: Employee,
actions: list[EmployeeAction],
reference_date: date,
) -> tuple[date, date] | None:
activation_dates = [
action_date
for action in actions
if action.action in {ACTION_HIRED, ACTION_REHIRED}
and (action_date := _action_date(action)) is not None
and action_date <= reference_date
]
created_date = timezone.localtime(employee.created_at).date()
start_date = max(activation_dates) if activation_dates else created_date
if start_date > reference_date:
return None
dismissal_dates = [
action_date
for action in actions
if action.action == ACTION_DISMISSED
and (action_date := _action_date(action)) is not None
and start_date <= action_date <= reference_date
]
end_date = min(dismissal_dates) if dismissal_dates else reference_date
return start_date, end_date
def _action_period(
action: EmployeeAction,
*,
start_date: date,
end_date: date,
) -> tuple[date, date] | None:
action_date = _action_date(action)
if action_date is None or action_date > end_date:
return None
period_start = max(action_date, start_date)
period_end = min(action.date_to or action_date, end_date)
if period_end < period_start:
return None
return period_start, period_end
def calculate_personnel_secret_metrics( # noqa: C901
employee: Employee,
actions: Iterable[EmployeeAction],
*,
reference_date: date | None = None,
) -> dict[str, object]:
"""Рассчитать «Надёжность» и «Стойкость» только по кадровым событиям."""
reference_date = reference_date or timezone.localdate()
sorted_actions = sorted(
(action for action in actions if _is_gamified_action(action)),
key=lambda item: (_action_date(item) or date.min, item.id or 0),
)
definitions_by_type = {
definition.progression_type: definition
for definition in PERSONNEL_SECRET_SKILL_DEFINITIONS
}
reliability_definition = definitions_by_type[
Skill.ProgressionType.ATTENDANCE_RELIABILITY
]
resilience_definition = definitions_by_type[
Skill.ProgressionType.PERSONNEL_RESILIENCE
]
skill_states = {
definition.system_code: _initial_secret_state(definition, rolling=True)
for definition in PERSONNEL_SECRET_SKILL_DEFINITIONS
}
employment_period = _employment_period(employee, sorted_actions, reference_date)
if employment_period is None:
return {
"worked_days": 0,
"current_reliability_streak": 0,
"best_reliability_streak": 0,
"current_resilience_streak": 0,
"best_resilience_streak": 0,
"leave_or_day_off_episodes": 0,
"sick_leave_episodes": 0,
"personnel_last_evaluated_date": "",
"skill_states": skill_states,
}
start_date, end_date = employment_period
reliability_blocked_dates: set[date] = set()
reliability_paused_dates: set[date] = set()
resilience_blocked_dates: set[date] = set()
resilience_paused_dates: set[date] = set()
reliability_episode_dates: set[date] = set()
sick_episode_dates: set[date] = set()
for action in sorted_actions:
if action.action not in {
ACTION_ON_LEAVE,
ACTION_ON_DAY_OFF,
ACTION_ON_SICK_LEAVE,
ACTION_ON_MATERNITY,
}:
continue
period = _action_period(action, start_date=start_date, end_date=end_date)
if period is None:
continue
period_start, period_end = period
period_dates = {
period_start + timedelta(days=offset)
for offset in range((period_end - period_start).days + 1)
}
if action.action in {ACTION_ON_LEAVE, ACTION_ON_DAY_OFF}:
reliability_blocked_dates.update(period_dates)
resilience_paused_dates.update(period_dates)
reliability_episode_dates.add(period_start)
elif action.action == ACTION_ON_SICK_LEAVE:
reliability_paused_dates.update(period_dates)
resilience_blocked_dates.update(period_dates)
sick_episode_dates.add(period_start)
else:
reliability_paused_dates.update(period_dates)
resilience_paused_dates.update(period_dates)
worked_days = 0
reliability_streak = 0
best_reliability_streak = 0
resilience_streak = 0
best_resilience_streak = 0
def current_metrics():
return {
"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,
"leave_or_day_off_episodes": len(reliability_episode_dates),
"sick_leave_episodes": len(sick_episode_dates),
"personnel_last_evaluated_date": end_date.isoformat(),
}
current_date = start_date
while current_date <= end_date:
for definition in PERSONNEL_SECRET_SKILL_DEFINITIONS:
rolling_dates = skill_states[definition.system_code]["_rolling_dates"]
_prune_rolling_dates(rolling_dates, current_date, 120)
skill_states[definition.system_code]["anti_streak"] = len(rolling_dates)
if current_date in reliability_episode_dates:
reliability_streak = 0
_register_rolling_violation(
skill_states[reliability_definition.system_code],
reliability_definition,
current_metrics(),
violation_date=current_date,
window_days=120,
)
if current_date in sick_episode_dates:
resilience_streak = 0
_register_rolling_violation(
skill_states[resilience_definition.system_code],
resilience_definition,
current_metrics(),
violation_date=current_date,
window_days=120,
)
if current_date not in reliability_blocked_dates | reliability_paused_dates:
worked_days += 1
reliability_streak += 1
best_reliability_streak = max(
best_reliability_streak,
reliability_streak,
)
_advance_rolling_recovery(
skill_states[reliability_definition.system_code],
reliability_definition,
metrics,
record_date=record.date,
current_metrics(),
record_date=current_date,
)
else:
_refresh_secret_state(
skill_states[reliability_definition.system_code],
reliability_definition,
metrics,
current_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),
):
if current_date not in resilience_blocked_dates | resilience_paused_dates:
resilience_streak += 1
best_resilience_streak = max(best_resilience_streak, resilience_streak)
_advance_rolling_recovery(
skill_states[resilience_definition.system_code],
resilience_definition,
current_metrics(),
record_date=current_date,
)
else:
_refresh_secret_state(
skill_states[resilience_definition.system_code],
resilience_definition,
current_metrics(),
)
current_date += timedelta(days=1)
for definition in PERSONNEL_SECRET_SKILL_DEFINITIONS:
state = skill_states[definition.system_code]
rolling_dates = state.pop("_rolling_dates")
_prune_rolling_dates(rolling_dates, reference_date, window_days)
_prune_rolling_dates(rolling_dates, end_date, 120)
state["anti_streak"] = len(rolling_dates)
state["recent_violation_dates"] = [
violation_date.isoformat() for violation_date in rolling_dates
]
return {
**current_metrics(),
"skill_states": skill_states,
}
return {**current_metrics(), "skill_states": skill_states}
def _candidate_rank(
@ -627,14 +733,15 @@ def _metrics_for_definition(
"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"],
"leave_or_day_off_episodes": metrics["leave_or_day_off_episodes"],
"last_evaluated_date": metrics["personnel_last_evaluated_date"],
**skill_state,
}
return {
"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"],
"last_evaluated_date": metrics["personnel_last_evaluated_date"],
**skill_state,
}
@ -655,7 +762,16 @@ def recalculate_attendance_secret_skills(employee_id: int) -> list[str]: # noqa
actions = list(
EmployeeAction.objects.filter(employee_id=employee_id).order_by("date", "id")
)
metrics = calculate_attendance_secret_metrics(records, actions)
attendance_metrics = calculate_attendance_secret_metrics(records)
personnel_metrics = calculate_personnel_secret_metrics(employee, actions)
metrics = {
**attendance_metrics,
**personnel_metrics,
"skill_states": {
**attendance_metrics["skill_states"],
**personnel_metrics["skill_states"],
},
}
changed_skills: list[str] = []
now = timezone.now()
@ -844,7 +960,7 @@ def secret_skill_progress_payload(
progress_label = (
recovery_label
if penalty_levels
else f"Полностью отработано: {progress_value} дн."
else f"Без отпуска и отгула: {progress_value} дн."
)
else:
progress_value = int(metrics.get("best_resilience_streak", 0) or 0)

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

@ -5,7 +5,12 @@ import pytest
from django.utils import timezone
from attendance.models import AttendanceAnalysisRun, AttendanceRecord
from employees.constants import ACTION_ON_SICK_LEAVE, ACTION_REMOTE
from employees.constants import (
ACTION_HIRED,
ACTION_ON_DAY_OFF,
ACTION_ON_SICK_LEAVE,
ACTION_REMOTE,
)
from employees.models import EmployeeAction, EmployeeSkillProgress, Skill
from employees.services.secret_skills import recalculate_attendance_secret_skills
@ -276,57 +281,62 @@ def test_self_organization_anti_streak_revokes_and_recovers(user_factory):
assert progress.progress_data["penalty_levels"] == 0
def test_reliability_counts_worked_days_and_rolling_absences(user_factory):
def test_reliability_uses_only_personnel_events_and_recovers_rank(user_factory):
employee = user_factory()
start = date(2025, 1, 1)
run = _run(employee, start, start + timedelta(days=129))
_action(employee, ACTION_HIRED, start)
for offset in range(60):
_record(run, employee, start + timedelta(days=offset))
recalculate_attendance_secret_skills(employee.id)
with patch(
"employees.services.secret_skills.timezone.localdate",
return_value=start + timedelta(days=59),
):
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,
_action(
employee,
ACTION_ON_DAY_OFF,
start + timedelta(days=offset),
arrival_time=None,
departure_time=None,
work_hours=0,
is_absent=True,
date_to=start + timedelta(days=offset),
)
recalculate_attendance_secret_skills(employee.id)
with patch(
"employees.services.secret_skills.timezone.localdate",
return_value=start + timedelta(days=70),
):
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)
with patch(
"employees.services.secret_skills.timezone.localdate",
return_value=start + timedelta(days=130),
):
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["worked_days"] == 128
assert progress.progress_data["leave_or_day_off_episodes"] == 3
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))
_action(employee, ACTION_HIRED, start)
for offset in range(60):
_record(run, employee, start + timedelta(days=offset))
recalculate_attendance_secret_skills(employee.id)
with patch(
"employees.services.secret_skills.timezone.localdate",
return_value=start + timedelta(days=59),
):
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"
@ -341,17 +351,22 @@ def test_resilience_counts_sick_episodes_and_recovers_rank(user_factory):
_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)
with patch(
"employees.services.secret_skills.timezone.localdate",
return_value=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)
with patch(
"employees.services.secret_skills.timezone.localdate",
return_value=start + timedelta(days=210),
):
recalculate_attendance_secret_skills(employee.id)
progress.refresh_from_db()
assert progress.special_rank == "bronze"
assert progress.is_active is True