lokses/ACORE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Правки логики

Browse Source
This commit is contained in:
lokses
2026-09-21 20:37:18 +03:00
parent f653a06820
commit 536679925a
30 changed files with 447 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/it_manager/views.py
+92 -26
View File
@@ -28,13 +28,52 @@ 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 _it_locations_qs():
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 (
Location.objects.filter(is_active=True)
accessible_locations(user).filter(is_active=True)
.select_related('location_region')
.prefetch_related(
models.Prefetch(
@@ -75,6 +114,8 @@ def _parse_assign_post(request) -> tuple[str, str | None, str | None, str | 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':
@@ -87,7 +128,7 @@ def _parse_assign_post(request) -> tuple[str, str | None, str | None, str | None
return target, None, None, (
'Оберіть аптеку.' if target == 'pharmacy' else 'Оберіть офіс / відділ.'
)
loc = Location.objects.filter(pk=location_id, is_active=True).first()
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':
@@ -182,7 +223,7 @@ def _apply_equipment_issue(unit, *, location_id=None, assigned_to_id=None, user=
def _it_manager_context(request):
equipment_qs = (
Equipment.objects.select_related(
scoped_queryset(request.user, Equipment.objects.all()).select_related(
'equipment_model__brand', 'equipment_model__equipment_type',
'assigned_to', 'location', 'status',
)
@@ -197,14 +238,14 @@ def _it_manager_context(request):
equipment_models = list(
EquipmentModel.objects.select_related('equipment_type', 'brand')
.annotate(units_total=models.Count('units'))
.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 = list(_it_locations_qs())
locations_all = list(_it_locations_qs(request.user))
it_pharmacies, it_offices = _split_it_locations(locations_all)
tickets_qs = (
Ticket.objects.select_related(
scoped_queryset(request.user, Ticket.objects.all()).select_related(
'requester', 'assignee', 'equipment', 'location',
'ticket_type', 'priority', 'status',
)
@@ -212,7 +253,7 @@ def _it_manager_context(request):
'comments__author',
models.Prefetch(
'work_logs',
queryset=WorkLog.objects.select_related('technician', 'work_type').order_by('-date', '-created_at'),
queryset=scoped_queryset(request.user, WorkLog.objects.all()).select_related('technician', 'work_type').order_by('-date', '-created_at'),
),
)
.annotate(worklogs_count=models.Count('work_logs', distinct=True))
@@ -250,7 +291,7 @@ def _it_manager_context(request):
worklog_month_id = 'all'
worklogs_qs = (
WorkLog.objects.select_related(
scoped_queryset(request.user, WorkLog.objects.all()).select_related(
'technician', 'location', 'equipment', 'ticket', 'work_type',
)
.prefetch_related(
@@ -279,7 +320,7 @@ def _it_manager_context(request):
stats_start, stats_end = month_start, None
worklog_stats_label = 'этом месяце'
month_logs = WorkLog.objects.filter(date__gte=stats_start)
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:
@@ -289,7 +330,7 @@ def _it_manager_context(request):
worklog_month_choices = []
seen_months = set()
for d in WorkLog.objects.dates('date', 'month', order='DESC'):
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}'))
@@ -303,7 +344,7 @@ def _it_manager_context(request):
worklog_technicians = (
Employee.objects.filter(is_active=True)
.filter(
Q(pk__in=WorkLog.objects.values('technician_id'))
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)
@@ -314,7 +355,7 @@ def _it_manager_context(request):
can_edit_worklogs = user_can_manage_it_monthly_report(request.user)
consumables_qs = (
Consumable.objects.select_related('consumable_type', 'color', 'status', 'equipment', 'location')
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)
@@ -325,10 +366,10 @@ def _it_manager_context(request):
providers = list(InternetProvider.objects.order_by('name'))
contracts_qs = (
InternetContract.objects.select_related(
scoped_queryset(request.user, InternetContract.objects.all()).select_related(
'provider', 'location', 'service_type', 'connection_type', 'status', 'parent_contract',
)
.prefetch_related('child_contracts')
.prefetch_related(models.Prefetch('child_contracts', queryset=scoped_queryset(request.user, InternetContract.objects.all())))
.order_by('provider__name', 'contract_number')
)
contracts = list(contracts_qs)
@@ -336,7 +377,7 @@ def _it_manager_context(request):
contracts_page = paginate(request, contracts_qs, page_param='contracts_page', size_param='contracts_per_page')
service_calls_qs = (
ServiceCall.objects.select_related('caller', 'handler', 'location', 'channel', 'ticket')
scoped_queryset(request.user, ServiceCall.objects.all()).select_related('caller', 'handler', 'location', 'channel', 'ticket')
.order_by('-date_time')
)
service_calls_page = paginate(
@@ -355,7 +396,7 @@ def _it_manager_context(request):
'it_locations': locations_all,
'it_pharmacies': it_pharmacies,
'it_offices': it_offices,
'it_employees': Employee.objects.filter(is_active=True).order_by('last_name', 'first_name'),
'it_employees': scoped_employees(request.user, Employee.objects.filter(is_active=True)).order_by('last_name', 'first_name'),
'tickets': tickets,
'tickets_page': tickets_page,
@@ -394,7 +435,7 @@ def _it_manager_context(request):
'contract_connection_types': _directory_choices('contract_connection_type'),
'contract_statuses': _directory_choices('contract_status'),
'framework_contracts': list(
InternetContract.objects.filter(is_framework=True)
scoped_queryset(request.user, InternetContract.objects.all()).filter(is_framework=True)
.select_related('provider')
.order_by('provider__name', 'contract_number')
),
@@ -408,6 +449,7 @@ def _it_manager_context(request):
@login_required
@_location_scoped_post
def it_manager_view(request):
if not _is_it_staff(request.user):
return deny_section_access(request)
@@ -428,6 +470,7 @@ def it_manager_view(request):
@login_required
@_location_scoped_post
def equipment_model_create_view(request):
"""Додати модель у каталог (тип + бренд + назва). Далі за нею
заводяться екземпляри лише з серійниками."""
@@ -459,6 +502,7 @@ def equipment_model_create_view(request):
@login_required
@_location_scoped_post
def equipment_bulk_create_view(request):
"""Масовий ввід екземплярів: обираємо модель з каталогу і вставляємо
список серійників (по одному в рядку) — створюється N одиниць на склад.
@@ -513,7 +557,7 @@ def equipment_bulk_create_view(request):
or _resolve_directory('equipment_status', 'stock')
existing = set(
Equipment.objects.filter(serial_number__in=serials)
scoped_queryset(request.user, Equipment.objects.all()).filter(serial_number__in=serials)
.values_list('serial_number', flat=True)
)
created = 0
@@ -552,6 +596,7 @@ def equipment_bulk_create_view(request):
@login_required
@_location_scoped_post
def equipment_bulk_assign_view(request):
"""Масова видача: тип (аптека / офіс / співробітник) → список."""
if not _is_it_staff(request.user):
@@ -569,7 +614,7 @@ def equipment_bulk_assign_view(request):
elif assign_err:
context_extra['equipment_form_error'] = assign_err
else:
units = Equipment.objects.filter(pk__in=ids)
units = scoped_queryset(request.user, Equipment.objects.all()).filter(pk__in=ids)
count = 0
for unit in units:
if _apply_equipment_issue(
@@ -590,12 +635,13 @@ def equipment_bulk_assign_view(request):
@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(Equipment, pk=pk)
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)
@@ -603,6 +649,8 @@ def equipment_update_view(request, pk):
# Повернення на склад: статус 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
@@ -639,12 +687,13 @@ def equipment_update_view(request, pk):
@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(Ticket, pk=pk)
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:
@@ -659,12 +708,13 @@ def ticket_update_view(request, pk):
@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(Ticket, pk=pk)
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(
@@ -678,6 +728,7 @@ def ticket_comment_view(request, pk):
@login_required
@_location_scoped_post
def worklog_create_view(request):
if not _is_it_staff(request.user):
return HttpResponseForbidden()
@@ -704,6 +755,8 @@ def worklog_create_view(request):
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
@@ -737,6 +790,7 @@ def worklog_create_view(request):
@login_required
@_location_scoped_post
def worklog_update_view(request, pk):
"""Inline-редактирование записи журнала — только руководителю отдела."""
if not _is_it_staff(request.user):
@@ -746,7 +800,7 @@ def worklog_update_view(request, pk):
if request.method != 'POST':
return HttpResponseForbidden()
log = get_object_or_404(WorkLog, pk=pk)
log = get_object_or_404(scoped_queryset(request.user, WorkLog.objects.all()), pk=pk)
description = request.POST.get('description', '').strip()
work_type_raw = request.POST.get('work_type', '')
work_type_id = _resolve_directory('worklog_type', work_type_raw)
@@ -782,6 +836,8 @@ def worklog_update_view(request, pk):
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
log.work_type_id = work_type_id
@@ -802,11 +858,14 @@ def worklog_update_view(request, pk):
@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', ''))
@@ -830,12 +889,13 @@ def consumable_create_view(request):
@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(Consumable, pk=pk)
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:
@@ -850,6 +910,7 @@ def consumable_update_view(request, pk):
@login_required
@_location_scoped_post
def provider_create_view(request):
if not _is_it_staff(request.user):
return HttpResponseForbidden()
@@ -875,6 +936,7 @@ def provider_create_view(request):
@login_required
@_location_scoped_post
def contract_create_view(request):
if not _is_it_staff(request.user):
return HttpResponseForbidden()
@@ -919,12 +981,13 @@ def contract_create_view(request):
@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(InternetContract, pk=pk)
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
@@ -950,11 +1013,14 @@ def contract_update_view(request, pk):
@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', ''))