"""Read representation for documents.""" from django.urls import reverse from drf_spectacular.utils import extend_schema_field from rest_framework import serializers from documents.models import ( Document, DocumentAcknowledgement, DocumentDisplayState, ) from ...employees.serializers import EmployeeBriefSerializer from .common import DepartmentBriefSerializer, FolderBriefSerializer from .tags import DocumentTagSerializer class DocumentReadSerializer(serializers.ModelSerializer): """Сериализатор для чтения документа. Включает список получателей и флаг, ознакомился ли текущий пользователь. """ created_by = EmployeeBriefSerializer(read_only=True) uploaded_by = EmployeeBriefSerializer(read_only=True) modified_by = EmployeeBriefSerializer(read_only=True) recipients = EmployeeBriefSerializer(many=True, read_only=True) departments = DepartmentBriefSerializer(many=True, read_only=True) acknowledgement_recipients = EmployeeBriefSerializer(many=True, read_only=True) acknowledgement_departments = DepartmentBriefSerializer(many=True, read_only=True) folder = FolderBriefSerializer(read_only=True) folder_path = serializers.CharField(read_only=True) tags = DocumentTagSerializer(many=True, read_only=True) created_at = serializers.DateTimeField(read_only=True) file_url = serializers.SerializerMethodField() file_name = serializers.CharField( source="file.name", read_only=True, allow_null=True ) file_size = serializers.IntegerField( source="file.size", read_only=True, allow_null=True ) is_acknowledged = serializers.SerializerMethodField() is_maximally_hidden = serializers.SerializerMethodField() acknowledgement_required_for_user = serializers.SerializerMethodField() acknowledged_count = serializers.SerializerMethodField() acknowledgement_total = serializers.SerializerMethodField() comments_count = serializers.SerializerMethodField() linked_tasks = serializers.SerializerMethodField() class Meta: model = Document fields = ( "id", "title", "description", "extracted_text", # Для полнотекстового поиска "folder", # Папка документа "folder_path", # Полный путь папки "tags", # Теги "created_by", # Создатель документа "created_at", # Дата создания "uploaded_by", "uploaded_at", "modified_by", "modified_at", "sent_to_all", "is_regulation", "acknowledgement_required", # Требуется ли ознакомление "acknowledgement_for_all", "acknowledgement_required_for_user", "acknowledged_count", "acknowledgement_total", "comments_count", "departments", "recipients", "acknowledgement_departments", "acknowledgement_recipients", "file_url", "file_name", "file_size", "is_acknowledged", "is_maximally_hidden", "linked_tasks", ) def get_file_url(self, obj: Document) -> str | None: if not obj.file_id: return None url = reverse("api:v1:documents-download", kwargs={"pk": obj.pk}) # Replacing an attachment changes the URL so open previews reload it. url = f"{url}?version={obj.file_id}" request = self.context.get("request") return request.build_absolute_uri(url) if request else url def get_is_acknowledged(self, obj: Document) -> bool: """Возвращает флаг ознакомления текущего пользователя. Args: obj (Document): Документ. Returns: bool: True, если текущий пользователь уже ознакомился. """ annotated = getattr(obj, "_is_acknowledged", None) if annotated is not None: return bool(annotated) request = self.context.get("request") if not request or not request.user.is_authenticated: return False return DocumentAcknowledgement.objects.filter( document=obj, user=request.user ).exists() def get_is_maximally_hidden(self, obj: Document) -> bool: """Максимально ли скрыта карточка для текущего пользователя.""" annotated = getattr(obj, "_is_maximally_hidden", None) if annotated is not None: return bool(annotated) request = self.context.get("request") if not request or not request.user.is_authenticated: return False return DocumentDisplayState.objects.filter( document=obj, user=request.user, is_maximally_hidden=True, ).exists() def get_acknowledgement_required_for_user(self, obj: Document) -> bool: """Нужно ли текущему сотруднику подтверждать ознакомление.""" request = self.context.get("request") if not request or not request.user.is_authenticated: return False audience_ids = getattr(obj, "_acknowledgement_user_ids", None) if audience_ids is not None: return request.user.pk in audience_ids from documents.audience import user_requires_document_acknowledgement return user_requires_document_acknowledgement(obj, request.user) def get_acknowledged_count(self, obj: Document) -> int: """Количество сотрудников из аудитории, уже ознакомившихся.""" attached = getattr(obj, "_acknowledged_count", None) if attached is not None: return int(attached) from documents.audience import document_acknowledgement_audience audience = document_acknowledgement_audience(obj) return obj.acknowledgements.filter(user__in=audience).count() def get_acknowledgement_total(self, obj: Document) -> int: """Размер аудитории обязательного ознакомления.""" attached = getattr(obj, "_acknowledgement_total", None) if attached is not None: return int(attached) from documents.audience import document_acknowledgement_audience return document_acknowledgement_audience(obj).count() def get_comments_count(self, obj: Document) -> int: """Количество неудалённых комментариев документа.""" attached = getattr(obj, "_comments_count", None) if attached is not None: return int(attached) from communications.comments_helpers import get_comment_count return get_comment_count(obj) @extend_schema_field(serializers.ListField(child=serializers.DictField())) def get_linked_tasks(self, obj: Document) -> list[dict]: prefetched = getattr(obj, "_linked_task_payloads", None) if prefetched is not None: return prefetched request = self.context.get("request") user = getattr(request, "user", None) if not user or not getattr(user, "is_authenticated", False): return [] from tasks.models import TaskLinkedObjectKind from tasks.presentation import linked_task_badges_for_object return linked_task_badges_for_object( obj=obj, kind=TaskLinkedObjectKind.DOCUMENT, user=user, )