lokses
4e803c53e0
- Модель Employee: totp_secret (шифруется), totp_enabled, totp_backup_codes - Сервис twofa: генерация секрета/QR, проверка TOTP, резервные коды (sha256) - TwoFactorMiddleware: enforcement для is_staff на портале и /admin/ - Страницы setup/verify/backup-codes; сброс 2fa_verified при каждом входе - Админка: колонка статуса 2FA, фильтр, действие «Сбросить 2FA» - Тесты: enforcement, setup, verify, резервные коды, disable Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
971 lines
39 KiB
Python
971 lines
39 KiB
Python
from django.core.exceptions import ValidationError
|
||
from django.db.models import Q, Count, Prefetch
|
||
from django.http import HttpResponseForbidden, JsonResponse, HttpResponseNotAllowed
|
||
from django.shortcuts import render, redirect, get_object_or_404
|
||
from django.contrib.auth import authenticate, login, logout
|
||
from django.contrib.auth.decorators import login_required
|
||
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.utils.htmx import render_page
|
||
from core.utils.pagination import paginate
|
||
from django.utils.translation import gettext_lazy as _
|
||
|
||
# Create your views here.
|
||
|
||
@login_required
|
||
def index_view(request):
|
||
return render(request, 'index.html')
|
||
|
||
@login_required
|
||
def dashboard_view(request):
|
||
from django.utils import timezone as dj_timezone
|
||
from core.middleware import ONLINE_WINDOW
|
||
from core.models.directories import DirectoryType
|
||
from apps.feed.models import Post
|
||
from apps.it_manager.models import Ticket
|
||
|
||
employees_active = Employee.objects.filter(is_active=True)
|
||
|
||
context = {
|
||
'today': dj_timezone.localdate(),
|
||
'latest_posts': Post.objects.filter(is_published=True).select_related('author')[:3],
|
||
'employees_online': employees_active.filter(last_seen__gte=dj_timezone.now() - ONLINE_WINDOW).count(),
|
||
'employees_active': employees_active.count(),
|
||
'pharmacies_active': Pharmacy.objects.filter(is_active=True).count(),
|
||
'open_tickets': Ticket.objects.filter(status__code__in=['new', 'in_progress', 'waiting']).count(),
|
||
'directory_types_count': DirectoryType.objects.filter(is_active=True).count(),
|
||
'directories_count': Directory.objects.count(),
|
||
}
|
||
return render_page(request, 'dashboard.html', context)
|
||
|
||
@login_required
|
||
def employees_view(request):
|
||
location_type = request.GET.get('location_type', '')
|
||
organization_id = request.GET.get('organization_id', '')
|
||
pharmacy_id = request.GET.get('pharmacy_id', '')
|
||
|
||
employees = Employee.objects.select_related(
|
||
'department', 'post', 'user_pharmacy', 'user_organization',
|
||
'user_pharmacy__organization',
|
||
).order_by('last_name', 'first_name')
|
||
|
||
if location_type == 'office':
|
||
employees = employees.filter(user_location_type='office')
|
||
elif location_type == 'pharmacy':
|
||
employees = employees.filter(user_location_type='pharmacy')
|
||
if organization_id:
|
||
employees = employees.filter(user_pharmacy__organization_id=organization_id)
|
||
if pharmacy_id:
|
||
employees = employees.filter(user_pharmacy_id=pharmacy_id)
|
||
|
||
active_count = employees.filter(is_active=True).count()
|
||
total_count = employees.count()
|
||
page = paginate(request, employees)
|
||
|
||
organizations = Directory.objects.filter(
|
||
directory_type__code='organization', is_active=True
|
||
).order_by('name')
|
||
|
||
pharmacies_qs = (
|
||
Pharmacy.objects
|
||
.select_related('organization')
|
||
.filter(is_active=True)
|
||
.order_by('organization__name', 'pharmacy_number')
|
||
)
|
||
|
||
pharmacies_by_org = {}
|
||
for ph in pharmacies_qs:
|
||
key = ph.organization_id
|
||
if key not in pharmacies_by_org:
|
||
pharmacies_by_org[key] = []
|
||
pharmacies_by_org[key].append({
|
||
'id': ph.id,
|
||
'label': f"Аптека №{ph.pharmacy_number} — {ph.address}",
|
||
})
|
||
|
||
current_pharmacies = (
|
||
pharmacies_qs.filter(organization_id=organization_id)
|
||
if organization_id else pharmacies_qs
|
||
)
|
||
|
||
context = {
|
||
'employees': page,
|
||
'page': page,
|
||
'active_count': active_count,
|
||
'total_count': total_count,
|
||
'organizations': organizations,
|
||
'current_pharmacies': current_pharmacies,
|
||
'pharmacies_by_org': pharmacies_by_org,
|
||
'filter_location_type': location_type,
|
||
'filter_organization_id': organization_id,
|
||
'filter_pharmacy_id': pharmacy_id,
|
||
}
|
||
return render_page(request, 'employees.html', context)
|
||
|
||
|
||
def _can_view_employee_pii(request, employee) -> bool:
|
||
"""Особисті дані (телефони, дата народження, ІПН, екстрений контакт,
|
||
статуси AD/Mattermost) бачить сам співробітник, його безпосередній
|
||
керівник або staff. Решта колег бачать лише робочу картку.
|
||
Межа узгоджена з employees/serializers.py (EmployeeSerializer/
|
||
EmployeeDetailSerializer) з коміту faa06b9."""
|
||
return (
|
||
request.user.is_staff
|
||
or request.user.pk == employee.pk
|
||
or employee.manager_id == request.user.pk
|
||
)
|
||
|
||
|
||
@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,
|
||
)
|
||
can_view_pii = _can_view_employee_pii(request, employee)
|
||
|
||
context = {
|
||
'employee': employee,
|
||
'can_view_pii': can_view_pii,
|
||
'can_edit': request.user.is_staff,
|
||
'subordinates': employee.subordinates.filter(is_active=True).select_related('post')
|
||
if employee.is_manager else [],
|
||
'manager_chain': employee.get_manager_chain(),
|
||
}
|
||
|
||
if can_view_pii:
|
||
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)
|
||
context['last_attendance'] = (
|
||
employee.attendance_logs.select_related('location').order_by('-timestamp').first()
|
||
)
|
||
|
||
return render_page(request, 'employee-detail.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
|
||
|
||
|
||
@login_required
|
||
def pharmacies_view(request):
|
||
pharmacies = Pharmacy.objects.select_related('organization', 'location_region', 'head_pharmacist').order_by('organization__name', 'pharmacy_number')
|
||
active_count = pharmacies.filter(is_active=True).count()
|
||
|
||
# Get unique regions for filter
|
||
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')
|
||
region_tree = build_region_tree(region_nodes)
|
||
|
||
context = {
|
||
'pharmacies': pharmacies,
|
||
'active_count': active_count,
|
||
'regions': regions,
|
||
'region_tree': region_tree,
|
||
}
|
||
return render_page(request, 'pharmacies.html', context)
|
||
|
||
|
||
@login_required
|
||
def dislocation_view(request):
|
||
"""Дислокація: аптеки на карті/у таблиці разом з їхнім персоналом (замінює вкладку «Аптеки» на /structure/)."""
|
||
pharmacies = (
|
||
Pharmacy.objects
|
||
.select_related('organization', 'location_region', 'head_pharmacist')
|
||
.prefetch_related(Prefetch(
|
||
'pharmacy_employees',
|
||
queryset=Employee.objects.filter(is_active=True).select_related('post'),
|
||
to_attr='active_staff',
|
||
))
|
||
.annotate(staff_count=Count('pharmacy_employees', filter=Q(pharmacy_employees__is_active=True)))
|
||
.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
|
||
|
||
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, [])
|
||
|
||
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),
|
||
}
|
||
return render_page(request, 'dislocation.html', context)
|
||
|
||
@login_required
|
||
def calendar_view(request):
|
||
return render_page(request, 'calendar.html')
|
||
|
||
@login_required
|
||
def timetrack_view(request):
|
||
"""Облік робочого часу: агрегація WorkShift (фактичні зміни з терміналів)
|
||
по співробітниках за обраний місяць. Дані реальні — приходи/уходи
|
||
матеріалізуються з AttendanceLog у WorkShift (apps.visit_manager)."""
|
||
import datetime
|
||
from django.utils import timezone
|
||
from django.db.models import Sum, Count, Q
|
||
from apps.visit_manager.models import WorkShift
|
||
|
||
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', '')
|
||
|
||
shifts = (WorkShift.objects
|
||
.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 = WorkShift.objects.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,
|
||
}
|
||
return render_page(request, 'timetrack.html', context)
|
||
|
||
@login_required
|
||
def reports_view(request):
|
||
return render_page(request, 'reports.html')
|
||
|
||
@login_required
|
||
def documents_view(request):
|
||
return render_page(request, 'documents.html')
|
||
|
||
@login_required
|
||
def recruiting_view(request):
|
||
return render_page(request, 'recruiting.html')
|
||
|
||
@login_required
|
||
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 render_page(request, 'settings.html')
|
||
|
||
|
||
# Поля, які співробітник редагує сам на /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):
|
||
"""Особиста сторінка співробітника: контактні дані «для себе» + налаштування інтерфейсу порталу."""
|
||
from core.utils.pagination import PAGE_SIZE_CHOICES
|
||
|
||
employee = request.user
|
||
saved = False
|
||
errors = None
|
||
|
||
if request.method == 'POST':
|
||
employee.phone = request.POST.get('phone', '').strip()
|
||
employee.personal_phone = request.POST.get('personal_phone', '').strip()
|
||
employee.personal_email = request.POST.get('personal_email', '').strip()
|
||
employee.emergency_contact_name = request.POST.get('emergency_contact_name', '').strip()
|
||
employee.emergency_contact_phone = request.POST.get('emergency_contact_phone', '').strip()
|
||
employee.emergency_contact_relation = request.POST.get('emergency_contact_relation', '').strip()
|
||
|
||
if request.FILES.get('photo'):
|
||
employee.photo = request.FILES['photo']
|
||
|
||
try:
|
||
page_size = int(request.POST.get('default_page_size', ''))
|
||
except (TypeError, ValueError):
|
||
page_size = None
|
||
if page_size in PAGE_SIZE_CHOICES:
|
||
employee.default_page_size = page_size
|
||
|
||
try:
|
||
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()
|
||
saved = True
|
||
except ValidationError as exc:
|
||
errors = exc.messages
|
||
|
||
context = {
|
||
'employee': employee,
|
||
'page_size_choices': PAGE_SIZE_CHOICES,
|
||
'saved': saved,
|
||
'errors': errors,
|
||
}
|
||
return render_page(request, 'profile.html', context)
|
||
|
||
|
||
def _build_chat_context(request):
|
||
from apps.mattermost_connector.models import MattermostUser
|
||
from apps.mattermost_connector.services.token_service import ensure_personal_access_token
|
||
|
||
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': {
|
||
'apiBase': f'{public_url}/api/v4',
|
||
'wsUrl': ws_url,
|
||
'token': token,
|
||
'currentUserId': request.user.mattermost_user_id,
|
||
},
|
||
'chat_colleagues': colleagues,
|
||
}
|
||
|
||
|
||
@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):
|
||
"""Фабрика простых страниц: HTMX-фрагмент или полный layout"""
|
||
@login_required
|
||
def view(request):
|
||
return render_page(request, template_name)
|
||
return view
|
||
|
||
|
||
email_view = _static_page_view('email.html')
|
||
purchases_view = _static_page_view('purchases.html')
|
||
help_view = _static_page_view('help.html')
|
||
|
||
|
||
@login_required
|
||
def pharmacy_detail_view(request, pk):
|
||
"""Аптека целиком: реквизиты, лицензия, адрес/карта, персонал,
|
||
IT-обладнання, інтернет-договори, показання лічильників."""
|
||
from apps.it_manager.models import Equipment, InternetContract
|
||
from apps.utilities.models import UtilityReading
|
||
|
||
pharmacy = get_object_or_404(
|
||
Pharmacy.objects.select_related('organization', 'location_region', 'head_pharmacist__post')
|
||
.prefetch_related('phones'),
|
||
pk=pk,
|
||
)
|
||
staff = list(
|
||
Employee.objects.filter(user_pharmacy=pharmacy, is_active=True)
|
||
.select_related('post').order_by('last_name', 'first_name')
|
||
)
|
||
equipment = list(
|
||
Equipment.objects.filter(location=pharmacy)
|
||
.select_related('equipment_model__equipment_type', 'equipment_model__brand', 'status', 'assigned_to')
|
||
.order_by('equipment_model__equipment_type__name', 'equipment_model__name')
|
||
)
|
||
internet_contracts = list(
|
||
InternetContract.objects.filter(location=pharmacy)
|
||
.select_related('provider', 'service_type', 'connection_type', 'status')
|
||
.order_by('-start_date')
|
||
)
|
||
utility_readings = list(
|
||
UtilityReading.objects.filter(pharmacy=pharmacy)
|
||
.select_related('utility_type', 'submitted_by')
|
||
.order_by('-reading_date')[:50]
|
||
)
|
||
context = {
|
||
'pharmacy': pharmacy,
|
||
'staff': staff,
|
||
'staff_count': len(staff),
|
||
'phones': list(pharmacy.phones.filter(is_active=True)),
|
||
'equipment': equipment,
|
||
'internet_contracts': internet_contracts,
|
||
'utility_readings': utility_readings,
|
||
'can_edit': request.user.is_staff,
|
||
}
|
||
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('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_portal_context(request):
|
||
from django.utils import timezone as dj_timezone
|
||
from datetime import timedelta
|
||
from apps.hiring.models import HiringRequest
|
||
from apps.locations.models.pharmacies import Pharmacy
|
||
|
||
today = dj_timezone.localdate()
|
||
month_ago = today - timedelta(days=30)
|
||
|
||
employees = Employee.objects.filter(is_active=True).select_related(
|
||
'post', 'department', 'user_pharmacy',
|
||
).order_by('last_name', 'first_name')
|
||
|
||
# Должности с их типами локаций: пустой список = должность доступна везде
|
||
# (та же семантика, что в Employee.clean()). Карта отдаётся в форму найма,
|
||
# чтобы фильтровать «Посаду» по выбранному типу локации.
|
||
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
|
||
}
|
||
|
||
context = {
|
||
'employees': employees,
|
||
'employees_total': employees.count(),
|
||
'employees_new': employees.filter(hire_date__gte=month_ago).count(),
|
||
'employees_probation': employees.filter(probation_end_date__gte=today).count(),
|
||
'hr_department_tree': _hr_department_tree(),
|
||
'hr_posts': hr_posts,
|
||
'hr_post_location_types': hr_post_location_types,
|
||
'today': today,
|
||
}
|
||
|
||
if request.user.is_staff:
|
||
hiring_requests_qs = HiringRequest.objects.select_related(
|
||
'employee', 'post', 'department', 'user_pharmacy', 'requested_by', 'decision_by',
|
||
).order_by('-created_at')
|
||
context.update({
|
||
'hiring_requests': paginate(
|
||
request, hiring_requests_qs, page_param='hiring_page', size_param='hiring_per_page',
|
||
),
|
||
'hr_access_systems': Directory.objects.filter(
|
||
directory_type__code='access_system', is_active=True,
|
||
).order_by('name'),
|
||
'hr_pharmacies': Pharmacy.objects.filter(is_active=True).order_by('pharmacy_number'),
|
||
'hr_trial_days_choices': range(1, 6),
|
||
'hr_departments_flat': Directory.objects.filter(
|
||
directory_type__code='department', is_active=True,
|
||
).order_by('name'),
|
||
})
|
||
|
||
return context
|
||
|
||
|
||
@login_required
|
||
def hr_portal_view(request):
|
||
return render_page(request, 'hr-portal.html', _hr_portal_context(request))
|
||
|
||
|
||
@login_required
|
||
def hiring_create_view(request):
|
||
if not request.user.is_staff:
|
||
return HttpResponseForbidden()
|
||
|
||
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
|
||
post_id = request.POST.get('post_id') or None
|
||
|
||
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(),
|
||
user_location_type=location_type,
|
||
department_id=department_id,
|
||
user_pharmacy_id=pharmacy_id,
|
||
post_id=post_id,
|
||
hire_date=hire_date,
|
||
trial_days=trial_days,
|
||
requested_access_ids=request.POST.getlist('requested_access'),
|
||
access_note=request.POST.get('access_note', '').strip(),
|
||
)
|
||
|
||
context = _hr_portal_context(request)
|
||
context['hr_active_tab'] = 'hiring'
|
||
return render_page(request, 'hr-portal.html', context)
|
||
|
||
|
||
@login_required
|
||
def hiring_approve_view(request, pk):
|
||
if not request.user.is_staff:
|
||
return HttpResponseForbidden()
|
||
|
||
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'] = 'hiring'
|
||
return render_page(request, 'hr-portal.html', context)
|
||
|
||
|
||
@login_required
|
||
def hiring_reject_view(request, pk):
|
||
if not request.user.is_staff:
|
||
return HttpResponseForbidden()
|
||
|
||
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'] = 'hiring'
|
||
return render_page(request, 'hr-portal.html', context)
|
||
|
||
|
||
@login_required
|
||
def databases_view(request):
|
||
"""Довідники: типи справочников с количеством записей"""
|
||
from django.db.models import Count, Max
|
||
from core.models.directories import DirectoryType
|
||
|
||
directory_types = list(
|
||
DirectoryType.objects
|
||
.annotate(records=Count('directory'), last_updated=Max('directory__updated_at'))
|
||
.order_by('name')
|
||
)
|
||
directories = Directory.objects.all()
|
||
|
||
context = {
|
||
'directory_types': directory_types,
|
||
'types_count': len(directory_types),
|
||
'records_count': directories.count(),
|
||
'active_records_count': directories.filter(is_active=True).count(),
|
||
'last_updated': max(
|
||
(dt.last_updated for dt in directory_types if dt.last_updated),
|
||
default=None,
|
||
),
|
||
}
|
||
return render_page(request, 'databases.html', context)
|
||
|
||
|
||
@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})
|
||
|
||
|
||
def login_view(request):
|
||
if request.method == 'POST':
|
||
username = request.POST.get('username')
|
||
password = request.POST.get('password')
|
||
user = authenticate(request, username=username, password=password)
|
||
if user is not None:
|
||
login(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)
|
||
return render(request, 'login.html', {'login_error': 'Неверный логин или пароль'})
|
||
return render(request, 'login.html')
|
||
|
||
|
||
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
|
||
user = request.user
|
||
|
||
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
|
||
request.session['2fa_backup_codes'] = codes # показываем один раз
|
||
return redirect('twofa_backup_codes')
|
||
error = 'Неверный код. Проверьте время на устройстве и попробуйте снова.'
|
||
|
||
uri = twofa.provisioning_uri(user, secret)
|
||
return render(request, '2fa-setup.html', {
|
||
'secret': secret, 'qr': twofa.qr_data_uri(uri), 'error': error,
|
||
})
|
||
|
||
|
||
@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')
|
||
|
||
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)
|
||
|
||
error = ''
|
||
if request.method == 'POST':
|
||
code = request.POST.get('code', '')
|
||
if twofa.verify_totp(user.totp_secret, code) or twofa.verify_and_consume_backup(user, code):
|
||
request.session['2fa_verified'] = True
|
||
return redirect(next_url)
|
||
error = 'Неверный код.'
|
||
return render(request, '2fa-verify.html', {'error': error, 'next': next_url})
|
||
|
||
|
||
def modal_add_employee_view(request):
|
||
return render(request, 'partials/modal-add-employee.html')
|