# backend/api/v1/search/views.py """ API views для глобального поиска через django-watson. """ from __future__ import annotations from typing import Any, Dict from django.contrib.contenttypes.models import ContentType from drf_spectacular.utils import OpenApiParameter, extend_schema from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request as DRFRequest from rest_framework.response import Response from api.v1.documents.permissions import user_can_view_document from api.v1.procurement.permissions import user_can_view_procurement_request from communications.utils import user_can_access_chat from scheduling.services import user_can_view_event from search.backends import search_with_abbreviations from tasks.access import user_can_access_task_board from .serializers import SearchResponseSerializer def _file_url(value: Any) -> str: """Безопасно возвращает относительный URL файла, если он существует.""" if not value: return "" try: return value.url except (AttributeError, ValueError): return "" def _user_avatar(value: Any) -> str: return _file_url(getattr(value, "avatar", None)) if value else "" def _get_model_name(obj: Any) -> str: """Определяет тип модели для результата поиска.""" model = type(obj) model_name = model.__name__.lower() mapping = { "post": "post", "employee": "employee", "department": "department", "request": "request", "chat": "chat", "message": "message", "calendarevent": "event", "event": "schedule_event", # django-scheduler Event "procurementrequest": "procurement_request", "equipment": "equipment", "document": "document", "notification": "notification", "taskboard": "task_board", "task": "task", "product": "product", "guest": "guest", } return mapping.get(model_name, model_name) def _search_queryset_for_model(model: Any): """Подгружает связи, нужные форматтерам и проверкам доступа.""" queryset = model._default_manager.all() model_name = model.__name__.lower() if model_name == "employee": queryset = queryset.visible_in_ui() select_related = { "employee": ("position",), "post": ("author",), "request": ("employee",), "message": ("author", "chat"), "procurementrequest": ("requestor", "department"), "document": ("uploaded_by",), "task": ("board", "assignee"), }.get(model_name, ()) if select_related: queryset = queryset.select_related(*select_related) if model_name == "product": queryset = queryset.prefetch_related("images") return queryset def _materialize_search_objects(search_results: Any) -> list[Any]: """Загружает полиморфные результаты пакетами, сохраняя порядок Watson.""" entries = list(search_results.values_list("content_type_id", "object_id")) content_types = ContentType.objects.in_bulk({entry[0] for entry in entries}) object_maps: Dict[int, Dict[str, Any]] = {} for content_type_id, content_type in content_types.items(): model = content_type.model_class() if model is None: object_maps[content_type_id] = {} continue ids = [ object_id for entry_type_id, object_id in entries if entry_type_id == content_type_id ] objects = _search_queryset_for_model(model).in_bulk(ids) object_maps[content_type_id] = { str(object_id): objects.get(model._meta.pk.to_python(object_id)) for object_id in ids } return [ object_maps.get(content_type_id, {}).get(str(object_id)) for content_type_id, object_id in entries ] def _format_employee(obj: Any) -> Dict[str, Any]: """Форматирует сотрудника для API.""" return { "model_name": "employee", "object_id": obj.pk, "title": f"{obj.last_name} {obj.first_name} {obj.patronymic or ''}".strip(), "description": getattr(obj.position, "name", "") if hasattr(obj, "position") else "", "image": _user_avatar(obj), "meta": { "email": obj.email, "phone": str(obj.phone_number) if obj.phone_number else "", }, } def _format_department(obj: Any) -> Dict[str, Any]: """Форматирует отдел для API.""" return { "model_name": "department", "object_id": obj.pk, "title": obj.name, "description": obj.description or "", "meta": {"head": str(obj.head) if hasattr(obj, "head") and obj.head else None}, } def _format_post(obj: Any) -> Dict[str, Any]: """Форматирует пост для API.""" return { "model_name": "post", "object_id": obj.pk, "title": obj.title, "description": obj.body[:200] + "..." if len(obj.body) > 200 else obj.body, "image": _user_avatar(getattr(obj, "author", None)), "meta": { "author": str(obj.author) if hasattr(obj, "author") else None, "created_at": obj.created_at.isoformat() if hasattr(obj, "created_at") else None, }, } def _format_request(obj: Any) -> Dict[str, Any]: """Форматирует заявление для API.""" return { "model_name": "request", "object_id": obj.pk, "title": obj.title or f"{obj.get_type_display()} - {obj.employee}", "description": obj.comment[:200] + "..." if obj.comment and len(obj.comment) > 200 else (obj.comment or ""), "image": _user_avatar(getattr(obj, "employee", None)), "meta": { "status": obj.status, "type": obj.type, "employee": str(obj.employee) if hasattr(obj, "employee") else None, }, } def _format_chat(obj: Any) -> Dict[str, Any]: """Форматирует чат для API.""" return { "model_name": "chat", "object_id": obj.pk, "title": obj.name or f"Чат #{obj.pk}", "description": obj.description or "", "image": _file_url(getattr(obj, "avatar", None)), "meta": { "type": obj.type if hasattr(obj, "type") else None, }, } def _format_message(obj: Any) -> Dict[str, Any]: """Форматирует сообщение для API.""" return { "model_name": "message", "object_id": obj.pk, "title": f"Сообщение от {obj.author}" if hasattr(obj, "author") else f"Сообщение #{obj.pk}", "description": obj.content[:200] + "..." if len(obj.content) > 200 else obj.content, "image": _user_avatar(getattr(obj, "author", None)), "meta": { "author": str(obj.author) if hasattr(obj, "author") else None, "chat": str(obj.chat) if hasattr(obj, "chat") else None, "chat_id": obj.chat.pk if hasattr(obj, "chat") else None, }, } def _format_event(obj: Any) -> Dict[str, Any]: """Форматирует календарное событие (CalendarEvent) для API.""" return { "model_name": "event", "object_id": obj.pk, "title": obj.title, "description": obj.description or "", "meta": { "start_date": obj.start_date.isoformat() if hasattr(obj, "start_date") else None, }, } def _format_schedule_event(obj: Any) -> Dict[str, Any]: """Форматирует событие расписания (django-scheduler Event) для API.""" return { "model_name": "schedule_event", "object_id": obj.pk, "title": obj.title, "description": obj.description or "", "meta": { "start": obj.start.isoformat() if hasattr(obj, "start") else None, "end": obj.end.isoformat() if hasattr(obj, "end") and obj.end else None, }, } def _format_procurement_request(obj: Any) -> Dict[str, Any]: """Форматирует заявку на закупку для API.""" return { "model_name": "procurement_request", "object_id": obj.pk, "title": obj.title, "description": obj.description[:200] + "..." if len(obj.description) > 200 else obj.description, "image": _user_avatar(getattr(obj, "requestor", None)), "meta": { "status": obj.status if hasattr(obj, "status") else None, "requestor": str(obj.requestor) if hasattr(obj, "requestor") else None, "department": str(obj.department) if hasattr(obj, "department") else None, }, } def _format_equipment(obj: Any) -> Dict[str, Any]: """Форматирует оборудование для API.""" return { "model_name": "equipment", "object_id": obj.pk, "title": obj.name, "description": ( f"Инв. №: {obj.inventory_number}, Сер. №: {obj.serial_number}" if obj.serial_number else f"Инв. №: {obj.inventory_number}" ), "meta": { "inventory_number": obj.inventory_number, "status": obj.status if hasattr(obj, "status") else None, "location": obj.location if hasattr(obj, "location") else None, }, } def _format_document(obj: Any) -> Dict[str, Any]: """Форматирует документ для API.""" return { "model_name": "document", "object_id": obj.pk, "title": obj.title, "description": ( obj.description[:200] + "..." if obj.description and len(obj.description) > 200 else (obj.description or "") ), "image": _user_avatar(getattr(obj, "uploaded_by", None)), "meta": { "uploaded_by": str(obj.uploaded_by) if hasattr(obj, "uploaded_by") and obj.uploaded_by else None, "uploaded_at": obj.uploaded_at.isoformat() if hasattr(obj, "uploaded_at") else None, }, } def _get_notification_search_title(obj: Any) -> str: """Строит заголовок уведомления для поиска без legacy-зависимостей.""" data = getattr(obj, "data", None) or {} if isinstance(data, dict) and data.get("title"): return str(data["title"]) verb = getattr(obj, "verb", "") or "notification" return str(verb).replace("_", " ").title() def _get_notification_search_message(obj: Any) -> str: """Возвращает текст уведомления для результатов поиска.""" return getattr(obj, "description", "") or "" def _format_notification(obj: Any) -> Dict[str, Any]: """Форматирует уведомление для API.""" message = _get_notification_search_message(obj) return { "model_name": "notification", "object_id": obj.pk, "title": _get_notification_search_title(obj), "description": message[:150] + "..." if len(message) > 150 else message, "image": _user_avatar(getattr(obj, "actor", None)), "meta": { "is_read": not getattr(obj, "unread", True), "created_at": obj.timestamp.isoformat() if hasattr(obj, "timestamp") and obj.timestamp else None, }, } def _format_task_board(obj: Any) -> Dict[str, Any]: return { "model_name": "task_board", "object_id": obj.pk, "title": obj.name, "description": obj.description or "", "image": _file_url(getattr(obj, "avatar", None)), "meta": { "access": obj.get_access_scope_display(), }, } def _format_task(obj: Any) -> Dict[str, Any]: return { "model_name": "task", "object_id": obj.pk, "title": obj.title, "description": obj.description[:200] + "..." if len(obj.description) > 200 else obj.description, "image": _user_avatar(getattr(obj, "assignee", None)), "meta": { "board": str(obj.board), "board_id": obj.board_id, "priority": obj.get_priority_display(), }, } def _format_product(obj: Any) -> Dict[str, Any]: images = list(obj.images.all()) image = next((item for item in images if item.is_primary), None) image = image or (images[0] if images else None) return { "model_name": "product", "object_id": str(obj.pk), "title": obj.name, "description": obj.description[:200] + "..." if len(obj.description) > 200 else obj.description, "image": _file_url(getattr(image, "thumbnail", None)) or _file_url(getattr(image, "file", None)), "meta": { "sku": obj.sku, "barcode": obj.barcode, "kind": obj.get_kind_display(), }, } def _format_guest(obj: Any) -> Dict[str, Any]: return { "model_name": "guest", "object_id": obj.pk, "title": obj.full_name, "description": " · ".join( value for value in [obj.organization, obj.position] if value ), "image": _file_url(getattr(obj, "avatar", None)), "meta": { "email": obj.email, "phone": obj.phone, }, } def _format_result(obj: Any, model_name: str) -> Dict[str, Any]: """Форматирует результат поиска в зависимости от типа модели.""" formatters = { "employee": _format_employee, "department": _format_department, "post": _format_post, "request": _format_request, "chat": _format_chat, "message": _format_message, "event": _format_event, "schedule_event": _format_schedule_event, "procurement_request": _format_procurement_request, "equipment": _format_equipment, "document": _format_document, "notification": _format_notification, "task_board": _format_task_board, "task": _format_task, "product": _format_product, "guest": _format_guest, } formatter = formatters.get(model_name) if formatter: return formatter(obj) # Фоллбэк для неизвестных типов return { "model_name": model_name, "object_id": obj.pk, "title": str(obj), "description": "", "image": "", "meta": {}, } def _is_hr(user: Any) -> bool: """Проверяет права на просмотр всех заявлений.""" return user.has_perm("requests_app.can_view_all_requests") or user.has_perm( "requests_app.can_process_requests" ) def _can_view_search_result( user: Any, obj: Any, model_name: str, access_cache: Dict[tuple[str, int], bool] | None = None, ) -> bool: """Не позволить глобальному индексу обходить права доменных API.""" if model_name == "request": return _is_hr(user) or getattr(obj, "employee_id", None) == user.id if model_name == "notification": return getattr(obj, "recipient_id", None) == user.id if model_name == "chat": cache_key = ("chat", obj.pk) if access_cache is not None and cache_key in access_cache: return access_cache[cache_key] allowed = user_can_access_chat(obj, user) if access_cache is not None: access_cache[cache_key] = allowed return allowed if model_name == "message": if getattr(obj, "is_deleted", False): return False chat = getattr(obj, "chat", None) cache_key = ("chat", getattr(obj, "chat_id", 0)) if access_cache is not None and cache_key in access_cache: return access_cache[cache_key] allowed = user_can_access_chat(chat, user) if access_cache is not None: access_cache[cache_key] = allowed return allowed if model_name == "document": return user_can_view_document(user, obj) if model_name == "schedule_event": return user_can_view_event(user, obj) if model_name == "procurement_request": return user_can_view_procurement_request(user, obj) if model_name == "task_board": return not getattr(obj, "is_archived", False) and user_can_access_task_board( user, obj ) if model_name == "task": board = getattr(obj, "board", None) return not getattr(board, "is_archived", False) and user_can_access_task_board( user, board ) return True @extend_schema( tags=["Search"], summary="Глобальный поиск по системе", parameters=[ OpenApiParameter("q", str, OpenApiParameter.QUERY, required=True), OpenApiParameter("limit", int, OpenApiParameter.QUERY), OpenApiParameter("offset", int, OpenApiParameter.QUERY), OpenApiParameter("type", str, OpenApiParameter.QUERY), ], responses=SearchResponseSerializer, ) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def search_api_view(request: DRFRequest) -> Response: """ API для глобального поиска через django-watson. Query Parameters: - q (str): Поисковый запрос - limit (int): Максимальное количество результатов на тип (default: 10) - offset (int): Смещение внутри каждого типа результатов (default: 0) - type (str): Вернуть записи только указанного типа Returns: { "query": "поисковый запрос", "results": [ { "model_name": ( "post|employee|department|request|chat|message|event|" "procurement_request|equipment|document|notification" ), "object_id": 123, "title": "Заголовок", "description": "Описание", "meta": {...} }, ... ], "counts": { "post": 5, "employee": 3, ... }, "hidden_counts": { "request": 2, ... }, "total": 15, "hidden_total": 2 } """ raw_q = request.query_params.get("q", "") or "" query = raw_q.strip() try: limit = int(request.query_params.get("limit", 10)) except (ValueError, TypeError): limit = 10 limit = max(1, min(limit, 50)) try: offset = int(request.query_params.get("offset", 0)) except (ValueError, TypeError): offset = 0 offset = max(0, offset) requested_model_type = (request.query_params.get("type", "") or "").strip() if not query: return Response( { "query": "", "results": [], "counts": {}, "hidden_counts": {}, "total": 0, "hidden_total": 0, }, status=status.HTTP_200_OK, ) # Проверяем весь набор совпадений: раннее обрезание общего индекса приводило # к тому, что многочисленные недоступные записи вытесняли доступные из выдачи. # Модуль товаров отложен; индекс сохраняем для будущего интерфейса, # но не предлагаем пользователям переходы на скрытые страницы. search_results = search_with_abbreviations(query).exclude( content_type__app_label="commerce", content_type__model="product" ) search_objects = _materialize_search_objects(search_results) # Группируем и форматируем результаты results = [] model_counts: Dict[str, int] = {} hidden_counts: Dict[str, int] = {} access_cache: Dict[tuple[str, int], bool] = {} for obj in search_objects: # Пропускаем удаленные объекты if obj is None: continue model_name = _get_model_name(obj) if not _can_view_search_result(request.user, obj, model_name, access_cache): hidden_counts[model_name] = hidden_counts.get(model_name, 0) + 1 continue # Подсчитываем model_counts[model_name] = model_counts.get(model_name, 0) + 1 # Возвращаем нужную страницу конкретного типа. Счётчики при этом # продолжают описывать весь доступный пользователю набор результатов. result_position = model_counts[model_name] type_matches = not requested_model_type or model_name == requested_model_type if type_matches and offset < result_position <= offset + limit: formatted = _format_result(obj, model_name) results.append(formatted) # Готовим ответ response_data = { "query": query, "results": results, "counts": model_counts, "hidden_counts": hidden_counts, "total": sum(model_counts.values()), "hidden_total": sum(hidden_counts.values()), } # Валидируем через сериализатор (опционально, для проверки структуры) serializer = SearchResponseSerializer(data=response_data) if serializer.is_valid(): return Response(serializer.validated_data, status=status.HTTP_200_OK) # Если валидация не прошла, все равно возвращаем результат (для отладки) return Response(response_data, status=status.HTTP_200_OK)