lokses
cfafe2f0b3
Иван Иванов -> i.ivanov (транслит pytils, нижний регистр, коллизия +счётчик).
Раньше slugify('Фамилия.Имя') терял точку и давал ivanovivan. Единый формат
в найме (_generate_username) и в импорте (transliterate_username).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
330 lines
12 KiB
Python
330 lines
12 KiB
Python
"""Бізнес-логіка найму: інтерн-слот + IT Ticket на випробуванні;
|
|
персональний Employee і PDF — лише після затвердження."""
|
|
import secrets
|
|
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from django.utils.translation import gettext as _
|
|
from pytils.translit import slugify
|
|
|
|
from apps.employees.models.base import Employee
|
|
from apps.hiring.models import HiringRequest, InternAccount
|
|
|
|
|
|
def _username_base(last_name, first_name):
|
|
"""Формат логина: <первая буква имени>.<фамилия>, транслит, нижний регистр.
|
|
Иван Иванов -> i.ivanov."""
|
|
ln = slugify(last_name)
|
|
fn = slugify(first_name)
|
|
initial = fn[:1]
|
|
return f'{initial}.{ln}'.strip('.') or 'user'
|
|
|
|
|
|
def _generate_username(last_name, first_name):
|
|
base = _username_base(last_name, first_name)
|
|
username = base
|
|
counter = 1
|
|
while Employee.objects.filter(username=username).exists():
|
|
counter += 1
|
|
username = f'{base}{counter}'
|
|
return username
|
|
|
|
|
|
def _generate_password():
|
|
return secrets.token_urlsafe(12)
|
|
|
|
|
|
def _coerce_location_fields(user_location_type, department_id, user_pharmacy_id, user_location_id):
|
|
"""Приводить поля локації до правил Employee.clean(): аптека XOR офіс/склад."""
|
|
if user_location_type == 'pharmacy':
|
|
return None, user_pharmacy_id, None
|
|
if user_location_type in ('office', 'warehouse'):
|
|
return department_id, None, user_location_id
|
|
return department_id, user_pharmacy_id, user_location_id
|
|
|
|
|
|
def _filter_access_ids_for_location(user_location_type, requested_access_ids):
|
|
"""Залишає лише системи доступу, дозволені для типу локації.
|
|
|
|
Порожній location_types у системи = доступна скрізь (як у посад).
|
|
"""
|
|
if not requested_access_ids:
|
|
return []
|
|
from core.models import Directory
|
|
allowed = []
|
|
for access in Directory.objects.filter(
|
|
pk__in=requested_access_ids,
|
|
directory_type__code='access_system',
|
|
is_active=True,
|
|
).prefetch_related('location_types'):
|
|
types = [lt.code for lt in access.location_types.all()]
|
|
if not types or user_location_type in types:
|
|
allowed.append(access.pk)
|
|
return allowed
|
|
|
|
|
|
def _validate_hire_location(*, user_location_type, department_id, user_pharmacy_id,
|
|
user_location_id, post_id, hire_date):
|
|
"""Перевіряє правила локації Employee.clean() без збереження в БД."""
|
|
emp = Employee(
|
|
username='__hire_validate__',
|
|
last_name='_',
|
|
first_name='_',
|
|
hire_date=hire_date,
|
|
user_location_type=user_location_type,
|
|
department_id=department_id,
|
|
user_pharmacy_id=user_pharmacy_id,
|
|
user_location_id=user_location_id,
|
|
post_id=post_id,
|
|
)
|
|
emp.clean()
|
|
|
|
|
|
def _directory(type_code, code):
|
|
from core.models import Directory
|
|
return Directory.objects.get(directory_type__code=type_code, code=code)
|
|
|
|
|
|
def _ticket_location(hiring_request):
|
|
if hiring_request.user_location_type == 'pharmacy' and hiring_request.user_pharmacy_id:
|
|
return hiring_request.user_pharmacy
|
|
return hiring_request.user_location
|
|
|
|
|
|
def _build_ticket_description(hiring_request, intern):
|
|
loc_label = hiring_request.get_user_location_type_display()
|
|
if hiring_request.user_pharmacy_id:
|
|
loc_detail = str(hiring_request.user_pharmacy)
|
|
elif hiring_request.user_location_id:
|
|
loc_detail = str(hiring_request.user_location)
|
|
elif hiring_request.department_id:
|
|
loc_detail = hiring_request.department.name
|
|
else:
|
|
loc_detail = '—'
|
|
|
|
lines = [
|
|
f"{_('Кандидат')}: {hiring_request.last_name} {hiring_request.first_name} {hiring_request.middle_name}".strip(),
|
|
f"{_('Должность')}: {hiring_request.post.name if hiring_request.post_id else '—'}",
|
|
f"{_('Тип локации')}: {loc_label}",
|
|
f"{_('Локация / отдел')}: {loc_detail}",
|
|
f"{_('Дата выхода')}: {hiring_request.hire_date:%d.%m.%Y}",
|
|
f"{_('Испытательный срок')}: {hiring_request.trial_days} {_('дн.')}",
|
|
f"{_('Интерн-аккаунт')}: {intern.label}",
|
|
f"{_('Логин')}: {intern.username}",
|
|
f"{_('Пароль')}: {intern.password}",
|
|
'',
|
|
f"{_('Требуемые доступы')}:",
|
|
]
|
|
accesses = list(hiring_request.requested_access.all())
|
|
if accesses:
|
|
for access in accesses:
|
|
lines.append(f' • {access.name}')
|
|
else:
|
|
lines.append(f" {_('(не указано)')}")
|
|
if hiring_request.access_note:
|
|
lines.append('')
|
|
lines.append(f"{_('Примечание')}: {hiring_request.access_note}")
|
|
lines.append('')
|
|
lines.append(_(
|
|
'Подготовьте рабочее место и выдайте доступы к интерн-аккаунту на время '
|
|
'испытания. После утверждения HR будет создан персональный аккаунт.'
|
|
))
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def _create_access_ticket(hiring_request, requester, intern):
|
|
from apps.it_manager.models import Ticket
|
|
|
|
fio = f'{hiring_request.last_name} {hiring_request.first_name}'.strip()
|
|
ticket = Ticket.objects.create(
|
|
title=f"{_('Подготовка рабочего места')}: {fio}",
|
|
ticket_type=_directory('ticket_type', 'access'),
|
|
priority=_directory('ticket_priority', 'high'),
|
|
requester=requester,
|
|
location=_ticket_location(hiring_request),
|
|
description=_build_ticket_description(hiring_request, intern),
|
|
)
|
|
hiring_request.granted_ticket = ticket
|
|
hiring_request.save(update_fields=['granted_ticket', 'updated_at'])
|
|
return ticket
|
|
|
|
|
|
def _close_ticket(ticket):
|
|
if ticket is None:
|
|
return
|
|
closed = _directory('ticket_status', 'closed')
|
|
if ticket.status_id == closed.pk:
|
|
return
|
|
ticket.status = closed
|
|
ticket.save(update_fields=['status', 'closed_at', 'updated_at'])
|
|
|
|
|
|
@transaction.atomic
|
|
def create_hiring_request(*, requester, last_name, first_name, middle_name='',
|
|
candidate_email='', candidate_phone='',
|
|
user_location_type, department_id=None, user_pharmacy_id=None,
|
|
user_location_id=None, post_id=None,
|
|
hire_date, trial_days, requested_access_ids=None, access_note=''):
|
|
"""Створює заявку на випробування: виділяє інтерн-слот і IT Ticket.
|
|
Персональний Employee ще не створюється.
|
|
"""
|
|
department_id, user_pharmacy_id, user_location_id = _coerce_location_fields(
|
|
user_location_type, department_id, user_pharmacy_id, user_location_id,
|
|
)
|
|
_validate_hire_location(
|
|
user_location_type=user_location_type,
|
|
department_id=department_id,
|
|
user_pharmacy_id=user_pharmacy_id,
|
|
user_location_id=user_location_id,
|
|
post_id=post_id,
|
|
hire_date=hire_date,
|
|
)
|
|
requested_access_ids = _filter_access_ids_for_location(
|
|
user_location_type, requested_access_ids or [],
|
|
)
|
|
|
|
hiring_request = HiringRequest.objects.create(
|
|
last_name=last_name,
|
|
first_name=first_name,
|
|
middle_name=middle_name,
|
|
candidate_email=candidate_email,
|
|
candidate_phone=candidate_phone,
|
|
user_location_type=user_location_type,
|
|
department_id=department_id,
|
|
user_pharmacy_id=user_pharmacy_id,
|
|
user_location_id=user_location_id,
|
|
post_id=post_id,
|
|
hire_date=hire_date,
|
|
trial_days=trial_days,
|
|
access_note=access_note,
|
|
status='trial',
|
|
requested_by=requester,
|
|
)
|
|
if requested_access_ids:
|
|
hiring_request.requested_access.set(requested_access_ids)
|
|
|
|
intern = InternAccount.allocate(hiring_request)
|
|
_create_access_ticket(hiring_request, requester, intern)
|
|
return hiring_request
|
|
|
|
|
|
def _should_assign_office_email(user_location_type) -> bool:
|
|
return user_location_type == 'office'
|
|
|
|
|
|
def _office_email_for(username: str) -> str:
|
|
from apps.employees.models import EmployeeDefaults
|
|
domain = EmployeeDefaults.load().normalized_office_domain
|
|
return f'{username}@{domain}'
|
|
|
|
|
|
@transaction.atomic
|
|
def approve_hiring_request(hiring_request, decided_by, note=''):
|
|
locked = HiringRequest.objects.select_for_update().get(pk=hiring_request.pk)
|
|
if locked.status != 'trial':
|
|
return locked
|
|
|
|
username = _generate_username(locked.last_name, locked.first_name)
|
|
|
|
employee = Employee(
|
|
username=username,
|
|
last_name=locked.last_name,
|
|
first_name=locked.first_name,
|
|
middle_name=locked.middle_name,
|
|
personal_phone=locked.candidate_phone or '',
|
|
personal_email=locked.candidate_email or '',
|
|
is_active=True,
|
|
hire_date=locked.hire_date,
|
|
probation_end_date=timezone.localdate(),
|
|
user_location_type=locked.user_location_type,
|
|
department_id=locked.department_id,
|
|
user_pharmacy_id=locked.user_pharmacy_id,
|
|
user_location_id=locked.user_location_id,
|
|
post_id=locked.post_id,
|
|
ad_sync_enabled=False,
|
|
)
|
|
if _should_assign_office_email(locked.user_location_type):
|
|
employee.email = _office_email_for(username)
|
|
employee.set_unusable_password()
|
|
employee.full_clean()
|
|
employee.save()
|
|
|
|
from apps.employees.services.password_invite import create_password_invite
|
|
invite = create_password_invite(employee, created_by=decided_by)
|
|
|
|
locked.employee = employee
|
|
locked.issued_password = ''
|
|
locked.status = 'approved'
|
|
locked.decision_by = decided_by
|
|
locked.decision_at = timezone.now()
|
|
locked.decision_note = note
|
|
locked.save(update_fields=[
|
|
'employee', 'issued_password', 'status',
|
|
'decision_by', 'decision_at', 'decision_note', 'updated_at',
|
|
])
|
|
locked.release_intern()
|
|
_close_ticket(locked.granted_ticket)
|
|
|
|
from core.security import log_security_event
|
|
|
|
log_security_event(
|
|
'employee.created_via_hiring',
|
|
actor=decided_by,
|
|
target=str(employee.pk),
|
|
details=f'username={employee.username} location={employee.user_location_type} invite={invite.pk}',
|
|
)
|
|
locked._password_invite = invite # для view/шаблону (не зберігається в БД)
|
|
return locked
|
|
|
|
|
|
@transaction.atomic
|
|
def reject_hiring_request(hiring_request, decided_by, note=''):
|
|
locked = HiringRequest.objects.select_for_update().get(pk=hiring_request.pk)
|
|
if locked.status != 'trial':
|
|
return locked
|
|
|
|
locked.release_intern()
|
|
_close_ticket(locked.granted_ticket)
|
|
|
|
locked.status = 'rejected'
|
|
locked.decision_by = decided_by
|
|
locked.decision_at = timezone.now()
|
|
locked.decision_note = note
|
|
locked.save(update_fields=[
|
|
'status', 'decision_by', 'decision_at', 'decision_note', 'updated_at',
|
|
])
|
|
return locked
|
|
|
|
|
|
@transaction.atomic
|
|
def build_hiring_credentials_pdf(hiring_request, request=None):
|
|
"""PDF з логіном і посиланням на встановлення пароля."""
|
|
from apps.hiring.pdf import build_credentials_pdf
|
|
from apps.employees.services.password_invite import (
|
|
build_invite_absolute_url,
|
|
create_password_invite,
|
|
get_active_invite_for_employee,
|
|
)
|
|
|
|
locked = HiringRequest.objects.select_for_update().get(pk=hiring_request.pk)
|
|
if locked.status != 'approved' or locked.employee_id is None:
|
|
return None
|
|
|
|
invite = get_active_invite_for_employee(locked.employee)
|
|
if invite is None and not locked.employee.has_usable_password():
|
|
invite = create_password_invite(locked.employee, created_by=None)
|
|
|
|
invite_url = build_invite_absolute_url(request, invite) if invite else ''
|
|
locked = (
|
|
HiringRequest.objects.select_related('employee', 'post')
|
|
.prefetch_related('requested_access')
|
|
.get(pk=locked.pk)
|
|
)
|
|
return build_credentials_pdf(locked, invite_url=invite_url)
|
|
|
|
|
|
@transaction.atomic
|
|
def consume_issued_password_for_pdf(hiring_request):
|
|
"""Зворотна сумісність: PDF без одноразового пароля."""
|
|
return build_hiring_credentials_pdf(hiring_request)
|