1822 lines
71 KiB
Python
1822 lines
71 KiB
Python
from core.location_access import accessible_pharmacies, scoped_queryset, has_all_locations
|
||
from django.core.exceptions import ValidationError
|
||
from django.db.models import Q, Count, Prefetch, Exists, OuterRef
|
||
from django.http import HttpResponse, HttpResponseForbidden, JsonResponse, HttpResponseNotAllowed
|
||
from django.shortcuts import render, redirect, get_object_or_404
|
||
from django.urls import reverse
|
||
from django.contrib.auth import authenticate, login, logout
|
||
from django.contrib.auth.decorators import login_required
|
||
from django.core.cache import cache
|
||
from django.views.decorators.http import require_POST
|
||
from django.utils.http import url_has_allowed_host_and_scheme
|
||
from django.conf import settings
|
||
from apps.employees.models.base import Employee
|
||
from apps.locations.models.pharmacies import Pharmacy
|
||
from core.models.directories import Directory
|
||
from core.models import ConnectorConfig
|
||
from core.employee_visibility import can_view_employee_pii
|
||
from core.portal_access import hr_manage_required, section_required, user_can_manage_hr_portal
|
||
from core.utils.htmx import render_page
|
||
from core.utils.pagination import paginate
|
||
from django.utils.translation import gettext_lazy as _
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Create your views here.
|
||
|
||
@login_required
|
||
def index_view(request):
|
||
return render(request, 'index.html')
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def access_preview_view(request):
|
||
"""Переключатель «смотреть как роль» (только DEBUG + staff)."""
|
||
from core.access_preview import (
|
||
ACCESS_PREVIEW_PRESETS,
|
||
SESSION_KEY,
|
||
access_preview_available,
|
||
)
|
||
|
||
if not access_preview_available(request.user):
|
||
return HttpResponseForbidden('Access preview только в DEBUG для staff.')
|
||
|
||
preview_id = (request.POST.get('preview') or '').strip()
|
||
allowed = {p.id for p in ACCESS_PREVIEW_PRESETS}
|
||
if preview_id not in allowed:
|
||
preview_id = ''
|
||
|
||
if preview_id:
|
||
request.session[SESSION_KEY] = preview_id
|
||
else:
|
||
request.session.pop(SESSION_KEY, None)
|
||
|
||
next_url = request.POST.get('next') or request.META.get('HTTP_REFERER') or '/'
|
||
if not url_has_allowed_host_and_scheme(next_url, allowed_hosts={request.get_host()}):
|
||
next_url = '/'
|
||
return redirect(next_url)
|
||
|
||
|
||
@login_required
|
||
def dashboard_view(request):
|
||
from django.utils import timezone as dj_timezone
|
||
from core.dashboard import build_dashboard_context
|
||
|
||
context = {'today': dj_timezone.localdate()}
|
||
context.update(build_dashboard_context(request))
|
||
return render_page(request, 'dashboard.html', context)
|
||
|
||
|
||
@login_required
|
||
def dashboard_customize_view(request):
|
||
"""Настройка персональной раскладки дашборда: вкл/выкл + порядок виджетов."""
|
||
from core import dashboard as dash
|
||
|
||
if request.method == 'POST':
|
||
if request.POST.get('reset'):
|
||
dash.reset_layout(request.user)
|
||
return redirect('dashboard')
|
||
order = [k.strip() for k in (request.POST.get('order') or '').split(',') if k.strip()]
|
||
enabled = request.POST.getlist('enabled')
|
||
if not order: # JS отключён — берём порядок чекбоксов
|
||
order = enabled
|
||
widths = {k: request.POST.get(f'width_{k}', dash.DEFAULT_WIDTH) for k in enabled}
|
||
dash.save_layout(request.user, order, enabled, widths=widths)
|
||
return redirect('dashboard')
|
||
|
||
context = {'rows': dash.customize_rows(request.user)}
|
||
return render_page(request, 'dashboard_customize.html', context)
|
||
|
||
@login_required
|
||
def employees_view(request):
|
||
"""Список перенесён в HR-портал (read-only для всех, управление — для HR)."""
|
||
return redirect('hr_portal')
|
||
|
||
|
||
def _can_view_employee_pii(request, employee) -> bool:
|
||
return can_view_employee_pii(request.user, employee)
|
||
|
||
|
||
def _save_own_profile(employee, post, files):
|
||
"""Збереження полів, які співробітник редагує сам (тільки передані в POST)."""
|
||
from core.utils.pagination import PAGE_SIZE_CHOICES
|
||
|
||
if files.get('photo'):
|
||
employee.photo = files['photo']
|
||
|
||
if 'phone' in post:
|
||
employee.phone = post.get('phone', '').strip()
|
||
if 'personal_phone' in post:
|
||
employee.personal_phone = post.get('personal_phone', '').strip()
|
||
if 'personal_email' in post:
|
||
employee.personal_email = post.get('personal_email', '').strip()
|
||
if 'emergency_contact_name' in post:
|
||
employee.emergency_contact_name = post.get('emergency_contact_name', '').strip()
|
||
if 'emergency_contact_phone' in post:
|
||
employee.emergency_contact_phone = post.get('emergency_contact_phone', '').strip()
|
||
if 'emergency_contact_relation' in post:
|
||
employee.emergency_contact_relation = post.get('emergency_contact_relation', '').strip()
|
||
|
||
if 'default_page_size' in post:
|
||
try:
|
||
page_size = int(post.get('default_page_size', ''))
|
||
except (TypeError, ValueError):
|
||
page_size = None
|
||
if page_size in PAGE_SIZE_CHOICES:
|
||
employee.default_page_size = page_size
|
||
|
||
unrelated_fields = [
|
||
f.name for f in Employee._meta.fields if f.name not in PROFILE_EDITABLE_FIELDS
|
||
]
|
||
employee.full_clean(exclude=unrelated_fields)
|
||
employee.save()
|
||
|
||
|
||
def _employee_profile_context(request, employee, *, active_tab: str = 'overview', **extra):
|
||
from core.utils.pagination import PAGE_SIZE_CHOICES
|
||
|
||
can_view_pii = _can_view_employee_pii(request, employee)
|
||
is_own_profile = request.user.pk == employee.pk
|
||
allowed_tabs = ['overview', 'contacts', 'work']
|
||
if is_own_profile:
|
||
allowed_tabs.append('settings')
|
||
if active_tab not in allowed_tabs:
|
||
active_tab = 'overview'
|
||
if active_tab == 'contacts' and not can_view_pii and not is_own_profile:
|
||
active_tab = 'overview'
|
||
|
||
context = {
|
||
'employee': employee,
|
||
'can_view_pii': can_view_pii,
|
||
'is_own_profile': is_own_profile,
|
||
'can_edit': request.user.is_staff,
|
||
'profile_active_tab': active_tab,
|
||
'page_size_choices': PAGE_SIZE_CHOICES,
|
||
'subordinates': employee.subordinates.filter(is_active=True).select_related('post')
|
||
if employee.is_manager else [],
|
||
'manager_chain': employee.get_manager_chain(),
|
||
'saved': False,
|
||
'errors': None,
|
||
}
|
||
|
||
from apps.tasks.models import Task
|
||
|
||
employee_tasks = (
|
||
Task.objects.filter(Q(responsible=employee) | Q(co_executors=employee))
|
||
.exclude(status='cancelled')
|
||
.distinct().order_by('-created_at')
|
||
)
|
||
shown_limit = 10
|
||
context['recent_tasks'] = list(employee_tasks[:shown_limit])
|
||
context['recent_tasks_total'] = employee_tasks.count()
|
||
context['recent_tasks_hidden'] = max(0, context['recent_tasks_total'] - shown_limit)
|
||
|
||
if can_view_pii:
|
||
context['last_attendance'] = (
|
||
employee.attendance_logs.select_related('location').order_by('-timestamp').first()
|
||
)
|
||
else:
|
||
context['last_attendance'] = None
|
||
|
||
context.update(extra)
|
||
|
||
if is_own_profile:
|
||
from core.portal_access import user_can_manage_hr_portal
|
||
context.update(_hr_requests_context(request, user_can_manage_hr_portal(request.user)))
|
||
context.update(_twofa_settings_context(employee))
|
||
|
||
return context
|
||
|
||
|
||
def _twofa_settings_context(user):
|
||
from core.security import user_account_mandatory_2fa, user_can_disable_2fa
|
||
|
||
return {
|
||
'twofa_enabled': bool(user.totp_enabled),
|
||
'twofa_mandatory': user_account_mandatory_2fa(user),
|
||
'twofa_can_disable': user_can_disable_2fa(user),
|
||
'twofa_backup_remaining': len(user.totp_backup_codes or []),
|
||
}
|
||
|
||
|
||
def _profile_settings_response(request, **extra):
|
||
employee = request.user
|
||
employee = (
|
||
Employee.objects.select_related(
|
||
'post', 'department', 'user_organization',
|
||
'user_pharmacy__organization', 'user_location', 'manager__post',
|
||
).prefetch_related('groups').get(pk=employee.pk)
|
||
)
|
||
context = _employee_profile_context(
|
||
request, employee, active_tab='settings', **extra,
|
||
)
|
||
return render_page(request, 'employee-profile.html', context)
|
||
|
||
|
||
@login_required
|
||
def employee_detail_view(request, pk):
|
||
employee = get_object_or_404(
|
||
Employee.objects.select_related(
|
||
'post', 'department', 'user_organization',
|
||
'user_pharmacy__organization', 'user_location', 'manager__post',
|
||
).prefetch_related('groups'),
|
||
pk=pk,
|
||
)
|
||
is_own_profile = request.user.pk == employee.pk
|
||
active_tab = (request.GET.get('tab') or request.POST.get('profile_tab') or 'overview').strip()
|
||
|
||
if request.method == 'POST':
|
||
if not is_own_profile:
|
||
from django.http import HttpResponseForbidden
|
||
return HttpResponseForbidden()
|
||
saved = False
|
||
errors = None
|
||
try:
|
||
_save_own_profile(employee, request.POST, request.FILES)
|
||
saved = True
|
||
except ValidationError as exc:
|
||
errors = list(exc.messages)
|
||
if active_tab not in ('contacts', 'settings'):
|
||
active_tab = 'contacts'
|
||
context = _employee_profile_context(
|
||
request, employee, active_tab=active_tab, saved=saved, errors=errors,
|
||
)
|
||
return render_page(request, 'employee-profile.html', context)
|
||
|
||
context = _employee_profile_context(request, employee, active_tab=active_tab)
|
||
return render_page(request, 'employee-profile.html', context)
|
||
|
||
|
||
def build_region_tree(region_nodes):
|
||
allowed_types = {'region', 'district', 'city', 'city_district'}
|
||
tree = []
|
||
|
||
def lookup_node(list_nodes, node_id):
|
||
for item in list_nodes:
|
||
if item['id'] == node_id:
|
||
return item
|
||
return None
|
||
|
||
for region in region_nodes:
|
||
ancestors = [node for node in list(region.get_ancestors()) if node.directory_type.code in allowed_types]
|
||
parent_list = tree
|
||
for ancestor in ancestors:
|
||
existing = lookup_node(parent_list, ancestor.id)
|
||
if existing is None:
|
||
existing = {
|
||
'id': ancestor.id,
|
||
'name': ancestor.name,
|
||
'children': []
|
||
}
|
||
parent_list.append(existing)
|
||
parent_list = existing['children']
|
||
return tree
|
||
|
||
|
||
def pharmacies_page_context(request, **extra):
|
||
pharmacies = accessible_pharmacies(request.user).select_related(
|
||
'organization', 'location_region', 'head_pharmacist',
|
||
).order_by('organization__name', 'pharmacy_number')
|
||
active_count = pharmacies.filter(is_active=True).count()
|
||
regions = pharmacies.values_list('location_region__name', flat=True).distinct().order_by('location_region__name')
|
||
region_ids = sorted({rid for rid in pharmacies.values_list('location_region_id', flat=True).distinct() if rid})
|
||
region_nodes = Directory.objects.filter(id__in=region_ids).select_related('parent', 'directory_type')
|
||
context = {
|
||
'pharmacies': pharmacies,
|
||
'active_count': active_count,
|
||
'regions': regions,
|
||
'region_tree': build_region_tree(region_nodes),
|
||
}
|
||
context.update(extra)
|
||
return context
|
||
|
||
|
||
@login_required
|
||
@section_required('pharmacies')
|
||
def pharmacies_view(request):
|
||
return render_page(request, 'pharmacies.html', pharmacies_page_context(request))
|
||
|
||
|
||
@login_required
|
||
def dislocation_view(request):
|
||
"""Дислокація: аптеки на карті/у таблиці разом з їхнім персоналом (замінює вкладку «Аптеки» на /structure/).
|
||
|
||
Справочник сети открыт всем сотрудникам с доступом к разделу — без скоупа
|
||
accessible_pharmacies (в отличие от /pharmacies/ и операционных разделов).
|
||
"""
|
||
from apps.it_manager.models import Equipment, InternetContract
|
||
from apps.locations.dislocation_columns import (
|
||
all_column_defs, column_defs_for, resolve_columns_for_user,
|
||
)
|
||
from apps.locations.models import DislocationColumnPreset, Pharmacy
|
||
from apps.power_assets.models import PowerAsset
|
||
|
||
mikrotik_q = (
|
||
Q(equipment_model__brand__name__icontains='mikrotik')
|
||
| Q(equipment_model__name__icontains='mikrotik')
|
||
| Q(equipment_model__brand__name__icontains='мікротік')
|
||
)
|
||
has_generator = Exists(
|
||
PowerAsset.objects.filter(
|
||
location_id=OuterRef('pk'),
|
||
asset_type__code='generator',
|
||
).exclude(status__code='written_off')
|
||
)
|
||
has_inverter = Exists(
|
||
PowerAsset.objects.filter(
|
||
location_id=OuterRef('pk'),
|
||
asset_type__code='inverter',
|
||
).exclude(status__code='written_off')
|
||
)
|
||
has_mikrotik = Exists(
|
||
Equipment.objects.filter(
|
||
location_id=OuterRef('pk'),
|
||
equipment_model__equipment_type__code='router',
|
||
)
|
||
.filter(mikrotik_q)
|
||
.exclude(status__code='written_off')
|
||
)
|
||
has_gpon = Exists(
|
||
InternetContract.objects.filter(
|
||
location_id=OuterRef('pk'),
|
||
status__code='active',
|
||
is_framework=False,
|
||
).filter(
|
||
Q(connection_type__code='gpon')
|
||
| Q(connection_type__name__icontains='gpon')
|
||
| Q(connection_type__name__icontains='гпон')
|
||
)
|
||
)
|
||
|
||
active_internet_qs = (
|
||
InternetContract.objects
|
||
.filter(status__code='active', is_framework=False)
|
||
.select_related('provider', 'connection_type')
|
||
.order_by('provider__name', 'contract_number')
|
||
)
|
||
|
||
pharmacies = (
|
||
Pharmacy.objects.all()
|
||
.select_related(
|
||
'organization', 'location_region', 'head_pharmacist',
|
||
'curator', 'responsible_manager',
|
||
)
|
||
.prefetch_related(
|
||
Prefetch(
|
||
'pharmacy_employees',
|
||
queryset=Employee.objects.filter(is_active=True).select_related('post'),
|
||
to_attr='active_staff',
|
||
),
|
||
Prefetch(
|
||
'internet_contracts',
|
||
queryset=active_internet_qs,
|
||
to_attr='active_internet',
|
||
),
|
||
)
|
||
.annotate(
|
||
staff_count=Count('pharmacy_employees', filter=Q(pharmacy_employees__is_active=True)),
|
||
has_generator=has_generator,
|
||
has_inverter=has_inverter,
|
||
has_mikrotik_router=has_mikrotik,
|
||
has_gpon=has_gpon,
|
||
)
|
||
.order_by('organization__name', 'pharmacy_number')
|
||
)
|
||
pharmacies = list(pharmacies)
|
||
for pharmacy in pharmacies:
|
||
if pharmacy.staff_count == 0:
|
||
pharmacy.staff_status = 'danger'
|
||
elif not pharmacy.head_pharmacist_id:
|
||
pharmacy.staff_status = 'warning'
|
||
else:
|
||
pharmacy.staff_status = 'ok'
|
||
pharmacy.has_map_location = pharmacy.has_coordinates
|
||
contracts = getattr(pharmacy, 'active_internet', []) or []
|
||
pharmacy.dislo_provider = ', '.join(dict.fromkeys(
|
||
c.provider.name for c in contracts if c.provider_id
|
||
))
|
||
pharmacy.dislo_contract = ', '.join(
|
||
c.contract_number for c in contracts if c.contract_number
|
||
)
|
||
|
||
active_count = sum(1 for p in pharmacies if p.is_active)
|
||
pharmacies_no_location = sum(1 for p in pharmacies if not p.has_map_location)
|
||
region_ids = sorted({p.location_region_id for p in pharmacies if p.location_region_id})
|
||
region_nodes = Directory.objects.filter(id__in=region_ids).select_related('parent', 'directory_type')
|
||
region_tree = build_region_tree(region_nodes)
|
||
|
||
# --- Офіс/склад: телефонний довідник по відділах ---
|
||
office_employees = list(
|
||
Employee.objects.filter(is_active=True, user_location_type__in=['office', 'warehouse'])
|
||
.select_related('post', 'department')
|
||
.order_by('last_name', 'first_name')
|
||
)
|
||
by_dept = {}
|
||
for emp in office_employees:
|
||
by_dept.setdefault(emp.department_id, []).append(emp)
|
||
|
||
office_departments = list(
|
||
Directory.objects.filter(directory_type__code='department', is_active=True)
|
||
.order_by('_path')
|
||
)
|
||
for dept in office_departments:
|
||
dept.indent_depth = (dept._path or '').count('.')
|
||
dept.staff = by_dept.get(dept.id, [])
|
||
office_departments = [d for d in office_departments if d.staff]
|
||
office_no_department = by_dept.get(None, [])
|
||
|
||
# --- Сервисы / поставщики (location_type=vendor) ---
|
||
from apps.locations.forms import PortalVendorForm
|
||
from apps.locations.models.base import Location, location_sort_key
|
||
from apps.locations.views import user_can_manage_vendors
|
||
|
||
vendors = sorted(
|
||
Location.objects.filter(location_type='vendor')
|
||
.select_related('location_region')
|
||
.order_by('name', 'id'),
|
||
key=location_sort_key,
|
||
)
|
||
can_manage_vendors = user_can_manage_vendors(request.user)
|
||
vendor_form = getattr(request, '_dislo_vendor_form', None)
|
||
if vendor_form is None and can_manage_vendors:
|
||
vendor_form = PortalVendorForm()
|
||
|
||
visible_keys = resolve_columns_for_user(request.user)
|
||
visible_columns = column_defs_for(visible_keys)
|
||
visible_set = set(visible_keys)
|
||
all_defs = all_column_defs()
|
||
panel_column_defs = list(visible_columns) + [d for d in all_defs if d.key not in visible_set]
|
||
presets = list(
|
||
DislocationColumnPreset.objects
|
||
.select_related('department')
|
||
.order_by('name')
|
||
)
|
||
can_manage_presets = bool(getattr(request.user, 'is_staff', False))
|
||
departments = []
|
||
if can_manage_presets:
|
||
departments = list(
|
||
Directory.objects.filter(directory_type__code='department', is_active=True)
|
||
.order_by('_path', 'name')
|
||
)
|
||
|
||
context = {
|
||
'pharmacies': pharmacies,
|
||
'active_count': active_count,
|
||
'organizations_count': len({p.organization_id for p in pharmacies if p.organization_id}),
|
||
'pharmacies_no_head': sum(1 for p in pharmacies if not p.head_pharmacist_id),
|
||
'pharmacies_empty': sum(1 for p in pharmacies if p.staff_count == 0),
|
||
'pharmacies_no_location': pharmacies_no_location,
|
||
'region_tree': region_tree,
|
||
'office_departments': office_departments,
|
||
'office_no_department': office_no_department,
|
||
'office_employees_count': len(office_employees),
|
||
'vendors': vendors,
|
||
'vendors_active_count': sum(1 for v in vendors if v.is_active),
|
||
'can_manage_vendors': can_manage_vendors,
|
||
'vendor_form': vendor_form,
|
||
'vendor_form_message': getattr(request, '_dislo_vendor_message', None),
|
||
'vendor_form_error': getattr(request, '_dislo_vendor_error', None),
|
||
'dislo_active_tab': (
|
||
getattr(request, '_dislo_tab', None)
|
||
or (request.GET.get('tab') or '').strip()
|
||
or 'pharmacy'
|
||
),
|
||
'visible_columns': visible_columns,
|
||
'visible_column_keys': visible_keys,
|
||
'all_column_defs': all_defs,
|
||
'panel_column_defs': panel_column_defs,
|
||
'dislocation_presets': presets,
|
||
'dislocation_presets_payload': [
|
||
{'code': p.code, 'columns': list(p.columns or []), 'is_default': p.is_default}
|
||
for p in presets
|
||
],
|
||
'dislo_colspan': len(visible_columns) + 1,
|
||
'can_manage_dislo_presets': can_manage_presets,
|
||
'dislocation_departments': departments,
|
||
}
|
||
return render_page(request, 'dislocation.html', context)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def dislocation_columns_view(request):
|
||
"""Сохраняет персональный набор колонок дислокации (или применяет пресет)."""
|
||
from apps.locations.dislocation_columns import normalize_columns
|
||
from apps.locations.models import DislocationColumnPreset
|
||
|
||
preset_code = (request.POST.get('preset_code') or '').strip()
|
||
if preset_code:
|
||
preset = get_object_or_404(DislocationColumnPreset, code=preset_code)
|
||
columns = normalize_columns(preset.columns)
|
||
else:
|
||
columns = normalize_columns(request.POST.getlist('columns'))
|
||
|
||
Employee.objects.filter(pk=request.user.pk).update(dislocation_columns=columns)
|
||
request.user.dislocation_columns = columns
|
||
return dislocation_view(request)
|
||
|
||
|
||
def _unique_dislo_preset_code(base: str, *, exclude_pk=None) -> str:
|
||
from django.utils.text import slugify
|
||
from apps.locations.models import DislocationColumnPreset
|
||
|
||
code = slugify(base) or 'preset'
|
||
code = code[:50]
|
||
candidate = code
|
||
n = 2
|
||
qs = DislocationColumnPreset.objects.all()
|
||
if exclude_pk:
|
||
qs = qs.exclude(pk=exclude_pk)
|
||
while qs.filter(code=candidate).exists():
|
||
suffix = f'-{n}'
|
||
candidate = f'{code[:50 - len(suffix)]}{suffix}'
|
||
n += 1
|
||
return candidate
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def dislocation_column_presets_view(request):
|
||
"""CRUD пресетов колонок дислокации (только staff)."""
|
||
from django.core.exceptions import PermissionDenied
|
||
from apps.locations.dislocation_columns import normalize_columns
|
||
from apps.locations.models import DislocationColumnPreset
|
||
|
||
if not getattr(request.user, 'is_staff', False):
|
||
raise PermissionDenied
|
||
|
||
action = (request.POST.get('action') or '').strip()
|
||
columns = normalize_columns(request.POST.getlist('columns'))
|
||
name = (request.POST.get('name') or '').strip()
|
||
code_raw = (request.POST.get('code') or '').strip()
|
||
preset_code = (request.POST.get('preset_code') or '').strip()
|
||
department_id = (request.POST.get('department_id') or '').strip()
|
||
is_default = request.POST.get('is_default') in ('1', 'true', 'on', 'yes')
|
||
|
||
department = None
|
||
if department_id.isdigit():
|
||
department = Directory.objects.filter(
|
||
pk=int(department_id),
|
||
directory_type__code='department',
|
||
is_active=True,
|
||
).first()
|
||
|
||
if action == 'create':
|
||
if not name:
|
||
return dislocation_view(request)
|
||
code = _unique_dislo_preset_code(code_raw or name)
|
||
DislocationColumnPreset.objects.create(
|
||
name=name,
|
||
code=code,
|
||
columns=columns,
|
||
department=department,
|
||
is_default=is_default,
|
||
)
|
||
elif action == 'update':
|
||
preset = get_object_or_404(DislocationColumnPreset, code=preset_code)
|
||
preset.columns = columns
|
||
if name:
|
||
preset.name = name
|
||
if code_raw and code_raw != preset.code:
|
||
preset.code = _unique_dislo_preset_code(code_raw, exclude_pk=preset.pk)
|
||
preset.department = department
|
||
preset.is_default = is_default
|
||
preset.save()
|
||
elif action == 'delete':
|
||
get_object_or_404(DislocationColumnPreset, code=preset_code).delete()
|
||
|
||
return dislocation_view(request)
|
||
|
||
@login_required
|
||
def calendar_view(request):
|
||
from apps.visit_manager.views_calendar import calendar_view as _calendar_view
|
||
return _calendar_view(request)
|
||
|
||
@login_required
|
||
@section_required('timetrack')
|
||
def timetrack_view(request):
|
||
from core.accounting_portal import render_accounting_portal
|
||
return render_accounting_portal(request, 'timetrack')
|
||
|
||
|
||
def _timetrack_context(request):
|
||
"""Облік робочого часу: агрегація змін. SoT — PVMS, якщо увімкнено."""
|
||
import datetime
|
||
from django.conf import settings
|
||
from django.utils import timezone
|
||
from django.db.models import Sum, Count, Q
|
||
|
||
UKR_MONTHS = ['', 'Січень', 'Лютий', 'Березень', 'Квітень', 'Травень', 'Червень',
|
||
'Липень', 'Серпень', 'Вересень', 'Жовтень', 'Листопад', 'Грудень']
|
||
|
||
today = timezone.localdate()
|
||
month_str = request.GET.get('month') or today.strftime('%Y-%m')
|
||
try:
|
||
year, month = (int(x) for x in month_str.split('-'))
|
||
period_start = datetime.date(year, month, 1)
|
||
except (ValueError, TypeError):
|
||
period_start = today.replace(day=1)
|
||
year, month = period_start.year, period_start.month
|
||
month_str = period_start.strftime('%Y-%m')
|
||
|
||
next_month_start = (datetime.date(year + 1, 1, 1) if month == 12
|
||
else datetime.date(year, month + 1, 1))
|
||
period_end = next_month_start - datetime.timedelta(days=1)
|
||
prev_month = (period_start - datetime.timedelta(days=1)).strftime('%Y-%m')
|
||
next_month = next_month_start.strftime('%Y-%m')
|
||
|
||
location_type = request.GET.get('location_type', '')
|
||
rows, counts, stats = None, None, None
|
||
source = 'acore'
|
||
|
||
use_pvms = bool(getattr(settings, 'PVMS_SYNC_ENABLED', False))
|
||
sot = bool(getattr(settings, 'PVMS_ATTENDANCE_SOT', False))
|
||
if use_pvms:
|
||
if not has_all_locations(request.user):
|
||
from django.core.exceptions import PermissionDenied
|
||
raise PermissionDenied('Сводка PVMS требует доступа ко всей сети.')
|
||
try:
|
||
from apps.portal_connector.pvms_client import PvmsClient
|
||
client = PvmsClient()
|
||
if client.enabled:
|
||
data = client.timetrack(
|
||
period_start.isoformat(),
|
||
period_end.isoformat(),
|
||
location_type=location_type if location_type in ('office', 'pharmacy', 'warehouse') else '',
|
||
)
|
||
rows = data.get('rows') or []
|
||
counts = data.get('counts') or {'all': 0, 'office': 0, 'pharmacy': 0}
|
||
stats = data.get('stats') or {
|
||
'employees': 0, 'shifts': 0, 'workdays': 0, 'hours': 0, 'incomplete': 0,
|
||
}
|
||
source = 'pvms'
|
||
except Exception:
|
||
if sot:
|
||
rows, counts, stats = [], {'all': 0, 'office': 0, 'pharmacy': 0}, {
|
||
'employees': 0, 'shifts': 0, 'workdays': 0, 'hours': 0, 'incomplete': 0,
|
||
}
|
||
source = 'pvms-error'
|
||
else:
|
||
rows = None
|
||
|
||
if rows is None:
|
||
from apps.visit_manager.models import WorkShift
|
||
|
||
shifts = (scoped_queryset(request.user, WorkShift.objects.all())
|
||
.filter(date__gte=period_start, date__lte=period_end)
|
||
.select_related('employee', 'location'))
|
||
if location_type in ('office', 'pharmacy', 'warehouse'):
|
||
shifts = shifts.filter(location__location_type=location_type)
|
||
|
||
per_emp = (shifts.values('employee_id', 'employee__last_name', 'employee__first_name')
|
||
.annotate(total=Sum('duration'),
|
||
days=Count('date', distinct=True),
|
||
incomplete=Count('id', filter=Q(left_at__isnull=True)))
|
||
.order_by('employee__last_name', 'employee__first_name'))
|
||
|
||
rows, total_hours = [], 0.0
|
||
for r in per_emp:
|
||
dur = r['total']
|
||
hours = round(dur.total_seconds() / 3600, 1) if dur else 0.0
|
||
total_hours += hours
|
||
rows.append({
|
||
'employee_id': r['employee_id'],
|
||
'name': f"{r['employee__last_name']} {r['employee__first_name']}".strip(),
|
||
'hours': hours,
|
||
'days': r['days'],
|
||
'incomplete': r['incomplete'],
|
||
'avg': round(hours / r['days'], 1) if r['days'] else 0.0,
|
||
})
|
||
|
||
base = scoped_queryset(request.user, WorkShift.objects.all()).filter(date__gte=period_start, date__lte=period_end)
|
||
counts = {
|
||
'all': base.values('employee_id').distinct().count(),
|
||
'office': base.filter(location__location_type='office').values('employee_id').distinct().count(),
|
||
'pharmacy': base.filter(location__location_type='pharmacy').values('employee_id').distinct().count(),
|
||
}
|
||
stats = {
|
||
'employees': len(rows),
|
||
'shifts': shifts.count(),
|
||
'workdays': shifts.values('date').distinct().count(),
|
||
'hours': round(total_hours, 1),
|
||
'incomplete': shifts.filter(left_at__isnull=True).count(),
|
||
}
|
||
|
||
fg, muted, warn = 'var(--color-foreground)', 'var(--color-muted-foreground)', 'var(--color-warning)'
|
||
metrics = [
|
||
{'label': _('Рабочих дней'), 'value': stats['workdays'], 'hint': _('с отметками'), 'color': fg},
|
||
{'label': _('Сотрудников'), 'value': stats['employees'], 'hint': _('отмечались'), 'color': fg},
|
||
{'label': _('Смен'), 'value': stats['shifts'], 'hint': _('за месяц'), 'color': fg},
|
||
{'label': _('Незакрытых'), 'value': stats['incomplete'],'hint': _('без ухода'), 'color': warn if stats['incomplete'] else muted},
|
||
{'label': _('Всего часов'), 'value': stats['hours'], 'hint': _('отработано'), 'color': fg},
|
||
]
|
||
|
||
page = paginate(request, rows)
|
||
context = {
|
||
'page': page,
|
||
'month': month_str,
|
||
'month_label': f'{UKR_MONTHS[month]} {year}',
|
||
'prev_month': prev_month,
|
||
'next_month': next_month,
|
||
'location_type': location_type,
|
||
'counts': counts,
|
||
'stats': stats,
|
||
'metrics': metrics,
|
||
'timetrack_source': source,
|
||
}
|
||
return context
|
||
|
||
@login_required
|
||
@section_required('reports')
|
||
def reports_view(request):
|
||
"""Совместимость: основной хаб — apps.reports.views.reports_hub_view."""
|
||
from apps.reports.views import reports_hub_view
|
||
return reports_hub_view(request)
|
||
|
||
@login_required
|
||
def documents_view(request):
|
||
from apps.nextcloud_connector.access import (
|
||
accessible_sections, department_tree_for_user, people_picker_payload,
|
||
)
|
||
from django.contrib.auth.models import Group
|
||
ctx = {
|
||
'sections': accessible_sections(request.user),
|
||
'department_tree': department_tree_for_user(request.user),
|
||
'empty_ids': set(),
|
||
}
|
||
ctx.update(people_picker_payload())
|
||
ctx['group_options'] = list(
|
||
Group.objects.order_by('name').values('pk', 'name')
|
||
)
|
||
return render_page(request, 'documents.html', ctx)
|
||
|
||
@login_required
|
||
def recruiting_view(request):
|
||
"""Рекрутинг перенесён во вкладку «Вакансии» HR-портала."""
|
||
return redirect('/hr-portal/?tab=vacancies')
|
||
|
||
@login_required
|
||
@section_required('structure')
|
||
def structure_view(request):
|
||
"""Структура офісу: відділи. Аптеки з їхнім персоналом — на /dislocation/."""
|
||
departments = list(
|
||
Directory.objects.filter(directory_type__code='department', is_active=True)
|
||
.order_by('_path')
|
||
)
|
||
office_employees = list(
|
||
Employee.objects.filter(is_active=True, user_location_type='office')
|
||
.select_related('post')
|
||
)
|
||
by_dept = {}
|
||
for emp in office_employees:
|
||
by_dept.setdefault(emp.department_id, []).append(emp)
|
||
|
||
palette = ['#3b82f6', '#8b5cf6', '#f59e0b', '#10b981', '#ec4899', '#06b6d4', '#64748b']
|
||
for i, dept in enumerate(departments):
|
||
emps = by_dept.get(dept.id, [])
|
||
dept.color = palette[i % len(palette)]
|
||
dept.indent_depth = (dept._path or '').count('.')
|
||
dept.employees_count = len(emps)
|
||
dept.posts_count = len({e.post_id for e in emps if e.post_id})
|
||
# Керівник: у нього є підлеглі у відділі, а його керівник — поза відділом
|
||
ids = {e.pk for e in emps}
|
||
managers_in_dept = {e.manager_id for e in emps if e.manager_id in ids}
|
||
candidates = [e for e in emps if e.pk in managers_in_dept]
|
||
dept.head = (
|
||
next((e for e in candidates if e.manager_id not in ids), None)
|
||
or (candidates[0] if candidates else None)
|
||
)
|
||
|
||
pharmacy_employees_count = Employee.objects.filter(
|
||
is_active=True, user_location_type='pharmacy',
|
||
).count()
|
||
|
||
context = {
|
||
'departments': departments,
|
||
'office_count': len(office_employees),
|
||
'office_no_dept': sum(1 for e in office_employees if not e.department_id),
|
||
'pharmacy_employees_count': pharmacy_employees_count,
|
||
}
|
||
return render_page(request, 'structure.html', context)
|
||
|
||
@login_required
|
||
def settings_view(request):
|
||
return redirect(f"{reverse('employee_detail', args=[request.user.pk])}?tab=settings")
|
||
|
||
|
||
# Поля, які співробітник редагує сам на /profile/. Кадрові поля (посада,
|
||
# відділ, дата прийому, username/робочий email тощо) — лише через Django admin.
|
||
PROFILE_EDITABLE_FIELDS = [
|
||
'photo', 'phone', 'personal_phone', 'personal_email',
|
||
'emergency_contact_name', 'emergency_contact_phone', 'emergency_contact_relation',
|
||
'default_page_size',
|
||
]
|
||
|
||
|
||
@login_required
|
||
def profile_view(request):
|
||
"""Редирект на объединённую карточку сотрудника; POST — сохранение своих данных."""
|
||
if request.method == 'POST':
|
||
return employee_detail_view(request, pk=request.user.pk)
|
||
tab = (request.GET.get('tab') or 'contacts').strip()
|
||
from django.shortcuts import redirect
|
||
from django.urls import reverse
|
||
return redirect(f"{reverse('employee_detail', args=[request.user.pk])}?tab={tab}")
|
||
|
||
|
||
def _build_chat_context(request):
|
||
from apps.mattermost_connector.models import MattermostUser
|
||
from apps.mattermost_connector.services.token_service import ensure_personal_access_token
|
||
|
||
try:
|
||
cfg = ConnectorConfig.get_for('mattermost')
|
||
public_url = ''
|
||
if cfg:
|
||
public_url = (cfg.extra_config or {}).get('public_url', '').strip().rstrip('/')
|
||
|
||
if not public_url:
|
||
return {'chat_state': 'not_configured'}
|
||
|
||
if not request.user.mattermost_user_id:
|
||
return {'chat_state': 'not_synced'}
|
||
|
||
token = ensure_personal_access_token(request.user)
|
||
if not token:
|
||
return {'chat_state': 'token_error'}
|
||
|
||
colleagues = [
|
||
{
|
||
'mattermostId': row['mattermost_id'],
|
||
'name': (f"{row['django_user__first_name']} {row['django_user__last_name']}".strip()
|
||
or row['django_user__username']),
|
||
}
|
||
for row in MattermostUser.objects.filter(is_active=True)
|
||
.exclude(django_user=request.user)
|
||
.select_related('django_user')
|
||
.values('mattermost_id', 'django_user__first_name', 'django_user__last_name', 'django_user__username')
|
||
]
|
||
|
||
ws_url = public_url.replace('https://', 'wss://', 1).replace('http://', 'ws://', 1) + '/api/v4/websocket'
|
||
return {
|
||
'chat_state': 'ready',
|
||
'chat_config': {
|
||
# Same-origin proxy — иначе браузер упирается в CORS Mattermost.
|
||
'apiBase': '/chat/api',
|
||
'wsUrl': ws_url,
|
||
'token': token,
|
||
'currentUserId': request.user.mattermost_user_id,
|
||
},
|
||
'chat_colleagues': colleagues,
|
||
}
|
||
except Exception:
|
||
logger.exception('Ошибка построения контекста чата для user_id=%s', getattr(request.user, 'pk', None))
|
||
return {'chat_state': 'token_error'}
|
||
|
||
|
||
@login_required
|
||
def chat_view(request):
|
||
return render_page(request, 'chat.html', _build_chat_context(request))
|
||
|
||
|
||
@login_required
|
||
def chat_sync_now_view(request):
|
||
if request.method != 'POST':
|
||
return HttpResponseNotAllowed(['POST'])
|
||
|
||
from apps.mattermost_connector.services.sync_service import sync_employee
|
||
sync_employee(request.user)
|
||
request.user.refresh_from_db()
|
||
return render_page(request, 'chat.html', _build_chat_context(request))
|
||
|
||
|
||
def _static_page_view(template_name, section=None):
|
||
"""Фабрика простых страниц: HTMX-фрагмент или полный layout"""
|
||
@login_required
|
||
def view(request):
|
||
return render_page(request, template_name)
|
||
if section:
|
||
return section_required(section)(view)
|
||
return view
|
||
|
||
|
||
email_view = _static_page_view('email.html', 'email') # legacy; route in mail_connector.urls
|
||
purchases_view = _static_page_view('purchases.html', 'purchases')
|
||
|
||
|
||
@login_required
|
||
def pharmacy_detail_view(request, pk):
|
||
"""Аптека целиком: реквизиты, лицензия, адрес/карта + операційні вкладки.
|
||
|
||
Публічне (будь-який logged-in): основна картка, контакти, ліцензія.
|
||
Операційні блоки — за політикою PharmacyCardVisibility (адмінка).
|
||
Редагування — за полем edit_pharmacy тієї ж політики.
|
||
"""
|
||
from apps.locations.models import Pharmacy
|
||
from apps.locations.views import _pharmacy_detail_context
|
||
|
||
pharmacy = get_object_or_404(
|
||
accessible_pharmacies(request.user).select_related(
|
||
'organization', 'location_region', 'head_pharmacist__post',
|
||
'responsible_manager', 'curator',
|
||
).prefetch_related('phones'),
|
||
pk=pk,
|
||
)
|
||
edit_mode = request.GET.get('edit') == '1'
|
||
context = _pharmacy_detail_context(request, pharmacy, edit_mode=edit_mode)
|
||
return render_page(request, 'pharmacy-detail.html', context)
|
||
|
||
|
||
def _hr_department_tree():
|
||
"""Дерево відділів {id, name, count, children: [...]} для деревовидного фільтра HR-порталу."""
|
||
depts = list(
|
||
Directory.objects.filter(directory_type__code='department', is_active=True)
|
||
.select_related('parent').order_by('_path', 'name')
|
||
)
|
||
counts = dict(
|
||
Employee.objects.filter(is_active=True, department__isnull=False)
|
||
.values_list('department_id').annotate(c=Count('id'))
|
||
)
|
||
|
||
nodes = {d.pk: {'id': d.pk, 'name': d.name, 'count': counts.get(d.pk, 0), 'children': []} for d in depts}
|
||
roots = []
|
||
for d in depts:
|
||
node = nodes[d.pk]
|
||
parent_node = nodes.get(d.parent_id)
|
||
if parent_node is not None:
|
||
parent_node['children'].append(node)
|
||
else:
|
||
roots.append(node)
|
||
return roots
|
||
|
||
|
||
def _hr_requests_context(request, can_manage: bool) -> dict:
|
||
from apps.employees.models import EmployeeHrRequest
|
||
from apps.employees.services.hr_requests import (
|
||
available_hr_request_types,
|
||
build_leave_balance,
|
||
open_substitution_requests,
|
||
next_pending_vacation,
|
||
pending_hr_requests_count,
|
||
)
|
||
from django.utils import timezone as dj_timezone
|
||
|
||
year = dj_timezone.localdate().year
|
||
create_types = available_hr_request_types(request.user)
|
||
ctx = {
|
||
'hr_request_types': create_types,
|
||
'hr_request_types_all': EmployeeHrRequest.REQUEST_TYPES,
|
||
'my_hr_requests': list(
|
||
EmployeeHrRequest.objects.filter(employee=request.user)
|
||
.select_related(
|
||
'substitute',
|
||
'employee__user_pharmacy',
|
||
'employee__user_pharmacy__organization',
|
||
)
|
||
.order_by('-created_at')[:30]
|
||
),
|
||
'incoming_substitutions': open_substitution_requests(request.user),
|
||
'leave_balance': build_leave_balance(request.user, year=year),
|
||
'leave_balance_year': year,
|
||
'next_vacation_request': next_pending_vacation(request.user),
|
||
'hr_pending_requests': [],
|
||
'hr_pending_leave_count': 0,
|
||
'hr_request_form': {},
|
||
'hr_substitute_choices': [],
|
||
'can_decide_hr_requests': can_manage,
|
||
}
|
||
if can_manage:
|
||
# Подмены утверждает откликнувшийся сотрудник — в очереди HR остальные типы.
|
||
ctx['hr_pending_requests'] = list(
|
||
EmployeeHrRequest.objects.filter(status='pending')
|
||
.exclude(request_type='substitution')
|
||
.select_related(
|
||
'employee',
|
||
'substitute',
|
||
'employee__user_pharmacy',
|
||
'employee__user_pharmacy__organization',
|
||
)
|
||
.order_by('created_at')[:100]
|
||
)
|
||
ctx['hr_pending_leave_count'] = pending_hr_requests_count()
|
||
return ctx
|
||
|
||
|
||
def _parse_optional_date(raw: str):
|
||
from datetime import datetime
|
||
if not raw:
|
||
return None
|
||
try:
|
||
return datetime.strptime(raw.strip(), '%Y-%m-%d').date()
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _validation_error_messages(exc: ValidationError) -> list[str]:
|
||
if hasattr(exc, 'message_dict'):
|
||
try:
|
||
messages = []
|
||
for msgs in exc.message_dict.values():
|
||
messages.extend(msgs)
|
||
if messages:
|
||
return messages
|
||
except AttributeError:
|
||
pass
|
||
return list(exc.messages)
|
||
|
||
|
||
def _hr_request_form_from_post(post) -> dict:
|
||
return {
|
||
'type': (post.get('request_type') or '').strip(),
|
||
'date_from': (post.get('date_from') or '').strip(),
|
||
'date_to': (post.get('date_to') or '').strip(),
|
||
'note': post.get('note', ''),
|
||
'substitute_id': (post.get('substitute_id') or '').strip(),
|
||
}
|
||
|
||
|
||
def _hr_request_portal_response(request, *, form_open=False, form_data=None, **extra):
|
||
context = _hr_portal_context(request)
|
||
context['hr_active_tab'] = 'requests'
|
||
if form_open:
|
||
context['hr_request_form_open'] = True
|
||
if form_data is not None:
|
||
context['hr_request_form'] = form_data
|
||
context.update(extra)
|
||
return render_page(request, 'hr-portal.html', context)
|
||
|
||
|
||
def _hr_request_profile_response(request, *, form_open=False, form_data=None, **extra):
|
||
employee = request.user
|
||
employee = (
|
||
Employee.objects.select_related(
|
||
'post', 'department', 'user_organization',
|
||
'user_pharmacy__organization', 'user_location', 'manager__post',
|
||
).prefetch_related('groups').get(pk=employee.pk)
|
||
)
|
||
context = _employee_profile_context(
|
||
request, employee, active_tab='overview', **extra,
|
||
)
|
||
if form_open:
|
||
context['hr_request_form_open'] = True
|
||
if form_data is not None:
|
||
context['hr_request_form'] = form_data
|
||
return render_page(request, 'employee-profile.html', context)
|
||
|
||
|
||
def _hr_request_response(request, *, form_open=False, form_data=None, **extra):
|
||
if request.POST.get('return_to') == 'profile':
|
||
return _hr_request_profile_response(
|
||
request, form_open=form_open, form_data=form_data, **extra,
|
||
)
|
||
return _hr_request_portal_response(
|
||
request, form_open=form_open, form_data=form_data, **extra,
|
||
)
|
||
|
||
|
||
def _hr_portal_context(request):
|
||
from django.utils import timezone as dj_timezone
|
||
from datetime import timedelta
|
||
from apps.hiring.models import HiringRequest
|
||
|
||
today = dj_timezone.localdate()
|
||
can_manage = user_can_manage_hr_portal(request.user)
|
||
|
||
from apps.hiring.recruiting import recruiting_context
|
||
recruiting = recruiting_context(request)
|
||
can_view_recruiting = recruiting['can_view_recruiting']
|
||
|
||
employees = Employee.objects.filter(is_active=True).select_related(
|
||
'post', 'department', 'user_pharmacy',
|
||
).order_by('last_name', 'first_name')
|
||
|
||
hr_posts = list(
|
||
Directory.objects.filter(directory_type__code='post', is_active=True)
|
||
.prefetch_related('location_types').order_by('name')
|
||
)
|
||
hr_post_location_types = {
|
||
p.id: [lt.code for lt in p.location_types.all()] for p in hr_posts
|
||
}
|
||
|
||
tab = (request.GET.get('tab') or '').strip()
|
||
allowed_tabs = {'employees', 'requests'}
|
||
if can_manage:
|
||
allowed_tabs.update({'probation', 'trainings', 'vacancies', 'hiring'})
|
||
if can_view_recruiting:
|
||
allowed_tabs.add('vacancies')
|
||
hr_active_tab = tab if tab in allowed_tabs else 'employees'
|
||
|
||
context = {
|
||
'employees': employees,
|
||
'employees_total': employees.count(),
|
||
'hr_department_tree': _hr_department_tree(),
|
||
'hr_posts': hr_posts,
|
||
'hr_post_location_types': hr_post_location_types,
|
||
'today': today,
|
||
'can_manage_hiring': can_manage,
|
||
'hr_active_tab': hr_active_tab,
|
||
'employees_new': 0,
|
||
'employees_on_probation': [],
|
||
'employees_probation': 0,
|
||
'intern_slots_total': 0,
|
||
'intern_slots_free': 0,
|
||
'hiring_requests': paginate(request, HiringRequest.objects.none()),
|
||
'trial_hiring_requests': [],
|
||
'hr_access_systems': [],
|
||
'hr_access_location_types': {},
|
||
'hr_pharmacies': [],
|
||
'hr_locations': [],
|
||
'hr_trial_days_choices': range(1, 6),
|
||
}
|
||
context.update(_hr_requests_context(request, can_manage))
|
||
context.update(recruiting)
|
||
|
||
if not can_manage:
|
||
return context
|
||
|
||
from apps.locations.models.pharmacies import Pharmacy
|
||
from apps.locations.models import Location
|
||
|
||
month_ago = today - timedelta(days=30)
|
||
employees_on_probation = [e for e in employees if e.is_on_probation]
|
||
trial_hiring_count = HiringRequest.objects.filter(status='trial').count()
|
||
intern_total = 0
|
||
intern_free = 0
|
||
try:
|
||
from apps.hiring.models import InternAccount
|
||
intern_qs = InternAccount.objects.filter(is_active=True)
|
||
intern_total = intern_qs.count()
|
||
intern_free = intern_qs.filter(assigned_to__isnull=True).count()
|
||
except Exception:
|
||
pass
|
||
|
||
context.update({
|
||
'employees_new': employees.filter(hire_date__gte=month_ago).count(),
|
||
'employees_on_probation': employees_on_probation,
|
||
'employees_probation': trial_hiring_count + len(employees_on_probation),
|
||
'intern_slots_total': intern_total,
|
||
'intern_slots_free': intern_free,
|
||
})
|
||
|
||
hiring_requests_qs = HiringRequest.objects.select_related(
|
||
'employee', 'post', 'department', 'user_pharmacy', 'user_location',
|
||
'requested_by', 'decision_by', 'granted_ticket', 'intern_account',
|
||
).order_by('-created_at')
|
||
trial_hiring_qs = hiring_requests_qs.filter(status='trial')
|
||
hr_access_systems = list(
|
||
Directory.objects.filter(
|
||
directory_type__code='access_system', is_active=True,
|
||
).prefetch_related('location_types').order_by('name')
|
||
)
|
||
hiring_requests_page = paginate(
|
||
request, hiring_requests_qs, page_param='hiring_page', size_param='hiring_per_page',
|
||
)
|
||
from apps.employees.services.password_invite import (
|
||
build_invite_absolute_url,
|
||
get_active_invite_for_employee,
|
||
)
|
||
for hr in hiring_requests_page.object_list:
|
||
invite = None
|
||
if hr.employee_id:
|
||
invite = get_active_invite_for_employee(hr.employee)
|
||
hr.password_invite_url = (
|
||
build_invite_absolute_url(request, invite) if invite else None
|
||
)
|
||
|
||
context.update({
|
||
'hiring_requests': hiring_requests_page,
|
||
'trial_hiring_requests': list(trial_hiring_qs),
|
||
'hr_access_systems': hr_access_systems,
|
||
'hr_access_location_types': {
|
||
a.id: [lt.code for lt in a.location_types.all()] for a in hr_access_systems
|
||
},
|
||
'hr_pharmacies': accessible_pharmacies(request.user).filter(is_active=True).order_by('pharmacy_number'),
|
||
'hr_locations': Location.objects.filter(
|
||
location_type__in=['office', 'warehouse'], is_active=True,
|
||
).order_by('location_type', 'address'),
|
||
})
|
||
|
||
return context
|
||
|
||
|
||
@login_required
|
||
def hr_portal_view(request):
|
||
return render_page(request, 'hr-portal.html', _hr_portal_context(request))
|
||
|
||
|
||
@login_required
|
||
@hr_manage_required
|
||
def hiring_create_view(request):
|
||
hiring_form_errors = None
|
||
hiring_just_created = None
|
||
if request.method == 'POST':
|
||
from datetime import datetime
|
||
from apps.hiring.services import create_hiring_request
|
||
|
||
last_name = request.POST.get('last_name', '').strip()
|
||
first_name = request.POST.get('first_name', '').strip()
|
||
middle_name = request.POST.get('middle_name', '').strip()
|
||
location_type = request.POST.get('user_location_type', '')
|
||
hire_date_raw = request.POST.get('hire_date', '')
|
||
trial_days_raw = request.POST.get('trial_days', '3')
|
||
|
||
if last_name and first_name and location_type:
|
||
try:
|
||
hire_date = datetime.strptime(hire_date_raw, '%Y-%m-%d').date()
|
||
except ValueError:
|
||
from django.utils import timezone as dj_timezone
|
||
hire_date = dj_timezone.localdate()
|
||
try:
|
||
trial_days = max(1, min(5, int(trial_days_raw)))
|
||
except ValueError:
|
||
trial_days = 3
|
||
|
||
department_id = request.POST.get('department_id') or None
|
||
pharmacy_id = request.POST.get('pharmacy_id') or None
|
||
location_id = request.POST.get('location_id') or None
|
||
post_id = request.POST.get('post_id') or None
|
||
|
||
try:
|
||
hiring_just_created = create_hiring_request(
|
||
requester=request.user,
|
||
last_name=last_name,
|
||
first_name=first_name,
|
||
middle_name=middle_name,
|
||
candidate_email=request.POST.get('candidate_email', '').strip(),
|
||
candidate_phone=request.POST.get('candidate_phone', '').strip(),
|
||
inn_code=request.POST.get('inn_code', '').strip(),
|
||
user_location_type=location_type,
|
||
department_id=department_id,
|
||
user_pharmacy_id=pharmacy_id,
|
||
user_location_id=location_id,
|
||
post_id=post_id,
|
||
hire_date=hire_date,
|
||
trial_days=trial_days,
|
||
employment_type=request.POST.get('employment_type', '').strip(),
|
||
requested_access_ids=request.POST.getlist('requested_access'),
|
||
access_note=request.POST.get('access_note', '').strip(),
|
||
)
|
||
except ValidationError as exc:
|
||
hiring_form_errors = exc.message_dict if hasattr(exc, 'message_dict') else {
|
||
'__all__': list(exc.messages),
|
||
}
|
||
|
||
context = _hr_portal_context(request)
|
||
if hiring_form_errors:
|
||
context['hr_active_tab'] = 'probation'
|
||
context['hiring_form_errors'] = hiring_form_errors
|
||
context['hiring_form_open'] = True
|
||
elif hiring_just_created is not None:
|
||
# После успешного создания показываем вкладку «На испытании»
|
||
# со свежим списком и баннером с данными интерна.
|
||
from apps.hiring.models import HiringRequest
|
||
context['hiring_just_created'] = (
|
||
HiringRequest.objects
|
||
.select_related('intern_account', 'granted_ticket')
|
||
.get(pk=hiring_just_created.pk)
|
||
)
|
||
context['hr_active_tab'] = 'probation'
|
||
else:
|
||
context['hr_active_tab'] = 'probation'
|
||
context['hiring_form_open'] = True
|
||
return render_page(request, 'hr-portal.html', context)
|
||
|
||
|
||
@login_required
|
||
@hr_manage_required
|
||
def hiring_approve_view(request, pk):
|
||
from apps.hiring.models import HiringRequest
|
||
from apps.hiring.services import approve_hiring_request
|
||
|
||
if request.method == 'POST':
|
||
hiring_request = get_object_or_404(HiringRequest, pk=pk)
|
||
approve_hiring_request(hiring_request, decided_by=request.user, note=request.POST.get('note', '').strip())
|
||
|
||
context = _hr_portal_context(request)
|
||
context['hr_active_tab'] = 'probation'
|
||
return render_page(request, 'hr-portal.html', context)
|
||
|
||
|
||
@login_required
|
||
@hr_manage_required
|
||
def hiring_reject_view(request, pk):
|
||
from apps.hiring.models import HiringRequest
|
||
from apps.hiring.services import reject_hiring_request
|
||
|
||
if request.method == 'POST':
|
||
hiring_request = get_object_or_404(HiringRequest, pk=pk)
|
||
reject_hiring_request(hiring_request, decided_by=request.user, note=request.POST.get('note', '').strip())
|
||
|
||
context = _hr_portal_context(request)
|
||
context['hr_active_tab'] = 'probation'
|
||
return render_page(request, 'hr-portal.html', context)
|
||
|
||
|
||
@login_required
|
||
@hr_manage_required
|
||
@require_POST
|
||
def hiring_regenerate_invite_view(request, pk):
|
||
from apps.hiring.models import HiringRequest
|
||
from apps.employees.services.password_invite import create_password_invite
|
||
|
||
hiring_request = get_object_or_404(HiringRequest, pk=pk, status='approved')
|
||
if not hiring_request.employee_id:
|
||
return HttpResponseForbidden()
|
||
create_password_invite(hiring_request.employee, created_by=request.user)
|
||
context = _hr_portal_context(request)
|
||
context['hr_active_tab'] = 'probation'
|
||
return render_page(request, 'hr-portal.html', context)
|
||
|
||
|
||
@login_required
|
||
@hr_manage_required
|
||
def hiring_credentials_pdf_view(request, pk):
|
||
from apps.hiring.models import HiringRequest
|
||
from apps.hiring.services import build_hiring_credentials_pdf
|
||
|
||
hiring_request = get_object_or_404(HiringRequest, pk=pk)
|
||
content = build_hiring_credentials_pdf(hiring_request, request=request)
|
||
if content is None:
|
||
return HttpResponseForbidden()
|
||
filename = f'credentials-{hiring_request.employee.username}.pdf'
|
||
response = HttpResponse(content, content_type='application/pdf')
|
||
response['Content-Disposition'] = f'attachment; filename="{filename}"'
|
||
return response
|
||
|
||
|
||
@login_required
|
||
@section_required('databases')
|
||
def databases_view(request):
|
||
from core.directory_portal import databases_view as _databases_view
|
||
return _databases_view(request)
|
||
|
||
|
||
@login_required
|
||
def it_support_view(request):
|
||
"""Підтримка IT: створення заявки та мої заявки"""
|
||
from apps.it_manager.models import Ticket
|
||
from apps.it_manager.views import _directory_choices, _resolve_directory
|
||
|
||
ticket_created = None
|
||
if request.method == 'POST':
|
||
title = request.POST.get('title', '').strip()
|
||
description = request.POST.get('description', '').strip()
|
||
ticket_type_id = _resolve_directory('ticket_type', request.POST.get('ticket_type', '')) \
|
||
or _resolve_directory('ticket_type', 'other')
|
||
priority_id = _resolve_directory('ticket_priority', request.POST.get('priority', '')) \
|
||
or _resolve_directory('ticket_priority', 'medium')
|
||
if title and description:
|
||
ticket_created = Ticket.objects.create(
|
||
title=title,
|
||
description=description,
|
||
ticket_type_id=ticket_type_id,
|
||
priority_id=priority_id,
|
||
requester=request.user,
|
||
)
|
||
|
||
my_tickets_qs = Ticket.objects.filter(requester=request.user).select_related(
|
||
'ticket_type', 'priority', 'status',
|
||
).order_by('-created_at')
|
||
my_tickets_page = paginate(request, my_tickets_qs, default_size=10, page_param='tickets_page', size_param='tickets_per_page')
|
||
context = {
|
||
'ticket_types': _directory_choices('ticket_type'),
|
||
'ticket_priorities': _directory_choices('ticket_priority'),
|
||
'my_tickets': my_tickets_page,
|
||
'my_open_count': Ticket.objects.filter(
|
||
requester=request.user, status__code__in=['new', 'in_progress', 'waiting'],
|
||
).count(),
|
||
'ticket_created': ticket_created,
|
||
}
|
||
return render_page(request, 'it-support.html', context)
|
||
|
||
|
||
# Довідники, у яких дозволено "на льоту" додавати нові записи прямо з форм.
|
||
# None — може будь-який залогінений (форма відкрита всім, напр. /it-support/);
|
||
# 'it' — лише IT-група (форма доступна тільки на /it-manager/).
|
||
DIRECTORY_QUICK_ADD_TYPES = {
|
||
'ticket_type': None,
|
||
'ticket_priority': None,
|
||
'equipment_type': 'it',
|
||
'equipment_brand': 'it',
|
||
'worklog_type': 'it',
|
||
'consumable_type': 'it',
|
||
'consumable_color': 'it',
|
||
'contract_service_type': 'it',
|
||
'contract_connection_type': 'it',
|
||
'servicecall_channel': 'it',
|
||
}
|
||
|
||
|
||
@login_required
|
||
def directory_quick_add_view(request):
|
||
"""Швидке додавання нового запису в довідник прямо з форми (select), без переходу в admin."""
|
||
if request.method != 'POST':
|
||
return HttpResponseNotAllowed(['POST'])
|
||
|
||
from apps.it_manager.views import _is_it_staff
|
||
from core.models.directories import DirectoryType
|
||
|
||
type_code = request.POST.get('type_code', '')
|
||
name = request.POST.get('name', '').strip()
|
||
|
||
if type_code not in DIRECTORY_QUICK_ADD_TYPES or not name:
|
||
return JsonResponse({'error': 'Некоректні дані'}, status=400)
|
||
|
||
required_role = DIRECTORY_QUICK_ADD_TYPES[type_code]
|
||
if required_role == 'it' and not _is_it_staff(request.user):
|
||
return HttpResponseForbidden()
|
||
|
||
directory_type = DirectoryType.objects.filter(code=type_code).first()
|
||
if not directory_type:
|
||
return JsonResponse({'error': 'Довідник не знайдено'}, status=400)
|
||
|
||
directory, _ = Directory.objects.get_or_create(
|
||
directory_type=directory_type, name=name, defaults={'is_active': True},
|
||
)
|
||
return JsonResponse({'id': directory.id, 'code': directory.code, 'name': directory.name})
|
||
|
||
|
||
LOGIN_MAX_ATTEMPTS = 5 # неудачных попыток
|
||
LOGIN_BLOCK_SECONDS = 5 * 60 # окно блокировки
|
||
TWOFA_MAX_ATTEMPTS = 5
|
||
TWOFA_BLOCK_SECONDS = 5 * 60
|
||
|
||
|
||
def _login_throttle_keys(request, username: str | None = None):
|
||
"""Per-username primary; IP secondary (shared behind Traefik is OK as soft limit)."""
|
||
keys = []
|
||
uname = (username or request.POST.get('username') or '').strip().lower()
|
||
if uname:
|
||
keys.append(f'login_attempts:user:{uname}')
|
||
ip = request.META.get('REMOTE_ADDR', '') or 'unknown'
|
||
keys.append(f'login_attempts:ip:{ip}')
|
||
return keys
|
||
|
||
|
||
def _throttle_blocked(keys) -> bool:
|
||
return any(cache.get(k, 0) >= LOGIN_MAX_ATTEMPTS for k in keys)
|
||
|
||
|
||
def _throttle_hit(keys) -> None:
|
||
for k in keys:
|
||
attempts = cache.get(k, 0)
|
||
cache.set(k, attempts + 1, LOGIN_BLOCK_SECONDS)
|
||
|
||
|
||
def _throttle_clear(keys) -> None:
|
||
for k in keys:
|
||
cache.delete(k)
|
||
|
||
|
||
def login_view(request):
|
||
if request.method == 'POST':
|
||
username = request.POST.get('username')
|
||
keys = _login_throttle_keys(request, username)
|
||
if _throttle_blocked(keys):
|
||
return render(request, 'login.html', {
|
||
'login_error': 'Слишком много попыток входа. Повторите позже.',
|
||
}, status=429)
|
||
|
||
password = request.POST.get('password')
|
||
user = authenticate(request, username=username, password=password)
|
||
if user is not None:
|
||
_throttle_clear(keys)
|
||
login(request, user)
|
||
from apps.mail_connector.login_hooks import (
|
||
on_portal_auth_complete,
|
||
on_portal_login,
|
||
should_defer_mail_activation,
|
||
)
|
||
on_portal_login(request, user, password)
|
||
if not should_defer_mail_activation(user):
|
||
on_portal_auth_complete(request, user)
|
||
next_url = request.GET.get('next') or request.POST.get('next') or '/'
|
||
if not url_has_allowed_host_and_scheme(next_url, allowed_hosts=settings.ALLOWED_HOSTS):
|
||
next_url = '/'
|
||
return redirect(next_url)
|
||
|
||
_throttle_hit(keys)
|
||
return render(request, 'login.html', {
|
||
'login_error': 'Неверный логин или пароль',
|
||
'setup_success': request.GET.get('setup') == '1',
|
||
'setup_username': request.GET.get('user', ''),
|
||
})
|
||
return render(request, 'login.html', {
|
||
'setup_success': request.GET.get('setup') == '1',
|
||
'setup_username': request.GET.get('user', ''),
|
||
})
|
||
|
||
|
||
@require_POST
|
||
def logout_view(request):
|
||
logout(request)
|
||
return redirect('login')
|
||
|
||
|
||
# ── Двухфакторная аутентификация (TOTP) ──────────────────────────────────
|
||
|
||
@login_required
|
||
def twofa_setup_view(request):
|
||
"""Настройка 2FA: QR + ручной ключ, подтверждение кодом → включение."""
|
||
from apps.employees.services import twofa
|
||
from core.security import log_security_event
|
||
|
||
user = request.user
|
||
return_to = (request.GET.get('return_to') or request.POST.get('return_to') or '').strip()
|
||
profile_flow = return_to == 'profile'
|
||
|
||
if user.totp_enabled:
|
||
if profile_flow:
|
||
return _profile_settings_response(request, twofa_message='2FA уже включена.')
|
||
return redirect('/')
|
||
|
||
reset_notice = ''
|
||
if request.GET.get('reset') == '1':
|
||
reset_notice = (
|
||
'Сохранённый ключ 2FA повреждён или не расшифровывается '
|
||
'(например, после смены FIELD_ENCRYPTION_KEY). Настройте 2FA заново.'
|
||
)
|
||
|
||
secret = request.session.get('2fa_setup_secret')
|
||
if not secret:
|
||
secret = twofa.generate_secret()
|
||
request.session['2fa_setup_secret'] = secret
|
||
|
||
error = ''
|
||
if request.method == 'POST':
|
||
if twofa.verify_totp(secret, request.POST.get('code', '')):
|
||
codes, hashes = twofa.generate_backup_codes()
|
||
user.totp_secret = secret
|
||
user.totp_enabled = True
|
||
user.totp_backup_codes = hashes
|
||
user.save(update_fields=['totp_secret', 'totp_enabled', 'totp_backup_codes'])
|
||
request.session.pop('2fa_setup_secret', None)
|
||
request.session['2fa_verified'] = True
|
||
from apps.mail_connector.login_hooks import on_portal_auth_complete
|
||
on_portal_auth_complete(request, user)
|
||
log_security_event('twofa_enabled', actor=user, target=user.username)
|
||
if profile_flow:
|
||
return _profile_settings_response(
|
||
request,
|
||
twofa_backup_codes=codes,
|
||
twofa_message='2FA успешно включена. Сохраните резервные коды.',
|
||
)
|
||
request.session['2fa_backup_codes'] = codes
|
||
return redirect('twofa_backup_codes')
|
||
error = 'Неверный код. Проверьте время на устройстве и попробуйте снова.'
|
||
|
||
uri = twofa.provisioning_uri(user, secret)
|
||
setup_ctx = {
|
||
'twofa_setup_secret': secret,
|
||
'twofa_setup_qr': twofa.qr_data_uri(uri),
|
||
'twofa_setup_error': error,
|
||
'twofa_reset_notice': reset_notice,
|
||
'twofa_show_setup': True,
|
||
}
|
||
if profile_flow:
|
||
return _profile_settings_response(request, **setup_ctx)
|
||
|
||
return render(request, '2fa-setup.html', {
|
||
'secret': secret, 'qr': twofa.qr_data_uri(uri), 'error': error,
|
||
'reset_notice': reset_notice,
|
||
})
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def twofa_disable_view(request):
|
||
"""Отключение добровольной 2FA (не для staff/IT)."""
|
||
from apps.employees.services import twofa
|
||
from core.security import log_security_event, user_can_disable_2fa
|
||
|
||
user = request.user
|
||
if not user_can_disable_2fa(user):
|
||
return _profile_settings_response(
|
||
request, twofa_error='Отключение 2FA недоступно для вашего аккаунта.',
|
||
)
|
||
if not twofa.verify_totp(user.totp_secret, request.POST.get('code', '')):
|
||
return _profile_settings_response(request, twofa_error='Неверный код из приложения.')
|
||
twofa.disable(user)
|
||
request.session.pop('2fa_verified', None)
|
||
log_security_event('twofa_disabled', actor=user, target=user.username)
|
||
return _profile_settings_response(request, twofa_message='2FA отключена.')
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def twofa_regenerate_backup_view(request):
|
||
"""Новые резервные коды (старые перестают работать)."""
|
||
from apps.employees.services import twofa
|
||
from core.security import log_security_event
|
||
|
||
user = request.user
|
||
if not user.totp_enabled:
|
||
return _profile_settings_response(request, twofa_error='2FA не включена.')
|
||
if not twofa.verify_totp(user.totp_secret, request.POST.get('code', '')):
|
||
return _profile_settings_response(request, twofa_error='Неверный код из приложения.')
|
||
codes, hashes = twofa.generate_backup_codes()
|
||
user.totp_backup_codes = hashes
|
||
user.save(update_fields=['totp_backup_codes'])
|
||
log_security_event('twofa_backup_regenerated', actor=user, target=user.username)
|
||
return _profile_settings_response(
|
||
request,
|
||
twofa_backup_codes=codes,
|
||
twofa_message='Новые резервные коды сгенерированы. Сохраните их — старые больше не работают.',
|
||
)
|
||
|
||
|
||
@login_required
|
||
def twofa_backup_codes_view(request):
|
||
"""Показ резервных кодов один раз (после включения 2FA)."""
|
||
codes = request.session.pop('2fa_backup_codes', None)
|
||
if not codes:
|
||
return redirect('/')
|
||
return render(request, '2fa-backup-codes.html', {'codes': codes})
|
||
|
||
|
||
def twofa_verify_view(request):
|
||
"""Ввод TOTP/резервного кода после входа (сессия ещё не 2fa_verified)."""
|
||
from apps.employees.services import twofa
|
||
user = request.user
|
||
if not user.is_authenticated:
|
||
return redirect('login')
|
||
if not user.totp_enabled:
|
||
return redirect('twofa_setup')
|
||
if twofa.reset_invalid_2fa(user):
|
||
return redirect(f"{reverse('twofa_setup')}?reset=1")
|
||
|
||
next_url = request.GET.get('next') or request.POST.get('next') or '/'
|
||
if not url_has_allowed_host_and_scheme(next_url, allowed_hosts=settings.ALLOWED_HOSTS):
|
||
next_url = '/'
|
||
if request.session.get('2fa_verified'):
|
||
return redirect(next_url)
|
||
|
||
throttle_key = f'2fa_attempts:{user.pk}'
|
||
error = ''
|
||
if request.method == 'POST':
|
||
attempts = cache.get(throttle_key, 0)
|
||
if attempts >= TWOFA_MAX_ATTEMPTS:
|
||
return render(
|
||
request,
|
||
'2fa-verify.html',
|
||
{
|
||
'error': 'Слишком много попыток. Повторите позже.',
|
||
'next': next_url,
|
||
},
|
||
status=429,
|
||
)
|
||
code = request.POST.get('code', '')
|
||
if twofa.verify_totp(user.totp_secret, code) or twofa.verify_and_consume_backup(user, code):
|
||
cache.delete(throttle_key)
|
||
request.session['2fa_verified'] = True
|
||
from apps.mail_connector.login_hooks import on_portal_auth_complete
|
||
on_portal_auth_complete(request, user)
|
||
return redirect(next_url)
|
||
cache.set(throttle_key, attempts + 1, TWOFA_BLOCK_SECONDS)
|
||
error = 'Неверный код.'
|
||
return render(request, '2fa-verify.html', {'error': error, 'next': next_url})
|
||
|
||
|
||
@login_required
|
||
@hr_manage_required
|
||
def modal_add_employee_view(request):
|
||
return render(request, 'partials/modal-add-employee.html')
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def hr_request_create_view(request):
|
||
from django.db import DatabaseError
|
||
from apps.employees.models import Employee
|
||
from apps.employees.services.hr_requests import create_hr_request
|
||
|
||
form_data = _hr_request_form_from_post(request.POST)
|
||
request_type = form_data['type']
|
||
date_from = _parse_optional_date(form_data['date_from'])
|
||
date_to = _parse_optional_date(form_data['date_to'])
|
||
note = form_data['note']
|
||
substitute = None
|
||
sid = form_data.get('substitute_id') or ''
|
||
if sid.isdigit():
|
||
substitute = Employee.objects.filter(pk=int(sid), is_active=True).first()
|
||
|
||
try:
|
||
created = create_hr_request(
|
||
employee=request.user,
|
||
request_type=request_type,
|
||
date_from=date_from,
|
||
date_to=date_to,
|
||
note=note,
|
||
substitute=substitute,
|
||
)
|
||
return _hr_request_response(request, hr_request_created=created)
|
||
except ValidationError as exc:
|
||
return _hr_request_response(
|
||
request,
|
||
hr_request_errors=_validation_error_messages(exc),
|
||
form_open=True,
|
||
form_data=form_data,
|
||
)
|
||
except DatabaseError:
|
||
return _hr_request_response(
|
||
request,
|
||
hr_request_errors=[
|
||
'Не удалось сохранить заявку. Обратитесь к администратору или выполните migrate.',
|
||
],
|
||
form_open=True,
|
||
form_data=form_data,
|
||
)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def hr_request_cancel_view(request, pk):
|
||
from apps.employees.models import EmployeeHrRequest
|
||
from apps.employees.services.hr_requests import cancel_hr_request
|
||
|
||
req = get_object_or_404(EmployeeHrRequest, pk=pk)
|
||
try:
|
||
cancel_hr_request(req, actor=request.user)
|
||
except ValidationError as exc:
|
||
errors = list(exc.messages)
|
||
return _hr_request_response(request, hr_request_errors=errors)
|
||
return _hr_request_response(request)
|
||
|
||
|
||
@login_required
|
||
@hr_manage_required
|
||
@require_POST
|
||
def hr_request_approve_view(request, pk):
|
||
from apps.employees.models import EmployeeHrRequest
|
||
from apps.employees.services.hr_requests import decide_hr_request
|
||
|
||
req = get_object_or_404(EmployeeHrRequest, pk=pk)
|
||
if req.request_type == 'substitution':
|
||
return _hr_request_portal_response(
|
||
request,
|
||
hr_request_errors=['Подмену подтверждает выбранный сотрудник, не HR.'],
|
||
)
|
||
try:
|
||
decide_hr_request(req, decided_by=request.user, approve=True)
|
||
except ValidationError as exc:
|
||
return _hr_request_portal_response(request, hr_request_errors=list(exc.messages))
|
||
return _hr_request_portal_response(request)
|
||
|
||
|
||
@login_required
|
||
@hr_manage_required
|
||
@require_POST
|
||
def hr_request_reject_view(request, pk):
|
||
from apps.employees.models import EmployeeHrRequest
|
||
from apps.employees.services.hr_requests import decide_hr_request
|
||
|
||
req = get_object_or_404(EmployeeHrRequest, pk=pk)
|
||
if req.request_type == 'substitution':
|
||
return _hr_request_portal_response(
|
||
request,
|
||
hr_request_errors=['Подмену отклоняет выбранный сотрудник.'],
|
||
)
|
||
try:
|
||
decide_hr_request(
|
||
req,
|
||
decided_by=request.user,
|
||
approve=False,
|
||
decision_note=request.POST.get('decision_note', ''),
|
||
)
|
||
except ValidationError as exc:
|
||
return _hr_request_portal_response(request, hr_request_errors=list(exc.messages))
|
||
return _hr_request_portal_response(request)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def hr_substitution_confirm_view(request, pk):
|
||
from apps.employees.models import EmployeeHrRequest
|
||
from apps.employees.services.hr_requests import confirm_substitution
|
||
|
||
req = get_object_or_404(EmployeeHrRequest, pk=pk)
|
||
try:
|
||
confirm_substitution(req, actor=request.user)
|
||
except ValidationError as exc:
|
||
return _hr_request_response(request, hr_request_errors=list(exc.messages))
|
||
return _hr_request_response(request)
|
||
|
||
|
||
@login_required
|
||
@require_POST
|
||
def hr_substitution_decline_view(request, pk):
|
||
from apps.employees.models import EmployeeHrRequest
|
||
from apps.employees.services.hr_requests import decline_substitution
|
||
|
||
req = get_object_or_404(EmployeeHrRequest, pk=pk)
|
||
try:
|
||
decline_substitution(
|
||
req,
|
||
actor=request.user,
|
||
decision_note=request.POST.get('decision_note', ''),
|
||
)
|
||
except ValidationError as exc:
|
||
return _hr_request_response(request, hr_request_errors=list(exc.messages))
|
||
return _hr_request_response(request)
|