from datetime import date, datetime import calendar from django.contrib.auth import get_user_model from django.contrib.auth.decorators import login_required from django.core.exceptions import ValidationError from django.db import IntegrityError, models from django.db.models import Count, Prefetch, Q, Sum from django.http import HttpResponse, HttpResponseForbidden from django.shortcuts import get_object_or_404 from django.utils import timezone from apps.it_manager.models import ( Consumable, Equipment, EquipmentAssignment, EquipmentModel, InternetContract, InternetProvider, ServiceCall, Ticket, TicketComment, WorkLog, ) from apps.it_manager.work_directory import ( catalog_payload_for_ui, detail_labels_from_db, get_catalog_work, parse_details_from_post, validate_worklog_details, ) from apps.it_manager.monthly_report import ( MONTHS_RU, user_can_manage_it_monthly_report, user_can_view_it_monthly_report, ) from apps.reports.it_monthly import parse_month_id from apps.locations.models.base import Location from apps.locations.models.pharmacies import Pharmacy from core.models.directories import Directory from core.portal_access import deny_section_access from core.utils.htmx import render_page from core.utils.pagination import paginate from core.location_access import ( accessible_locations, scoped_queryset, scoped_employees, has_all_locations, require_location, ) from functools import wraps from django.http import Http404 Employee = get_user_model() def _location_scoped_post(view): """Проверить все присланные ссылки до любых изменений, включая bulk.""" @wraps(view) def wrapped(request, *args, **kwargs): if request.method == 'POST' and not has_all_locations(request.user): for field, model in [('location_id', Location), ('equipment_id', Equipment), ('ticket_id', Ticket), ('parent_contract_id', InternetContract)]: value = request.POST.get(field) if value: if not value.isdigit(): raise Http404 qs = accessible_locations(request.user) if model is Location else scoped_queryset(request.user, model.objects.all()) get_object_or_404(qs, pk=value) for field in ('assigned_to_id', 'caller_id'): value = request.POST.get(field) if value: if not value.isdigit(): raise Http404 get_object_or_404(scoped_employees(request.user, Employee.objects.all()), pk=value) ids = request.POST.getlist('equipment_ids') if ids: if any(not value.isdigit() for value in ids): raise Http404 if scoped_queryset(request.user, Equipment.objects.all()).filter(pk__in=ids).count() != len(set(ids)): raise Http404 # Общий склад без локации и рамочные договоры относятся ко всей сети. if request.POST.get('is_framework'): return HttpResponseForbidden('Рамочные договоры требуют доступа ко всей сети.') return view(request, *args, **kwargs) return wrapped def _it_locations_qs(user): """Локации для селектов IT: аптеки с организацией (чтобы отличать №12 у разных юрлиц). Аптеки — все статусы (включая закрытые): в журнале работ нужны демонтаж, выгрузка и т.п. Офис/склад — только активные. """ return ( accessible_locations(user) .filter(Q(location_type='pharmacy') | Q(is_active=True)) .select_related('location_region') .prefetch_related( models.Prefetch( 'pharmacy', queryset=Pharmacy.objects.select_related('organization'), ), ) ) def _it_locations_sorted(user) -> list: """Офис/склад, затем аптеки по коду объекта (А1…А70, П1…), не по адресу.""" from apps.locations.models.base import location_sort_key return sorted(_it_locations_qs(user), key=location_sort_key) def _split_it_locations(locations): pharmacies, offices = [], [] for loc in locations: if loc.location_type == 'pharmacy': pharmacies.append(loc) else: # office + warehouse → «Офис / отдел» offices.append(loc) return pharmacies, offices def _infer_assign_target(equipment) -> str: if equipment.assigned_to_id: return 'employee' if equipment.location_id: loc = getattr(equipment, 'location', None) if loc is not None and loc.location_type == 'pharmacy': return 'pharmacy' return 'office' return '' def _parse_assign_post(request) -> tuple[str, str | None, str | None, str | None]: """assign_target, location_id, assigned_to_id, error_message.""" target = (request.POST.get('assign_target') or '').strip() location_id = request.POST.get('location_id') or None assigned_to_id = request.POST.get('assigned_to_id') or None if not target: if not has_all_locations(request.user): return '', None, None, 'Выберите доступную локацию или сотрудника.' return '', None, None, None if target == 'employee': if not assigned_to_id: return target, None, None, 'Оберіть співробітника.' return target, None, assigned_to_id, None if target in ('pharmacy', 'office'): if not location_id: return target, None, None, ( 'Оберіть аптеку.' if target == 'pharmacy' else 'Оберіть офіс / відділ.' ) loc = accessible_locations(request.user).filter(pk=location_id, is_active=True).first() if not loc: return target, None, None, 'Локацію не знайдено.' if target == 'pharmacy' and loc.location_type != 'pharmacy': return target, None, None, 'Оберіть аптеку зі списку аптек.' if target == 'office' and loc.location_type == 'pharmacy': return target, None, None, 'Оберіть офіс / відділ зі списку.' return target, str(loc.pk), None, None return target, None, None, 'Невідомий тип видачі.' def _is_it_staff(user): from core.portal_access import user_can_access return user_can_access(user, 'it_manager') def _it_assignees(): """Кандидаты в исполнители: группа IT, а также staff/superuser (если в группу IT ещё никого не добавили — селект не пустой).""" return ( Employee.objects.filter(is_active=True) .filter(Q(groups__name='IT') | Q(is_staff=True) | Q(is_superuser=True)) .distinct() .order_by('last_name', 'first_name') ) def _directory_choices(type_code): """(code, name)-кортежі активних записів довідника — та сама форма, що й колишні CharField(choices=...), для сумісності з шаблонами.""" return list( Directory.objects.filter(directory_type__code=type_code, is_active=True) .order_by('name').values_list('code', 'name') ) def _resolve_directory(type_code, code): """Резолвить код у Directory.pk для запису у FK-поле з форми (POST шле код, не pk — так шаблони лишаються незмінними).""" if not code: return None return Directory.objects.filter(directory_type__code=type_code, code=code).values_list('id', flat=True).first() def _close_open_assignments(equipment): """Закрити всі активні записи історії переміщень для одиниці.""" EquipmentAssignment.objects.filter( equipment=equipment, returned_at__isnull=True, ).update(returned_at=timezone.now()) def _apply_equipment_issue(unit, *, location_id=None, assigned_to_id=None, user=None, set_location=False, set_assignee=False, force_assigned_status=True): """Видача / переміщення одиниці з записом історії. set_location / set_assignee — чи оновлювати відповідне поле (для bulk: міняємо лише те, що передали у формі). """ new_location = location_id if set_location else unit.location_id new_assignee = assigned_to_id if set_assignee else unit.assigned_to_id loc_changed = str(unit.location_id or '') != str(new_location or '') emp_changed = str(unit.assigned_to_id or '') != str(new_assignee or '') if not loc_changed and not emp_changed: return False _close_open_assignments(unit) if set_location: unit.location_id = location_id or None if set_assignee: unit.assigned_to_id = assigned_to_id or None has_target = bool(unit.location_id or unit.assigned_to_id) if force_assigned_status and has_target: assigned_status_id = _resolve_directory('equipment_status', 'assigned') if assigned_status_id: unit.status_id = assigned_status_id elif not has_target: stock_id = _resolve_directory('equipment_status', 'stock') if stock_id: unit.status_id = stock_id unit.save() if has_target: EquipmentAssignment.objects.create( equipment=unit, assigned_to_id=unit.assigned_to_id, location_id=unit.location_id, assigned_by=user, ) return True def _it_manager_context(request): equipment_base = ( scoped_queryset(request.user, Equipment.objects.all()).select_related( 'equipment_model__brand', 'equipment_model__equipment_type', 'assigned_to', 'location', 'status', ) .order_by('equipment_model__equipment_type__name', 'equipment_model__name', 'serial_number') ) equipment = list(equipment_base) equipment_statuses = _directory_choices('equipment_status') equipment_counts = {code: 0 for code, _ in equipment_statuses} for row in equipment_base.values('status__code').annotate(c=Count('id')): code = row['status__code'] if code: equipment_counts[code] = row['c'] equipment_total_count = equipment_base.count() eq_q = (request.GET.get('q') or '').strip() eq_status = (request.GET.get('status') or '').strip() eq_type = (request.GET.get('equipment_type') or '').strip() eq_brand = (request.GET.get('brand') or '').strip() eq_location_raw = (request.GET.get('location_id') or '').strip() eq_location_id = int(eq_location_raw) if eq_location_raw.isdigit() else None eq_assignee_raw = (request.GET.get('assigned_to') or '').strip() eq_assigned_to_id = int(eq_assignee_raw) if eq_assignee_raw.isdigit() else None equipment_qs = equipment_base if eq_q: eq_q_filter = ( Q(serial_number__icontains=eq_q) | Q(inventory_number__icontains=eq_q) | Q(notes__icontains=eq_q) | Q(equipment_model__name__icontains=eq_q) | Q(equipment_model__brand__name__icontains=eq_q) ) # Скан EAN-13 (200XXXXXXXXXC) или QR EQ{id} digits = ''.join(c for c in eq_q if c.isdigit()) scan_pk = None if eq_q.upper().startswith('EQ') and eq_q[2:].isdigit(): scan_pk = int(eq_q[2:]) elif len(digits) >= 12 and digits.startswith('200'): try: scan_pk = int(digits[3:12]) except ValueError: scan_pk = None if scan_pk: eq_q_filter |= Q(pk=scan_pk) equipment_qs = equipment_qs.filter(eq_q_filter) if eq_status: equipment_qs = equipment_qs.filter(status__code=eq_status) if eq_type: equipment_qs = equipment_qs.filter(equipment_model__equipment_type__code=eq_type) if eq_brand.isdigit(): equipment_qs = equipment_qs.filter(equipment_model__brand_id=int(eq_brand)) if eq_location_id: equipment_qs = equipment_qs.filter(location_id=eq_location_id) if eq_assigned_to_id: equipment_qs = equipment_qs.filter(assigned_to_id=eq_assigned_to_id) equipment_filters = { 'q': eq_q, 'status': eq_status, 'equipment_type': eq_type, 'brand': eq_brand, 'location_id': eq_location_id, 'assigned_to': eq_assigned_to_id, } equipment_filters_active = any([ eq_q, eq_status, eq_type, eq_brand, eq_location_id, eq_assigned_to_id, ]) equipment_qs = equipment_qs.prefetch_related( Prefetch( 'assignments', queryset=EquipmentAssignment.objects.select_related( 'assigned_to', 'location', 'assigned_by', ).order_by('-assigned_at'), ), ) equipment_page = paginate(request, equipment_qs, page_param='equipment_page', size_param='equipment_per_page') equipment_models = list( EquipmentModel.objects.select_related('equipment_type', 'brand') .annotate(units_total=models.Count('units', filter=Q(units__in=scoped_queryset(request.user, Equipment.objects.all())))) .order_by('equipment_type__name', 'brand__name', 'name') ) locations_all = _it_locations_sorted(request.user) it_pharmacies, it_offices = _split_it_locations(locations_all) tickets_qs = ( scoped_queryset(request.user, Ticket.objects.all()).select_related( 'requester', 'assignee', 'equipment', 'location', 'ticket_type', 'priority', 'status', ) .prefetch_related( 'comments__author', models.Prefetch( 'work_logs', queryset=scoped_queryset(request.user, WorkLog.objects.all()).select_related( 'technician', 'work_type', 'catalog_work', 'catalog_work__parent', ).order_by('-date', '-created_at'), ), ) .annotate(worklogs_count=models.Count('work_logs', distinct=True)) .order_by('-created_at') ) tickets = list(tickets_qs) ticket_statuses = _directory_choices('ticket_status') ticket_counts = {code: 0 for code, _ in ticket_statuses} for ticket in tickets: ticket_counts[ticket.status.code] = ticket_counts.get(ticket.status.code, 0) + 1 ticket_overdue_count = sum(1 for t in tickets if t.is_overdue) tickets_page = paginate(request, tickets_qs, page_param='tickets_page', size_param='tickets_per_page') # Для формы журнала: сначала открытые заявки, потом остальные open_ticket_codes = ('new', 'in_progress', 'waiting') worklog_tickets = sorted( tickets, key=lambda t: (0 if t.status.code in open_ticket_codes else 1, -t.pk), ) month_start = timezone.localdate().replace(day=1) worklog_technician_raw = (request.GET.get('worklog_technician') or '').strip() worklog_technician_id = int(worklog_technician_raw) if worklog_technician_raw.isdigit() else None worklog_category_raw = (request.GET.get('worklog_category') or '').strip() worklog_category_id = int(worklog_category_raw) if worklog_category_raw.isdigit() else None worklog_month_raw = (request.GET.get('worklog_month') or '').strip() worklog_month_start = None worklog_month_end = None if worklog_month_raw and worklog_month_raw != 'all': y, m = parse_month_id(worklog_month_raw) last_day = calendar.monthrange(y, m)[1] worklog_month_start = date(y, m, 1) worklog_month_end = date(y, m, last_day) worklog_month_id = f'{y:04d}-{m:02d}' else: worklog_month_id = 'all' worklogs_qs = ( scoped_queryset(request.user, WorkLog.objects.all()).select_related( 'technician', 'location', 'equipment', 'ticket', 'work_type', 'catalog_work', 'catalog_work__parent', ) .prefetch_related( models.Prefetch( 'location__pharmacy', queryset=Pharmacy.objects.select_related('organization'), ), ) .order_by('-date', '-created_at') ) if worklog_technician_id: worklogs_qs = worklogs_qs.filter(technician_id=worklog_technician_id) if worklog_category_id: worklogs_qs = worklogs_qs.filter(catalog_work__parent_id=worklog_category_id) if worklog_month_start and worklog_month_end: worklogs_qs = worklogs_qs.filter( date__gte=worklog_month_start, date__lte=worklog_month_end, ) worklogs_page = paginate(request, worklogs_qs, page_param='worklogs_page', size_param='worklogs_per_page') it_work_catalog = catalog_payload_for_ui() worklog_detail_labels = it_work_catalog.get('detail_labels') or detail_labels_from_db() # Статистика: выбранный месяц или текущий календарный if worklog_month_start and worklog_month_end: stats_start, stats_end = worklog_month_start, worklog_month_end worklog_stats_label = ( f'{MONTHS_RU[worklog_month_start.month]} {worklog_month_start.year}' ) else: stats_start, stats_end = month_start, None worklog_stats_label = 'этом месяце' month_logs = scoped_queryset(request.user, WorkLog.objects.all()).filter(date__gte=stats_start) if stats_end is not None: month_logs = month_logs.filter(date__lte=stats_end) if worklog_technician_id: month_logs = month_logs.filter(technician_id=worklog_technician_id) month_hours = month_logs.aggregate(total=Sum('hours_spent'))['total'] or 0 month_count = month_logs.count() worklog_month_choices = [] seen_months = set() for d in scoped_queryset(request.user, WorkLog.objects.all()).dates('date', 'month', order='DESC'): mid = f'{d.year:04d}-{d.month:02d}' seen_months.add(mid) worklog_month_choices.append((mid, f'{MONTHS_RU[d.month]} {d.year}')) today = timezone.localdate() current_mid = f'{today.year:04d}-{today.month:02d}' if current_mid not in seen_months: worklog_month_choices.insert( 0, (current_mid, f'{MONTHS_RU[today.month]} {today.year}'), ) worklog_technicians = ( Employee.objects.filter(is_active=True) .filter( Q(pk__in=scoped_queryset(request.user, WorkLog.objects.all()).values('technician_id')) | Q(groups__name='IT') | Q(is_staff=True) | Q(is_superuser=True) ) .distinct() .order_by('last_name', 'first_name') ) can_edit_worklogs = user_can_manage_it_monthly_report(request.user) consumables_qs = ( scoped_queryset(request.user, Consumable.objects.all()).select_related('consumable_type', 'color', 'status', 'equipment', 'location') .order_by('equipment', 'consumable_type__name', 'color__name') ) consumables = list(consumables_qs) consumable_low_count = sum(1 for c in consumables if c.is_low) consumables_page = paginate(request, consumables_qs, page_param='consumables_page', size_param='consumables_per_page') printers = [e for e in equipment if e.equipment_type.code == 'printer'] providers = list(InternetProvider.objects.order_by('name')) contracts_qs = ( scoped_queryset(request.user, InternetContract.objects.all()).select_related( 'provider', 'location', 'service_type', 'connection_type', 'status', 'parent_contract', ) .prefetch_related(models.Prefetch('child_contracts', queryset=scoped_queryset(request.user, InternetContract.objects.all()))) .order_by('provider__name', 'contract_number') ) contracts = list(contracts_qs) contracts_expiring_count = sum(1 for c in contracts if c.is_expiring_soon) contracts_page = paginate(request, contracts_qs, page_param='contracts_page', size_param='contracts_per_page') service_calls_qs = ( scoped_queryset(request.user, ServiceCall.objects.all()).select_related('caller', 'handler', 'location', 'channel', 'ticket') .order_by('-date_time') ) service_calls_page = paginate( request, service_calls_qs, page_param='servicecalls_page', size_param='servicecalls_per_page', ) service_calls_count = service_calls_qs.count() return { 'equipment_list': equipment, 'equipment_page': equipment_page, 'equipment_total_count': equipment_total_count, 'equipment_counts': equipment_counts, 'equipment_filters': equipment_filters, 'equipment_filters_active': equipment_filters_active, 'equipment_types': _directory_choices('equipment_type'), 'equipment_statuses': equipment_statuses, 'equipment_brands': Directory.objects.filter(directory_type__code='equipment_brand', is_active=True).order_by('name'), 'equipment_models': equipment_models, 'it_locations': locations_all, 'it_pharmacies': it_pharmacies, 'it_offices': it_offices, 'it_employees': scoped_employees(request.user, Employee.objects.filter(is_active=True)).order_by('last_name', 'first_name'), 'tickets': tickets, 'tickets_page': tickets_page, 'worklog_tickets': worklog_tickets, 'ticket_counts': ticket_counts, 'ticket_overdue_count': ticket_overdue_count, 'ticket_statuses': ticket_statuses, 'ticket_priorities': _directory_choices('ticket_priority'), 'ticket_types': _directory_choices('ticket_type'), 'it_staff': _it_assignees(), 'worklogs_page': worklogs_page, 'worklog_types': _directory_choices('worklog_type'), 'worklog_technicians': worklog_technicians, 'worklog_technician_id': worklog_technician_id, 'worklog_category_id': worklog_category_id, 'worklog_month_id': worklog_month_id, 'worklog_month_choices': worklog_month_choices, 'worklog_stats_label': worklog_stats_label, 'it_can_edit_worklogs': can_edit_worklogs, 'it_work_catalog': it_work_catalog, 'worklog_detail_labels': worklog_detail_labels, 'month_hours': month_hours, 'month_count': month_count, 'consumables_page': consumables_page, 'consumables_count': len(consumables), 'consumable_low_count': consumable_low_count, 'consumable_types': _directory_choices('consumable_type'), 'consumable_colors': _directory_choices('consumable_color'), 'consumable_statuses': _directory_choices('consumable_status'), 'printers': printers, 'providers': providers, 'contracts_page': contracts_page, 'contracts_count': len(contracts), 'contracts_expiring_count': contracts_expiring_count, 'contract_service_types': _directory_choices('contract_service_type'), 'contract_connection_types': _directory_choices('contract_connection_type'), 'contract_statuses': _directory_choices('contract_status'), 'framework_contracts': list( scoped_queryset(request.user, InternetContract.objects.all()).filter(is_framework=True) .select_related('provider') .order_by('provider__name', 'contract_number') ), 'service_calls_page': service_calls_page, 'service_calls_count': service_calls_count, 'servicecall_channels': _directory_choices('servicecall_channel'), 'it_can_view_monthly_report': user_can_view_it_monthly_report(request.user), } @login_required @_location_scoped_post def it_manager_view(request): if not _is_it_staff(request.user): return deny_section_access(request) context = _it_manager_context(request) tab = (request.GET.get('tab') or '').strip() if tab in { 'equipment', 'tickets', 'worklog', 'consumables', 'providers', 'servicecalls', }: context['it_active_tab'] = tab elif ( 'worklog_technician' in request.GET or 'worklog_month' in request.GET or 'worklog_category' in request.GET or request.GET.get('worklogs_page') ): context['it_active_tab'] = 'worklog' elif ( request.GET.get('q') or request.GET.get('status') or request.GET.get('equipment_type') or request.GET.get('brand') or request.GET.get('location_id') or request.GET.get('assigned_to') or request.GET.get('equipment_page') ): context['it_active_tab'] = 'equipment' return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def equipment_model_create_view(request): """Додати модель у каталог (тип + бренд + назва). Далі за нею заводяться екземпляри лише з серійниками.""" if not _is_it_staff(request.user): return HttpResponseForbidden() context_extra = {} if request.method == 'POST': equipment_type_id = _resolve_directory('equipment_type', request.POST.get('equipment_type', '')) name = request.POST.get('name', '').strip() if equipment_type_id and name: warranty_months = request.POST.get('warranty_months', '').strip() EquipmentModel.objects.get_or_create( equipment_type_id=equipment_type_id, brand_id=request.POST.get('brand_id') or None, name=name, defaults={ 'warranty_months': int(warranty_months) if warranty_months.isdigit() else None, 'notes': request.POST.get('notes', '').strip(), }, ) else: context_extra['equipment_form_error'] = 'Вкажіть тип і назву моделі.' context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context.update(context_extra) return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def equipment_bulk_create_view(request): """Масовий ввід екземплярів: обираємо модель з каталогу і вставляємо список серійників (по одному в рядку) — створюється N одиниць на склад. Так більше не треба щоразу переписувати тип/бренд/модель.""" if not _is_it_staff(request.user): return HttpResponseForbidden() context_extra = {} if request.method == 'POST': model_id = request.POST.get('equipment_model_id') or None raw_serials = request.POST.get('serial_numbers', '') # серійники — по одному в рядку (або через кому); чистимо й дедуплікуємо serials, seen = [], set() for chunk in raw_serials.replace(',', '\n').splitlines(): s = chunk.strip() if s and s not in seen: seen.add(s) serials.append(s) raw_inventory = request.POST.get('inventory_numbers', '') inventory_list = None if raw_inventory.strip(): inv_lines = raw_inventory.splitlines() while inv_lines and not inv_lines[-1].strip(): inv_lines.pop() inventory_list = [(line.strip() or None) for line in inv_lines] model = EquipmentModel.objects.filter(pk=model_id).first() if model_id else None if not model: context_extra['equipment_form_error'] = 'Оберіть модель з каталогу.' elif not serials: context_extra['equipment_form_error'] = 'Вставте хоча б один серійний номер.' elif inventory_list is not None and len(inventory_list) != len(serials): context_extra['equipment_form_error'] = ( 'Количество инвентарных номеров должно совпадать с количеством серийных ' f'({len(serials)}).' ) else: purchase_date = request.POST.get('purchase_date') or None warranty_until = request.POST.get('warranty_until') or None # авто-розрахунок гарантії від дати покупки, якщо в моделі є термін if purchase_date and not warranty_until and model.warranty_months: try: d = datetime.strptime(purchase_date, '%Y-%m-%d').date() month = d.month - 1 + model.warranty_months year = d.year + month // 12 from calendar import monthrange day = min(d.day, monthrange(year, month % 12 + 1)[1]) warranty_until = d.replace(year=year, month=month % 12 + 1, day=day) except ValueError: warranty_until = None location_id = None assigned_to_id = None target, location_id, assigned_to_id, assign_err = _parse_assign_post(request) if assign_err: context_extra['equipment_form_error'] = assign_err context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context.update(context_extra) return render_page(request, 'it-manager.html', context) # якщо одразу видаємо — статус assigned, інакше stock status_code = 'assigned' if (location_id or assigned_to_id) else 'stock' status_id = _resolve_directory('equipment_status', status_code) \ or _resolve_directory('equipment_status', 'stock') if inventory_list: non_empty_inv = [inv for inv in inventory_list if inv] if len(non_empty_inv) != len(set(non_empty_inv)): context_extra['equipment_form_error'] = 'В списке есть повторяющиеся инвентарные номера.' context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context.update(context_extra) return render_page(request, 'it-manager.html', context) existing_inv = set( Equipment.objects.filter(inventory_number__in=non_empty_inv) .values_list('inventory_number', flat=True) ) if existing_inv: context_extra['equipment_form_error'] = ( 'Инвентарные номера уже заняты: ' + ', '.join(sorted(existing_inv)) ) context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context.update(context_extra) return render_page(request, 'it-manager.html', context) existing = set( scoped_queryset(request.user, Equipment.objects.all()).filter(serial_number__in=serials) .values_list('serial_number', flat=True) ) created = 0 for idx, s in enumerate(serials): if s in existing: continue inv = inventory_list[idx] if inventory_list else None try: unit = Equipment.objects.create( equipment_model=model, serial_number=s, inventory_number=inv, status_id=status_id, location_id=location_id, assigned_to_id=assigned_to_id, purchase_date=purchase_date, warranty_until=warranty_until, ) except IntegrityError: context_extra['equipment_form_error'] = ( f'Не удалось создать «{s}»: серийный или инвентарный номер уже существует.' ) context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context.update(context_extra) return render_page(request, 'it-manager.html', context) if location_id or assigned_to_id: EquipmentAssignment.objects.create( equipment=unit, location_id=location_id, assigned_to_id=assigned_to_id, assigned_by=request.user, ) created += 1 skipped = len(serials) - created msg = f'Додано {created} од. «{model}»' if (location_id or assigned_to_id) and created: msg += ' і видано' if skipped: msg += f'; пропущено {skipped} (серійник вже існує)' context_extra['equipment_form_message'] = msg context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context.update(context_extra) return render_page(request, 'it-manager.html', context) @login_required def equipment_labels_view(request): """Печатная страница этикеток по шаблону; codes=qr|ean|both.""" from django.shortcuts import render from apps.it_manager.equipment_labels import ( get_default_template, label_payload_for_unit, parse_label_codes, ) if not _is_it_staff(request.user): return deny_section_access(request) codes = parse_label_codes(request.GET.get('codes')) raw_ids = request.GET.getlist('ids') or request.GET.get('ids', '').split(',') ids = [] for part in raw_ids: part = (part or '').strip() if part.isdigit(): ids.append(int(part)) seen, ordered = set(), [] for pk in ids: if pk not in seen: seen.add(pk) ordered.append(pk) template = get_default_template() units_by_id = { u.pk: u for u in scoped_queryset(request.user, Equipment.objects.all()) .filter(pk__in=ordered) .select_related( 'equipment_model', 'equipment_model__brand', 'equipment_model__equipment_type', ) } labels = [] for pk in ordered: unit = units_by_id.get(pk) if unit is not None: labels.append(label_payload_for_unit(unit, codes=codes, template=template)) return render(request, 'pages/it-equipment-labels.html', { 'labels': labels, 'codes': codes, 'template_width_mm': template.width_mm, 'template_height_mm': template.height_mm, }) @login_required def equipment_label_template_view(request): """Редактор общего шаблона этикетки (GET страница / POST JSON).""" import json from django.http import JsonResponse from django.shortcuts import render from apps.it_manager.equipment_labels import get_default_template, normalize_layout if not _is_it_staff(request.user): return deny_section_access(request) template = get_default_template() if request.method == 'POST': try: payload = json.loads(request.body.decode('utf-8') or '{}') except (json.JSONDecodeError, UnicodeDecodeError): return JsonResponse({'ok': False, 'error': 'Некорректный JSON'}, status=400) try: width = float(payload.get('width_mm') or template.width_mm) height = float(payload.get('height_mm') or template.height_mm) except (TypeError, ValueError): return JsonResponse({'ok': False, 'error': 'Некорректный размер'}, status=400) if width < 20 or width > 200 or height < 15 or height > 200: return JsonResponse({'ok': False, 'error': 'Размер вне диапазона 20–200 × 15–200 мм'}, status=400) layout = normalize_layout(payload.get('layout')) if not layout: return JsonResponse({'ok': False, 'error': 'Добавьте хотя бы один блок'}, status=400) template.width_mm = width template.height_mm = height template.layout = layout template.updated_by = request.user template.save() return JsonResponse({'ok': True, 'updated_at': template.updated_at.isoformat()}) return render(request, 'pages/it-equipment-label-template.html', { 'template': template, 'template_json': json.dumps({ 'width_mm': template.width_mm, 'height_mm': template.height_mm, 'layout': template.layout or [], }, ensure_ascii=False), }) @login_required @_location_scoped_post def equipment_bulk_assign_view(request): """Масова видача: тип (аптека / офіс / співробітник) → список.""" if not _is_it_staff(request.user): return HttpResponseForbidden() context_extra = {} if request.method == 'POST': ids = request.POST.getlist('equipment_ids') target, location_id, assigned_to_id, assign_err = _parse_assign_post(request) if not ids: context_extra['equipment_form_error'] = 'Відмітьте хоча б один екземпляр.' elif not target: context_extra['equipment_form_error'] = 'Оберіть тип видачі.' elif assign_err: context_extra['equipment_form_error'] = assign_err else: units = scoped_queryset(request.user, Equipment.objects.all()).filter(pk__in=ids) count = 0 for unit in units: if _apply_equipment_issue( unit, location_id=location_id, assigned_to_id=assigned_to_id, user=request.user, set_location=True, set_assignee=True, ): count += 1 context_extra['equipment_form_message'] = f'Видано {count} од. обладнання' context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context.update(context_extra) return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def equipment_update_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': equipment = get_object_or_404(scoped_queryset(request.user, Equipment.objects.all()), pk=pk) new_status_code = request.POST.get('status') or '' new_status_id = _resolve_directory('equipment_status', new_status_code) target, location_id, assigned_to_id, assign_err = _parse_assign_post(request) inventory_number = (request.POST.get('inventory_number') or '').strip() or None notes = request.POST.get('notes', '') purchase_date = request.POST.get('purchase_date') or None warranty_until = request.POST.get('warranty_until') or None if inventory_number and ( Equipment.objects.exclude(pk=equipment.pk) .filter(inventory_number=inventory_number) .exists() ): context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context['equipment_form_error'] = ( f'Инвентарный номер «{inventory_number}» уже используется.' ) return render_page(request, 'it-manager.html', context) def _apply_meta(unit): unit.inventory_number = inventory_number unit.notes = notes unit.purchase_date = purchase_date unit.warranty_until = warranty_until # Повернення на склад: статус stock або порожній тип видачі if new_status_code == 'stock' or not target: if not has_all_locations(request.user): return HttpResponseForbidden('Общий склад требует доступа ко всей сети.') _close_open_assignments(equipment) if new_status_code == 'stock' and new_status_id: equipment.status_id = new_status_id else: stock_id = _resolve_directory('equipment_status', 'stock') if stock_id: equipment.status_id = stock_id equipment.location_id = None equipment.assigned_to_id = None _apply_meta(equipment) try: equipment.save() except IntegrityError: context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context['equipment_form_error'] = ( f'Инвентарный номер «{inventory_number}» уже используется.' ) return render_page(request, 'it-manager.html', context) elif assign_err: context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context['equipment_form_error'] = assign_err return render_page(request, 'it-manager.html', context) else: _apply_equipment_issue( equipment, location_id=location_id, assigned_to_id=assigned_to_id, user=request.user, set_location=True, set_assignee=True, force_assigned_status=True, ) equipment.refresh_from_db() _apply_meta(equipment) if new_status_id and new_status_code not in ('', 'stock') and equipment.status.code != new_status_code: equipment.status_id = new_status_id try: equipment.save() except IntegrityError: context = _it_manager_context(request) context['it_active_tab'] = 'equipment' context['equipment_form_error'] = ( f'Инвентарный номер «{inventory_number}» уже используется.' ) return render_page(request, 'it-manager.html', context) context = _it_manager_context(request) context['it_active_tab'] = 'equipment' return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def ticket_update_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': ticket = get_object_or_404(scoped_queryset(request.user, Ticket.objects.all()), pk=pk) new_status_id = _resolve_directory('ticket_status', request.POST.get('status')) if new_status_id: ticket.status_id = new_status_id ticket.assignee_id = request.POST.get('assignee_id') or None ticket.save() context = _it_manager_context(request) context['it_active_tab'] = 'tickets' return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def ticket_comment_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': ticket = get_object_or_404(scoped_queryset(request.user, Ticket.objects.all()), pk=pk) text = request.POST.get('text', '').strip() if text: TicketComment.objects.create( ticket=ticket, author=request.user, text=text, is_internal=bool(request.POST.get('is_internal')), ) context = _it_manager_context(request) context['it_active_tab'] = 'tickets' return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def worklog_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() worklog_error = '' worklog_message = '' if request.method == 'POST': catalog_work = get_catalog_work(request.POST.get('catalog_work_id')) description = request.POST.get('description', '').strip() details = parse_details_from_post(request.POST, catalog_work) if not catalog_work: worklog_error = 'Выберите типовую работу из справочника.' else: details_err = validate_worklog_details(catalog_work, details) if details_err: worklog_error = details_err else: try: log_date = datetime.strptime(request.POST.get('date', ''), '%Y-%m-%d').date() except ValueError: log_date = timezone.localdate() try: hours_spent = float(request.POST.get('hours_spent', '').strip() or 0) or None except ValueError: hours_spent = None location_id = request.POST.get('location_id') or None if not location_id and not has_all_locations(request.user): return HttpResponseForbidden('Укажите доступную локацию.') equipment_id = request.POST.get('equipment_id') or None ticket_id = request.POST.get('ticket_id') or None try: WorkLog.objects.create( date=log_date, technician=request.user, catalog_work=catalog_work, details=details, work_type_id=None, location_id=location_id, equipment_id=equipment_id, ticket_id=ticket_id, description=description, result=request.POST.get('result', '').strip(), hours_spent=hours_spent, ) except Exception: worklog_error = 'Не удалось сохранить запись. Проверьте локацию, заявку и оборудование.' else: if ticket_id: worklog_message = 'Работа привязана к заявке и сохранена.' else: worklog_message = 'Запись о выполненной работе сохранена.' context = _it_manager_context(request) context['it_active_tab'] = 'worklog' if worklog_message: context['it_worklog_message'] = worklog_message if worklog_error: context['it_worklog_error'] = worklog_error return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def worklog_update_view(request, pk): """Inline-редактирование записи журнала — только руководителю отдела.""" if not _is_it_staff(request.user): return HttpResponseForbidden() if not user_can_manage_it_monthly_report(request.user): return HttpResponseForbidden() if request.method != 'POST': return HttpResponseForbidden() log = get_object_or_404(scoped_queryset(request.user, WorkLog.objects.all()), pk=pk) description = request.POST.get('description', '').strip() catalog_raw = (request.POST.get('catalog_work_id') or '').strip() catalog_work = get_catalog_work(catalog_raw) if catalog_raw else None # Если запись уже из справочника — сохраняем FK даже без смены в форме if not catalog_work and log.catalog_work_id and not catalog_raw: catalog_work = log.catalog_work work_type_raw = request.POST.get('work_type', '') work_type_id = _resolve_directory('worklog_type', work_type_raw) if work_type_raw else None if catalog_work: details = parse_details_from_post(request.POST, catalog_work) # если detail_* не пришли — оставить прежние значения для ключей без POST if log.catalog_work_id == catalog_work.pk and log.details: merged = dict(log.details) for key in details: if f'detail_{key}' in request.POST: merged[key] = details[key] details = merged details_err = validate_worklog_details(catalog_work, details) if details_err: return HttpResponse( f'{details_err}', status=400, ) elif log.catalog_work_id: return HttpResponse( 'Укажите типовую работу', status=400, ) else: # Историческая запись без справочника if not description: return HttpResponse( 'Нужно описание', status=400, ) if not work_type_id or work_type_raw == '__new__': return HttpResponse( 'Укажите вид работ', status=400, ) try: log_date = datetime.strptime(request.POST.get('date', ''), '%Y-%m-%d').date() except ValueError: log_date = log.date try: hours_raw = (request.POST.get('hours_spent') or '').strip() hours_spent = float(hours_raw) if hours_raw else None if hours_spent == 0: hours_spent = None except ValueError: hours_spent = log.hours_spent technician_raw = (request.POST.get('technician_id') or '').strip() technician_id = int(technician_raw) if technician_raw.isdigit() else log.technician_id if not Employee.objects.filter(pk=technician_id, is_active=True).exists(): technician_id = log.technician_id location_raw = (request.POST.get('location_id') or '').strip() location_id = int(location_raw) if location_raw.isdigit() else None if not location_id and not has_all_locations(request.user): return HttpResponseForbidden('Укажите доступную локацию.') log.date = log_date log.technician_id = technician_id if catalog_work: log.catalog_work = catalog_work log.details = details log.work_type_id = None else: log.work_type_id = work_type_id log.description = description log.result = request.POST.get('result', '').strip() log.hours_spent = hours_spent log.location_id = location_id try: log.save() except Exception: return HttpResponse( 'Ошибка сохранения', status=400, ) return HttpResponse( 'Сохранено' ) @login_required @_location_scoped_post def worklog_delete_view(request, pk): """Удаление записи журнала — только руководителю отдела.""" if not _is_it_staff(request.user): return HttpResponseForbidden() if not user_can_manage_it_monthly_report(request.user): return HttpResponseForbidden() if request.method != 'POST': return HttpResponseForbidden() log = get_object_or_404(scoped_queryset(request.user, WorkLog.objects.all()), pk=pk) log.delete() # Пустой ответ + hx-swap outerHTML на убирает строку из таблицы return HttpResponse('') @login_required @_location_scoped_post def consumable_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': if not request.POST.get('location_id') and not has_all_locations(request.user): return HttpResponseForbidden('Укажите доступную локацию.') model_number = request.POST.get('model_number', '').strip() consumable_type_id = _resolve_directory('consumable_type', request.POST.get('consumable_type', '')) color_id = _resolve_directory('consumable_color', request.POST.get('color', '')) if model_number and consumable_type_id and color_id: Consumable.objects.create( consumable_type_id=consumable_type_id, brand=request.POST.get('brand', '').strip(), model_number=model_number, color_id=color_id, equipment_id=request.POST.get('equipment_id') or None, location_id=request.POST.get('location_id') or None, page_yield=request.POST.get('page_yield') or None, purchase_date=request.POST.get('purchase_date') or None, notes=request.POST.get('notes', '').strip(), ) context = _it_manager_context(request) context['it_active_tab'] = 'consumables' return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def consumable_update_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': consumable = get_object_or_404(scoped_queryset(request.user, Consumable.objects.all()), pk=pk) new_status_code = request.POST.get('status') new_status_id = _resolve_directory('consumable_status', new_status_code) if new_status_id: consumable.status_id = new_status_id if new_status_code == 'installed' and not consumable.installed_date: consumable.installed_date = timezone.localdate() consumable.save() context = _it_manager_context(request) context['it_active_tab'] = 'consumables' return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def provider_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': name = request.POST.get('name', '').strip() if name: InternetProvider.objects.get_or_create( name=name, defaults={ 'phone': request.POST.get('phone', '').strip(), 'email': request.POST.get('email', '').strip(), 'website': request.POST.get('website', '').strip(), 'support_phone': request.POST.get('support_phone', '').strip(), 'support_email': request.POST.get('support_email', '').strip(), 'notes': request.POST.get('notes', '').strip(), }, ) context = _it_manager_context(request) context['it_active_tab'] = 'providers' return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def contract_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': provider_id = request.POST.get('provider_id') or None location_id = request.POST.get('location_id') or None contract_number = request.POST.get('contract_number', '').strip() start_date = request.POST.get('start_date') or None service_type_id = _resolve_directory('contract_service_type', request.POST.get('service_type', '')) is_framework = bool(request.POST.get('is_framework')) location_id = None if is_framework else (request.POST.get('location_id') or None) if provider_id and contract_number and start_date and service_type_id and (is_framework or location_id): parent_id = request.POST.get('parent_contract_id') or None contract = InternetContract( provider_id=provider_id, location_id=location_id, contract_number=contract_number, service_type_id=service_type_id, connection_type_id=_resolve_directory('contract_connection_type', request.POST.get('connection_type', '')), start_date=start_date, end_date=request.POST.get('end_date') or None, monthly_cost=request.POST.get('monthly_cost') or None, bandwidth_down=request.POST.get('bandwidth_down') or None, bandwidth_up=request.POST.get('bandwidth_up') or None, account_number=request.POST.get('account_number', '').strip(), static_ips=request.POST.get('static_ips', '').strip() if request.POST.get('dedicated_ip') else '', notes=request.POST.get('notes', '').strip(), is_framework=is_framework, parent_contract_id=parent_id if not is_framework else None, ) try: contract.full_clean() contract.save() except ValidationError: pass context = _it_manager_context(request) context['it_active_tab'] = 'providers' return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def contract_update_view(request, pk): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': contract = get_object_or_404(scoped_queryset(request.user, InternetContract.objects.all()), pk=pk) new_status_id = _resolve_directory('contract_status', request.POST.get('status')) if new_status_id: contract.status_id = new_status_id parent_id = request.POST.get('parent_contract_id') if parent_id is not None: contract.parent_contract_id = parent_id or None if 'is_framework' in request.POST: contract.is_framework = bool(request.POST.get('is_framework')) if contract.is_framework: contract.parent_contract_id = None contract.location_id = None elif contract.is_framework: contract.location_id = None try: contract.full_clean() contract.save() except ValidationError: pass context = _it_manager_context(request) context['it_active_tab'] = 'providers' return render_page(request, 'it-manager.html', context) @login_required @_location_scoped_post def servicecall_create_view(request): if not _is_it_staff(request.user): return HttpResponseForbidden() if request.method == 'POST': if not request.POST.get('location_id') and not has_all_locations(request.user): return HttpResponseForbidden('Укажите доступную локацию.') caller_id = request.POST.get('caller_id') or None description = request.POST.get('description', '').strip() channel_id = _resolve_directory('servicecall_channel', request.POST.get('channel', '')) if caller_id and description and channel_id: ServiceCall.objects.create( caller_id=caller_id, handler=request.user, channel_id=channel_id, location_id=request.POST.get('location_id') or None, description=description, resolution=request.POST.get('resolution', '').strip(), minutes_spent=request.POST.get('minutes_spent') or None, ticket_id=request.POST.get('ticket_id') or None, ) context = _it_manager_context(request) context['it_active_tab'] = 'servicecalls' return render_page(request, 'it-manager.html', context)