471 lines
17 KiB
Python
471 lines
17 KiB
Python
"""ViewSets для API модуля закупок."""
|
||
|
||
from datetime import date
|
||
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from django.db.models import Q
|
||
from django_filters.rest_framework import DjangoFilterBackend
|
||
from drf_spectacular.utils import OpenApiParameter, extend_schema
|
||
from rest_framework import filters, status, viewsets
|
||
from rest_framework.decorators import action
|
||
from rest_framework.response import Response
|
||
|
||
from communications import comments_helpers
|
||
from communications.models import Message
|
||
from employees.models import Department
|
||
from procurement.constants import (
|
||
EquipmentStatus,
|
||
)
|
||
from procurement.models import (
|
||
Equipment,
|
||
EquipmentCategory,
|
||
EquipmentTransferLog,
|
||
MaintenanceRecord,
|
||
)
|
||
|
||
from ..equipment_policy import EquipmentCreationPolicy
|
||
from ..permissions import (
|
||
CanManageEquipment,
|
||
CanManageEquipmentCategory,
|
||
)
|
||
from ..serializers import (
|
||
EquipmentCategorySerializer,
|
||
EquipmentDetailSerializer,
|
||
EquipmentListSerializer,
|
||
MaintenanceRecordSerializer,
|
||
)
|
||
|
||
|
||
class EquipmentCategoryViewSet(viewsets.ModelViewSet):
|
||
"""ViewSet для категорий оборудования."""
|
||
|
||
queryset = EquipmentCategory.objects.prefetch_related("children")
|
||
serializer_class = EquipmentCategorySerializer
|
||
filter_backends = [filters.SearchFilter, filters.OrderingFilter]
|
||
search_fields = ["name", "description"]
|
||
ordering = ["name"]
|
||
|
||
def get_permissions(self):
|
||
"""Чтение доступно всем сотрудникам, изменение только уполномоченным."""
|
||
return [CanManageEquipmentCategory()]
|
||
|
||
@action(detail=True, methods=["get"])
|
||
def children(self, request, pk=None):
|
||
"""Получить подкатегории."""
|
||
category = self.get_object()
|
||
serializer = self.get_serializer(category.children.all(), many=True)
|
||
return Response(serializer.data)
|
||
|
||
@action(detail=False, methods=["get"])
|
||
def tree(self, request):
|
||
"""Получить дерево категорий."""
|
||
# Получаем только корневые категории
|
||
root_categories = self.get_queryset().filter(parent=None)
|
||
serializer = self.get_serializer(root_categories, many=True)
|
||
return Response(serializer.data)
|
||
|
||
|
||
class EquipmentViewSet(viewsets.ModelViewSet):
|
||
"""ViewSet для оборудования."""
|
||
|
||
queryset = Equipment.objects.select_related(
|
||
"category", "department", "responsible_person"
|
||
)
|
||
filter_backends = [
|
||
DjangoFilterBackend,
|
||
filters.SearchFilter,
|
||
filters.OrderingFilter,
|
||
]
|
||
filterset_fields = {
|
||
"status": ["exact"],
|
||
"category": ["exact"],
|
||
"department": ["exact"],
|
||
"responsible_person": ["exact"],
|
||
"purchase_date": ["gte", "lte"],
|
||
}
|
||
search_fields = [
|
||
"name",
|
||
"inventory_number",
|
||
"serial_number",
|
||
"location",
|
||
]
|
||
ordering_fields = ["purchase_date", "name", "created_at"]
|
||
ordering = ["-created_at"]
|
||
|
||
def get_queryset(self):
|
||
"""Annotate comments_count."""
|
||
from django.db.models import Count, IntegerField, OuterRef, Subquery
|
||
|
||
from communications.models import Chat
|
||
|
||
ct = ContentType.objects.get_for_model(Equipment)
|
||
qs = super().get_queryset()
|
||
|
||
comments_sub = (
|
||
Chat.objects.filter(
|
||
type="comments",
|
||
context_content_type=ct,
|
||
context_object_id=OuterRef("pk"),
|
||
)
|
||
.annotate(msg_count=Count("messages", filter=Q(messages__is_deleted=False)))
|
||
.values("msg_count")[:1]
|
||
)
|
||
return qs.annotate(
|
||
comments_count=Subquery(comments_sub, output_field=IntegerField(default=0))
|
||
)
|
||
|
||
def get_serializer_class(self):
|
||
"""Выбрать сериализатор."""
|
||
if self.action == "retrieve":
|
||
return EquipmentDetailSerializer
|
||
return EquipmentListSerializer
|
||
|
||
def get_permissions(self):
|
||
"""Права доступа."""
|
||
permission_classes = [CanManageEquipment]
|
||
return [permission() for permission in permission_classes]
|
||
|
||
def create(self, request, *args, **kwargs):
|
||
"""Создание оборудования с поддержкой массового создания.
|
||
|
||
Если передан параметр quantity > 1, создаётся несколько единиц
|
||
с автоматически сгенерированными инвентарными номерами.
|
||
"""
|
||
try:
|
||
quantity = int(request.data.get("quantity", 1))
|
||
except (ValueError, TypeError):
|
||
quantity = 1
|
||
quantity = max(1, min(quantity, 100)) # Ограничение от 1 до 100
|
||
|
||
if quantity == 1:
|
||
# Стандартное создание одной единицы
|
||
return super().create(request, *args, **kwargs)
|
||
|
||
# Массовое создание
|
||
created_equipment = []
|
||
errors = []
|
||
|
||
for i in range(quantity):
|
||
serializer = self.get_serializer(data=request.data)
|
||
if serializer.is_valid():
|
||
self.perform_create(serializer)
|
||
created_equipment.append(serializer.data)
|
||
else:
|
||
errors.append({"index": i, "errors": serializer.errors})
|
||
|
||
if errors and not created_equipment:
|
||
return Response(
|
||
{"detail": "Не удалось создать оборудование", "errors": errors},
|
||
status=status.HTTP_400_BAD_REQUEST,
|
||
)
|
||
|
||
return Response(
|
||
{
|
||
"created_count": len(created_equipment),
|
||
"equipment": created_equipment,
|
||
"errors": errors if errors else None,
|
||
},
|
||
status=status.HTTP_201_CREATED,
|
||
)
|
||
|
||
@action(detail=False, methods=["get"])
|
||
def my_equipment(self, request):
|
||
"""Оборудование, за которое отвечает текущий пользователь."""
|
||
queryset = self.get_queryset().filter(responsible_person=request.user)
|
||
page = self.paginate_queryset(queryset)
|
||
if page is not None:
|
||
serializer = self.get_serializer(page, many=True)
|
||
return self.get_paginated_response(serializer.data)
|
||
serializer = self.get_serializer(queryset, many=True)
|
||
return Response(serializer.data)
|
||
|
||
@action(detail=False, methods=["get"])
|
||
def warranty_expiring(self, request):
|
||
"""Оборудование с истекающей гарантией (< 30 дней)."""
|
||
from datetime import timedelta
|
||
|
||
threshold = date.today() + timedelta(days=30)
|
||
|
||
queryset = self.get_queryset().filter(
|
||
warranty_until__lte=threshold, warranty_until__gte=date.today()
|
||
)
|
||
serializer = self.get_serializer(queryset, many=True)
|
||
return Response(serializer.data)
|
||
|
||
@action(detail=False, methods=["get"], url_path="generate-inventory-number")
|
||
def generate_inventory_number(self, request):
|
||
"""Генерация уникального инвентарного номера."""
|
||
from procurement.services import InventoryNumberGenerator
|
||
|
||
inventory_number = InventoryNumberGenerator.generate()
|
||
return Response({"inventory_number": inventory_number})
|
||
|
||
@action(detail=False, methods=["get"], url_path="create-options")
|
||
def create_options(self, request):
|
||
"""Возвращает доступные опции для создания оборудования.
|
||
|
||
Определяет уровень прав пользователя и возвращает:
|
||
- allowed_departments: список отделов для выбора
|
||
- can_choose_department: может ли пользователь выбирать отдел
|
||
- can_choose_responsible: может ли выбирать ответственного
|
||
- default_responsible: ответственный по умолчанию (если нет выбора)
|
||
"""
|
||
policy = EquipmentCreationPolicy.resolve(request.user)
|
||
return Response(policy.as_options())
|
||
|
||
@action(detail=True, methods=["post"])
|
||
def transfer(self, request, pk=None):
|
||
"""Перевод оборудования в другой отдел или другому пользователю."""
|
||
equipment = self.get_object()
|
||
|
||
to_department_id = request.data.get("to_department")
|
||
to_person_id = request.data.get("to_person")
|
||
to_location = request.data.get("to_location")
|
||
reason = request.data.get("reason", "")
|
||
|
||
if not to_department_id and not to_person_id:
|
||
return Response(
|
||
{"error": "Укажите отдел или ответственного"},
|
||
status=status.HTTP_400_BAD_REQUEST,
|
||
)
|
||
|
||
# Сохраняем старые значения
|
||
from_department = equipment.department
|
||
from_person = equipment.responsible_person
|
||
|
||
# Обновляем оборудование
|
||
if to_department_id:
|
||
try:
|
||
to_department = Department.objects.get(pk=to_department_id)
|
||
equipment.department = to_department
|
||
except Department.DoesNotExist:
|
||
return Response(
|
||
{"error": "Отдел не найден"},
|
||
status=status.HTTP_404_NOT_FOUND,
|
||
)
|
||
else:
|
||
to_department = from_department
|
||
|
||
if to_person_id:
|
||
from django.contrib.auth import get_user_model
|
||
|
||
User = get_user_model()
|
||
try:
|
||
to_person = User.objects.get(pk=to_person_id, is_active=True)
|
||
equipment.responsible_person = to_person
|
||
except User.DoesNotExist:
|
||
return Response(
|
||
{"error": "Активный сотрудник не найден"},
|
||
status=status.HTTP_400_BAD_REQUEST,
|
||
)
|
||
else:
|
||
to_person = from_person
|
||
|
||
if to_location:
|
||
equipment.location = to_location
|
||
|
||
equipment.save()
|
||
|
||
# Создаём лог перевода
|
||
EquipmentTransferLog.objects.create(
|
||
equipment=equipment,
|
||
from_department=from_department,
|
||
to_department=to_department,
|
||
from_person=from_person,
|
||
to_person=to_person,
|
||
to_location=to_location or "",
|
||
reason=reason,
|
||
created_by=request.user,
|
||
)
|
||
|
||
return Response(
|
||
{
|
||
"status": "transferred",
|
||
"equipment_id": equipment.id,
|
||
"from_department": str(from_department),
|
||
"to_department": str(to_department),
|
||
}
|
||
)
|
||
|
||
@action(detail=True, methods=["post"])
|
||
def write_off(self, request, pk=None):
|
||
"""Списание оборудования."""
|
||
equipment = self.get_object()
|
||
reason = request.data.get("reason", "")
|
||
|
||
if equipment.status == EquipmentStatus.RETIRED:
|
||
return Response(
|
||
{"error": "Оборудование уже списано"},
|
||
status=status.HTTP_400_BAD_REQUEST,
|
||
)
|
||
|
||
equipment.status = EquipmentStatus.RETIRED
|
||
equipment.notes = (equipment.notes or "") + f"\n\nСписано: {reason}"
|
||
equipment.save()
|
||
|
||
return Response({"status": "written_off", "equipment_id": equipment.id})
|
||
|
||
@action(detail=True, methods=["post"])
|
||
def add_maintenance(self, request, pk=None):
|
||
"""Добавить запись об обслуживании."""
|
||
|
||
equipment = self.get_object()
|
||
|
||
maintenance_type = request.data.get("type", "repair")
|
||
description = request.data.get("description", "")
|
||
cost = request.data.get("cost")
|
||
maintenance_date = request.data.get("date", date.today())
|
||
|
||
record = MaintenanceRecord.objects.create(
|
||
equipment=equipment,
|
||
type=maintenance_type,
|
||
description=description,
|
||
cost=cost,
|
||
date=maintenance_date,
|
||
performed_by=request.user,
|
||
)
|
||
|
||
return Response(
|
||
{
|
||
"status": "created",
|
||
"maintenance_id": record.id,
|
||
"equipment_id": equipment.id,
|
||
},
|
||
status=status.HTTP_201_CREATED,
|
||
)
|
||
|
||
@action(detail=True, methods=["get"])
|
||
def transfer_history(self, request, pk=None):
|
||
"""История переводов оборудования."""
|
||
equipment = self.get_object()
|
||
|
||
logs = EquipmentTransferLog.objects.filter(equipment=equipment).order_by(
|
||
"-created_at"
|
||
)
|
||
|
||
data = [
|
||
{
|
||
"id": log.id,
|
||
"from_department": str(log.from_department),
|
||
"to_department": str(log.to_department),
|
||
"from_person": (str(log.from_person) if log.from_person else None),
|
||
"to_person": (str(log.to_person) if log.to_person else None),
|
||
"reason": log.reason,
|
||
"created_by": str(log.created_by) if log.created_by else None,
|
||
"date": log.created_at.isoformat(),
|
||
}
|
||
for log in logs
|
||
]
|
||
|
||
return Response(data)
|
||
|
||
@action(detail=True, methods=["get", "post"])
|
||
def comments(self, request, pk=None):
|
||
"""Список/создание комментариев для оборудования."""
|
||
from api.v1.employees.serializers.employee import (
|
||
EmployeeBriefSerializer,
|
||
)
|
||
|
||
equipment = self.get_object()
|
||
|
||
if request.method in ("GET", "HEAD"):
|
||
messages = comments_helpers.get_comments(equipment)
|
||
comments_data = []
|
||
for msg in messages:
|
||
author_ser = EmployeeBriefSerializer(msg.author)
|
||
comments_data.append(
|
||
{
|
||
"id": msg.id,
|
||
"equipment": equipment.id,
|
||
"author": author_ser.data,
|
||
"text": msg.content,
|
||
"created_at": msg.created_at,
|
||
}
|
||
)
|
||
return Response(comments_data)
|
||
|
||
text = request.data.get("text", "").strip()
|
||
if not text:
|
||
return Response(
|
||
{"text": ["Это поле не может быть пустым."]},
|
||
status=status.HTTP_400_BAD_REQUEST,
|
||
)
|
||
|
||
message = comments_helpers.create_comment(
|
||
obj=equipment,
|
||
author=request.user,
|
||
content=text,
|
||
)
|
||
author_ser = EmployeeBriefSerializer(message.author)
|
||
return Response(
|
||
{
|
||
"id": message.id,
|
||
"equipment": equipment.id,
|
||
"author": author_ser.data,
|
||
"text": message.content,
|
||
"created_at": message.created_at,
|
||
},
|
||
status=status.HTTP_201_CREATED,
|
||
)
|
||
|
||
@extend_schema(
|
||
parameters=[
|
||
OpenApiParameter(
|
||
name="comment_id",
|
||
type=int,
|
||
location=OpenApiParameter.PATH,
|
||
description="ID комментария в чате комментариев оборудования.",
|
||
)
|
||
]
|
||
)
|
||
@action(
|
||
detail=True,
|
||
methods=["delete"],
|
||
url_path="comments/(?P<comment_id>[^/.]+)",
|
||
)
|
||
def delete_comment(self, request, pk=None, comment_id=None):
|
||
"""Удаление комментария к оборудованию."""
|
||
equipment = self.get_object()
|
||
chat = comments_helpers.get_or_create_comments_chat(equipment)
|
||
if isinstance(chat, tuple):
|
||
chat = chat[0]
|
||
|
||
try:
|
||
message = Message.objects.get(id=comment_id, chat=chat)
|
||
except Message.DoesNotExist:
|
||
return Response(
|
||
{"detail": "Комментарий не найден"},
|
||
status=status.HTTP_404_NOT_FOUND,
|
||
)
|
||
|
||
if not (request.user.is_staff or message.author == request.user):
|
||
return Response(
|
||
{"detail": "Нет прав на удаление"},
|
||
status=status.HTTP_403_FORBIDDEN,
|
||
)
|
||
|
||
comments_helpers.delete_comment(
|
||
message=message,
|
||
deleted_by=request.user,
|
||
soft_delete=True,
|
||
)
|
||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||
|
||
|
||
class MaintenanceRecordViewSet(viewsets.ModelViewSet):
|
||
"""ViewSet для записей обслуживания."""
|
||
|
||
queryset = MaintenanceRecord.objects.select_related("equipment", "performed_by")
|
||
serializer_class = MaintenanceRecordSerializer
|
||
filter_backends = [
|
||
DjangoFilterBackend,
|
||
filters.OrderingFilter,
|
||
]
|
||
filterset_fields = ["equipment", "type", "performed_by"]
|
||
ordering_fields = ["date"]
|
||
ordering = ["-date"]
|
||
|
||
def perform_create(self, serializer):
|
||
"""При создании записи устанавливаем performed_by."""
|
||
serializer.save(performed_by=self.request.user)
|