630 lines
22 KiB
Python
630 lines
22 KiB
Python
# 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)
|