лишнее из документов удалить
This commit is contained in:
lizerk
parent
cc24a073f4
commit
0f8dcff4ad
@ -1,573 +0,0 @@
|
||||
"""
|
||||
Signals для автоматической генерации уведомлений в модуле Documents.
|
||||
|
||||
Обрабатывает события:
|
||||
- Новый документ на ознакомление
|
||||
- Все сотрудники ознакомились
|
||||
- Напоминание о необходимости ознакомления (через Celery task)
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
from django.db.models.signals import post_save, m2m_changed
|
||||
from django.dispatch import receiver
|
||||
from django.contrib.auth import get_user_model
|
||||
|
||||
from .models import Document, DocumentAcknowledgement, DocumentComment
|
||||
from notifications.signals import notify
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
Employee = get_user_model()
|
||||
|
||||
|
||||
@receiver(post_save, sender=Document)
|
||||
def create_document_notification(sender, instance, created, **kwargs):
|
||||
"""
|
||||
Создает уведомления при создании нового документа асинхронно.
|
||||
|
||||
Уведомления отправляются:
|
||||
- Всем активным сотрудникам (если sent_to_all=True)
|
||||
- Сотрудникам выбранных отделов (через m2m_changed signal)
|
||||
- Выбранным получателям (через m2m_changed signal)
|
||||
"""
|
||||
logger.info(
|
||||
f"[notification_signals] post_save Document id={instance.pk} "
|
||||
f"created={created} sent_to_all={instance.sent_to_all}"
|
||||
)
|
||||
|
||||
if not created:
|
||||
return
|
||||
|
||||
document = instance
|
||||
|
||||
# Если документ отправляется всем - создаем уведомления напрямую
|
||||
# channels.py автоматически отправит через Celery
|
||||
if document.sent_to_all:
|
||||
logger.info(
|
||||
f"[notification_signals] Creating notifications for doc={document.pk} (sent_to_all=True)"
|
||||
)
|
||||
notify_all_employees(document)
|
||||
else:
|
||||
logger.info(
|
||||
f"[notification_signals] Skipping notifications (sent_to_all=False), "
|
||||
f"waiting for m2m_changed"
|
||||
)
|
||||
|
||||
|
||||
@receiver(m2m_changed, sender=Document.recipients.through)
|
||||
def notify_specific_recipients(sender, instance, action, pk_set, **kwargs):
|
||||
"""
|
||||
Уведомляет конкретных получателей при добавлении в список.
|
||||
Использует быструю обработку для большого количества получателей.
|
||||
"""
|
||||
logger.info(
|
||||
f"[notification_signals] m2m_changed Document id={instance.pk} "
|
||||
f"action={action} pk_set={pk_set} sent_to_all={instance.sent_to_all}"
|
||||
)
|
||||
|
||||
if action != 'post_add':
|
||||
logger.info(
|
||||
f"[notification_signals] Skipping (action={action} != 'post_add')"
|
||||
)
|
||||
return
|
||||
|
||||
document = instance
|
||||
|
||||
# Не отправляем индивидуальные уведомления если документ для всех
|
||||
if document.sent_to_all:
|
||||
logger.info(
|
||||
f"[notification_signals] Skipping (sent_to_all=True)"
|
||||
)
|
||||
return
|
||||
|
||||
recipient_count = len(pk_set or [])
|
||||
logger.info(
|
||||
f"[notification_signals] Processing {recipient_count} recipients"
|
||||
)
|
||||
|
||||
# Определяем, это массовая рассылка или нет
|
||||
from django.conf import settings
|
||||
BULK_THRESHOLD = getattr(settings, 'NOTIFICATION_BULK_THRESHOLD', 10)
|
||||
is_bulk = recipient_count >= BULK_THRESHOLD
|
||||
|
||||
if is_bulk:
|
||||
logger.info(
|
||||
f"[notification_signals] Bulk mode activated "
|
||||
f"({recipient_count} >= {BULK_THRESHOLD})"
|
||||
)
|
||||
|
||||
# Уведомляем новых получателей
|
||||
created_count = 0
|
||||
for user_id in pk_set:
|
||||
try:
|
||||
user = Employee.objects.get(id=user_id)
|
||||
logger.debug(
|
||||
f"[notification_signals] Creating notification for user={user_id}"
|
||||
)
|
||||
uploader_name = (
|
||||
document.uploaded_by.get_full_name()
|
||||
if document.uploaded_by
|
||||
else 'Администратор'
|
||||
)
|
||||
notify.send(
|
||||
sender=document.uploaded_by,
|
||||
recipient=user,
|
||||
verb='document_ready',
|
||||
action_object=document,
|
||||
description=(
|
||||
f'{uploader_name} загрузил документ "{document.title}". '
|
||||
f'Требуется ознакомление.'
|
||||
),
|
||||
action_url='/documents',
|
||||
data={
|
||||
'title': 'Новый документ на ознакомление',
|
||||
'document_id': document.id,
|
||||
'uploaded_by_id': (
|
||||
document.uploaded_by.id if document.uploaded_by else None
|
||||
),
|
||||
'sent_to_all': document.sent_to_all,
|
||||
},
|
||||
)
|
||||
created_count += 1
|
||||
except Employee.DoesNotExist:
|
||||
logger.warning(f"[notification_signals] User {user_id} not found, skipping")
|
||||
except Exception as e:
|
||||
logger.error(f"[notification_signals] Error creating notification for user {user_id}: {e}")
|
||||
|
||||
logger.info(f"[notification_signals] Created {created_count} notifications")
|
||||
|
||||
|
||||
@receiver(m2m_changed, sender=Document.departments.through)
|
||||
def notify_department_employees(sender, instance, action, pk_set, **kwargs):
|
||||
"""
|
||||
Уведомляет всех сотрудников выбранных отделов при добавлении отделов.
|
||||
Уведомления получат ВСЕ текущие и будущие сотрудники этих отделов.
|
||||
"""
|
||||
logger.info(
|
||||
f"[notification_signals] m2m_changed Document.departments "
|
||||
f"id={instance.pk} action={action} pk_set={pk_set}"
|
||||
)
|
||||
|
||||
if action != 'post_add':
|
||||
logger.info(
|
||||
f"[notification_signals] Skipping (action={action} != 'post_add')"
|
||||
)
|
||||
return
|
||||
|
||||
document = instance
|
||||
|
||||
# Не отправляем если документ для всех
|
||||
if document.sent_to_all:
|
||||
logger.info(
|
||||
"[notification_signals] Skipping (sent_to_all=True)"
|
||||
)
|
||||
return
|
||||
|
||||
if not pk_set:
|
||||
logger.warning("[notification_signals] No departments provided")
|
||||
return
|
||||
|
||||
# Получаем всех активных сотрудников выбранных отделов
|
||||
from employees.models import Department
|
||||
|
||||
all_employees = set()
|
||||
for dept_id in pk_set:
|
||||
try:
|
||||
department = Department.objects.get(id=dept_id)
|
||||
dept_employees = department.active_employees
|
||||
logger.info(
|
||||
f"[notification_signals] Department '{department.name}' "
|
||||
f"has {len(dept_employees)} active employees: "
|
||||
f"{[e.id for e in dept_employees]}"
|
||||
)
|
||||
all_employees.update(dept_employees)
|
||||
logger.info(
|
||||
f"[notification_signals] After update, all_employees has: "
|
||||
f"{[e.id for e in all_employees]}"
|
||||
)
|
||||
except Department.DoesNotExist:
|
||||
logger.warning(
|
||||
f"[notification_signals] Department {dept_id} not found"
|
||||
)
|
||||
continue
|
||||
|
||||
# Исключаем загрузившего
|
||||
if document.uploaded_by and document.uploaded_by in all_employees:
|
||||
all_employees.remove(document.uploaded_by)
|
||||
|
||||
recipient_count = len(all_employees)
|
||||
logger.info(
|
||||
f"[notification_signals] Processing {recipient_count} employees "
|
||||
f"from {len(pk_set)} departments"
|
||||
)
|
||||
|
||||
# Определяем режим рассылки
|
||||
from django.conf import settings
|
||||
BULK_THRESHOLD = getattr(settings, 'NOTIFICATION_BULK_THRESHOLD', 10)
|
||||
is_bulk = recipient_count >= BULK_THRESHOLD
|
||||
|
||||
if is_bulk:
|
||||
logger.info(
|
||||
f"[notification_signals] Bulk mode activated "
|
||||
f"({recipient_count} >= {BULK_THRESHOLD})"
|
||||
)
|
||||
|
||||
# Создаём уведомления
|
||||
created_count = 0
|
||||
uploader_name = (
|
||||
document.uploaded_by.get_full_name()
|
||||
if document.uploaded_by
|
||||
else 'Администратор'
|
||||
)
|
||||
for employee in all_employees:
|
||||
try:
|
||||
notify.send(
|
||||
sender=document.uploaded_by,
|
||||
recipient=employee,
|
||||
verb='document_ready',
|
||||
action_object=document,
|
||||
description=(
|
||||
f'{uploader_name} загрузил документ "{document.title}". '
|
||||
f'Требуется ознакомление.'
|
||||
),
|
||||
action_url='/documents',
|
||||
data={
|
||||
'title': 'Новый документ на ознакомление',
|
||||
'document_id': document.id,
|
||||
'uploaded_by_id': (
|
||||
document.uploaded_by.id if document.uploaded_by else None
|
||||
),
|
||||
'sent_to_all': document.sent_to_all,
|
||||
},
|
||||
)
|
||||
created_count += 1
|
||||
except Exception as e:
|
||||
logger.error(f"[notification_signals] Error creating notification for employee {employee.id}: {e}")
|
||||
|
||||
logger.info(f"[notification_signals] Created {created_count} notifications")
|
||||
|
||||
|
||||
@receiver(post_save, sender=DocumentAcknowledgement)
|
||||
def check_all_acknowledged(sender, instance, created, **kwargs):
|
||||
"""
|
||||
Проверяет, все ли получатели ознакомились с документом.
|
||||
Если да - уведомляет загрузившего.
|
||||
"""
|
||||
if not created:
|
||||
return
|
||||
|
||||
acknowledgement = instance
|
||||
document = acknowledgement.document
|
||||
|
||||
# Определяем общее количество получателей
|
||||
if document.sent_to_all:
|
||||
total_recipients = Employee.objects.filter(is_active=True).count()
|
||||
else:
|
||||
# Считаем получателей из recipients + departments
|
||||
recipients_count = document.recipients.count()
|
||||
|
||||
# Добавляем сотрудников из отделов
|
||||
from employees.models import Department
|
||||
department_employees = set()
|
||||
for department in document.departments.all():
|
||||
department_employees.update(department.active_employees)
|
||||
|
||||
# Исключаем дубликаты с recipients
|
||||
total_recipients = recipients_count + len(department_employees)
|
||||
|
||||
# Проверяем количество ознакомившихся
|
||||
acknowledged_count = document.acknowledgements.count()
|
||||
|
||||
# Если все ознакомились - уведомляем загрузившего
|
||||
if (acknowledged_count >= total_recipients and
|
||||
total_recipients > 0 and
|
||||
document.uploaded_by):
|
||||
|
||||
notify.send(
|
||||
sender=None,
|
||||
recipient=document.uploaded_by,
|
||||
verb='document_signed_all',
|
||||
action_object=document,
|
||||
description=(
|
||||
f'Все сотрудники ознакомились с документом '
|
||||
f'"{document.title}"'
|
||||
),
|
||||
action_url='/documents',
|
||||
data={
|
||||
'title': 'Все ознакомились с документом',
|
||||
'document_id': document.id,
|
||||
'total_recipients': total_recipients,
|
||||
'acknowledged_count': acknowledged_count,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ===== Вспомогательные функции =====
|
||||
|
||||
def notify_all_employees(document):
|
||||
"""
|
||||
Отправляет уведомление о новом документе всем активным сотрудникам.
|
||||
Веб-уведомления отправляются сразу, email/Telegram - асинхронно.
|
||||
"""
|
||||
logger.info(
|
||||
f"[notification_signals] notify_all_employees doc={document.id}"
|
||||
)
|
||||
|
||||
active_employees = Employee.objects.filter(is_active=True)
|
||||
|
||||
# Исключаем загрузившего документ
|
||||
if document.uploaded_by:
|
||||
active_employees = active_employees.exclude(
|
||||
id=document.uploaded_by.id
|
||||
)
|
||||
|
||||
count = active_employees.count()
|
||||
logger.info(
|
||||
f"[notification_signals] Creating {count} notifications "
|
||||
f"(web immediately, email/telegram deferred)"
|
||||
)
|
||||
|
||||
# Создаём уведомления с ЧАСТИЧНОЙ отправкой
|
||||
created_count = 0
|
||||
uploader_name = (
|
||||
document.uploaded_by.get_full_name()
|
||||
if document.uploaded_by
|
||||
else 'Администратор'
|
||||
)
|
||||
for employee in active_employees:
|
||||
try:
|
||||
notify.send(
|
||||
sender=document.uploaded_by,
|
||||
recipient=employee,
|
||||
verb='document_ready',
|
||||
action_object=document,
|
||||
description=(
|
||||
f'{uploader_name} загрузил документ "{document.title}". '
|
||||
f'Требуется ознакомление.'
|
||||
),
|
||||
action_url='/documents',
|
||||
data={
|
||||
'title': 'Новый документ на ознакомление',
|
||||
'document_id': document.id,
|
||||
'uploaded_by_id': (
|
||||
document.uploaded_by.id if document.uploaded_by else None
|
||||
),
|
||||
'sent_to_all': document.sent_to_all,
|
||||
},
|
||||
)
|
||||
created_count += 1
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error creating notification for user {employee.id}: {e}"
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"[notification_signals] Created {created_count}/{count} notifications "
|
||||
f"for doc={document.id}. Web sent immediately, email/telegram pending."
|
||||
)
|
||||
|
||||
# Запускаем асинхронную отправку email/telegram
|
||||
try:
|
||||
from django.core.management import call_command
|
||||
import threading
|
||||
|
||||
def send_pending():
|
||||
try:
|
||||
call_command('send_pending_notifications', '--batch-size=100')
|
||||
except Exception as e:
|
||||
logger.error(f"Error in background send: {e}")
|
||||
|
||||
# Запускаем в отдельном потоке
|
||||
thread = threading.Thread(target=send_pending, daemon=True)
|
||||
thread.start()
|
||||
logger.info("[notification_signals] Started background email/telegram send")
|
||||
except Exception as e:
|
||||
logger.error(f"Error starting background send: {e}")
|
||||
def create_document_ready_notification(document, recipient, send_immediately=True):
|
||||
"""
|
||||
Создает уведомление о новом документе для конкретного получателя.
|
||||
"""
|
||||
logger.info(
|
||||
f"[notification_signals] create_document_ready_notification "
|
||||
f"doc={document.id} recipient={recipient.id}"
|
||||
)
|
||||
|
||||
uploader_name = (
|
||||
document.uploaded_by.get_full_name()
|
||||
if document.uploaded_by
|
||||
else 'Администратор'
|
||||
)
|
||||
|
||||
notify.send(
|
||||
sender=document.uploaded_by,
|
||||
recipient=recipient,
|
||||
verb='document_ready',
|
||||
action_object=document,
|
||||
description=(
|
||||
f'{uploader_name} загрузил документ "{document.title}". '
|
||||
f'Требуется ознакомление.'
|
||||
),
|
||||
action_url='/documents',
|
||||
data={
|
||||
'title': 'Новый документ на ознакомление',
|
||||
'document_id': document.id,
|
||||
'uploaded_by_id': (
|
||||
document.uploaded_by.id if document.uploaded_by else None
|
||||
),
|
||||
'sent_to_all': document.sent_to_all,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# НОВЫЕ СИГНАЛЫ ДЛЯ КОММЕНТАРИЕВ И СВЯЗАННЫХ ДОКУМЕНТОВ
|
||||
# =============================================================================
|
||||
|
||||
@receiver(post_save, sender=DocumentComment)
|
||||
def notify_on_new_comment(sender, instance, created, **kwargs):
|
||||
"""
|
||||
Отправка уведомлений при создании нового комментария.
|
||||
|
||||
- Если это ответ на комментарий (parent != None): уведомить автора родительского комментария
|
||||
- Если это обычный комментарий: уведомить автора документа (если он не автор комментария)
|
||||
"""
|
||||
if not created:
|
||||
return
|
||||
|
||||
comment = instance
|
||||
document = comment.document
|
||||
author = comment.author
|
||||
|
||||
logger.info(
|
||||
f"[notification_signals] New comment on document={document.pk} "
|
||||
f"by user={author.pk} parent={comment.parent_id}"
|
||||
)
|
||||
|
||||
# Если это ответ на комментарий
|
||||
if comment.parent:
|
||||
parent_author = comment.parent.author
|
||||
|
||||
# Не уведомляем самого себя
|
||||
if parent_author.id == author.id:
|
||||
logger.info(
|
||||
f"[notification_signals] Skipping self-reply notification"
|
||||
)
|
||||
return
|
||||
|
||||
logger.info(
|
||||
f"[notification_signals] Notifying parent author={parent_author.pk} "
|
||||
f"about reply"
|
||||
)
|
||||
|
||||
notify.send(
|
||||
sender=author,
|
||||
recipient=parent_author,
|
||||
verb='document_comment_reply',
|
||||
action_object=comment,
|
||||
description=(
|
||||
f'{author.get_full_name() or author.username} ответил на ваш '
|
||||
f'комментарий к документу "{document.title}"'
|
||||
),
|
||||
action_url='/documents',
|
||||
data={
|
||||
'title': 'Ответ на ваш комментарий',
|
||||
'document_id': document.id,
|
||||
'comment_id': comment.id,
|
||||
'parent_comment_id': comment.parent_id,
|
||||
'author_id': author.id,
|
||||
},
|
||||
)
|
||||
else:
|
||||
# Обычный комментарий - уведомляем автора документа
|
||||
if not document.uploaded_by:
|
||||
logger.info(
|
||||
f"[notification_signals] Document has no author, skipping"
|
||||
)
|
||||
return
|
||||
|
||||
doc_author = document.uploaded_by
|
||||
|
||||
# Не уведомляем самого себя
|
||||
if doc_author.id == author.id:
|
||||
logger.info(
|
||||
f"[notification_signals] Skipping self-comment notification"
|
||||
)
|
||||
return
|
||||
|
||||
logger.info(
|
||||
f"[notification_signals] Notifying document author={doc_author.pk} "
|
||||
f"about new comment"
|
||||
)
|
||||
|
||||
notify.send(
|
||||
sender=author,
|
||||
recipient=doc_author,
|
||||
verb='document_comment',
|
||||
action_object=comment,
|
||||
description=(
|
||||
f'{author.get_full_name() or author.username} оставил комментарий '
|
||||
f'к вашему документу "{document.title}"'
|
||||
),
|
||||
action_url='/documents',
|
||||
data={
|
||||
'title': 'Новый комментарий к документу',
|
||||
'document_id': document.id,
|
||||
'comment_id': comment.id,
|
||||
'author_id': author.id,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@receiver(m2m_changed, sender=Document.related_documents.through)
|
||||
def notify_on_related_document_added(sender, instance, action, pk_set, **kwargs):
|
||||
"""
|
||||
Уведомление при добавлении связанного документа.
|
||||
Уведомляем автора связанного документа (если это не тот же пользователь).
|
||||
"""
|
||||
if action != 'post_add':
|
||||
return
|
||||
|
||||
if not pk_set:
|
||||
return
|
||||
|
||||
main_document = instance
|
||||
|
||||
logger.info(
|
||||
f"[notification_signals] Related documents added to doc={main_document.pk} "
|
||||
f"related_ids={pk_set}"
|
||||
)
|
||||
|
||||
# Получаем связанные документы
|
||||
related_documents = Document.objects.filter(pk__in=pk_set).select_related('uploaded_by')
|
||||
|
||||
for related_doc in related_documents:
|
||||
# Если у связанного документа есть автор
|
||||
if not related_doc.uploaded_by:
|
||||
continue
|
||||
|
||||
# Не уведомляем если авторы совпадают
|
||||
if (main_document.uploaded_by and
|
||||
related_doc.uploaded_by.id == main_document.uploaded_by.id):
|
||||
continue
|
||||
|
||||
logger.info(
|
||||
f"[notification_signals] Notifying author={related_doc.uploaded_by.pk} "
|
||||
f"that their document was linked"
|
||||
)
|
||||
|
||||
notify.send(
|
||||
sender=main_document.uploaded_by,
|
||||
recipient=related_doc.uploaded_by,
|
||||
verb='document_related',
|
||||
action_object=main_document,
|
||||
description=(
|
||||
f'Ваш документ "{related_doc.title}" связан '
|
||||
f'с документом "{main_document.title}"'
|
||||
),
|
||||
action_url='/documents',
|
||||
data={
|
||||
'title': 'Документ связан с другим',
|
||||
'document_id': main_document.id,
|
||||
'related_document_id': related_doc.id,
|
||||
},
|
||||
)
|
||||
@ -1,177 +0,0 @@
|
||||
# # import os
|
||||
# # from dotenv import load_dotenv
|
||||
# # from asgiref.sync import async_to_sync
|
||||
|
||||
# # from aiogram import Bot
|
||||
# # from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
|
||||
|
||||
# # from .models import Document
|
||||
# # from django.contrib.auth import get_user_model
|
||||
# # from bots.models import BotSubscriber
|
||||
# # import logging
|
||||
|
||||
# # logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# # load_dotenv()
|
||||
# # BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
|
||||
|
||||
# # User = get_user_model()
|
||||
# # telegram_bot = Bot(token=BOT_TOKEN)
|
||||
|
||||
|
||||
# # def send_document_to_recipients(document_id: int):
|
||||
# # """
|
||||
# # Уведомляем в Telegram о появлении нового документа:
|
||||
# # — всем активным, если sent_to_all=True,
|
||||
# # — или тем, кто в recipients.
|
||||
# # """
|
||||
# # try:
|
||||
# # doc = Document.objects.get(pk=document_id)
|
||||
# # except Document.DoesNotExist:
|
||||
# # return
|
||||
|
||||
# # # собираем подписчиков
|
||||
# # if doc.sent_to_all:
|
||||
# # subs = BotSubscriber.objects.filter(
|
||||
# # telegram_id__isnull=False,
|
||||
# # user__is_active=True
|
||||
# # )
|
||||
# # else:
|
||||
# # subs = BotSubscriber.objects.filter(
|
||||
# # telegram_id__isnull=False,
|
||||
# # user__in=doc.recipients.filter(is_active=True)
|
||||
# # )
|
||||
|
||||
# # if not subs.exists():
|
||||
# # return
|
||||
|
||||
# # # клавиатура: ведёт пользователя в бот, в раздел «Документы»
|
||||
# # kb = InlineKeyboardMarkup(inline_keyboard=[[
|
||||
# # InlineKeyboardButton(text="📄 Открыть документы",
|
||||
# # callback_data="menu_docs")
|
||||
# # ]])
|
||||
|
||||
# # text = (
|
||||
# # f"🆕 *Новый документ*: {doc.title}\n\n"
|
||||
# # f"{doc.description or ''}\n\n"
|
||||
# # f"Зайдите в раздел «Документы» в боте, чтобы скачать файл."
|
||||
# # )
|
||||
|
||||
# # # шлём всем подписчикам
|
||||
# # for sub in subs:
|
||||
# # try:
|
||||
# # async_to_sync(telegram_bot.send_message)(
|
||||
# # chat_id=sub.telegram_id,
|
||||
# # text=text,
|
||||
# # parse_mode="Markdown",
|
||||
# # disable_web_page_preview=True,
|
||||
# # reply_markup=kb,
|
||||
# # )
|
||||
# # except Exception as e:
|
||||
# # # Записываем в лог, на каком именно подписчике и с каким текстом упало
|
||||
# # logger.error(
|
||||
# # f"[Telegram] Ошибка при отправке документу={doc.pk} "
|
||||
# # f"пользователю={sub.user.pk} tg_id={sub.telegram_id}: {e}",
|
||||
# # exc_info=True
|
||||
# # )
|
||||
# # else:
|
||||
# # logger.info(
|
||||
# # f"[Telegram] Уведомление по документу={doc.pk} "
|
||||
# # f"успешно отправлено пользователю={sub.user.pk}"
|
||||
# # )
|
||||
|
||||
|
||||
# # documents/tasks.py
|
||||
# import os
|
||||
# from dotenv import load_dotenv
|
||||
# from celery import shared_task
|
||||
# from django.conf import settings
|
||||
# from django.urls import reverse
|
||||
# from django.utils import timezone
|
||||
# from telegram import Bot as TgBot, InlineKeyboardButton, InlineKeyboardMarkup
|
||||
|
||||
# from .models import Document
|
||||
# from django.contrib.auth import get_user_model
|
||||
# from bots.models import BotSubscriber
|
||||
# import logging
|
||||
|
||||
# logger = logging.getLogger(__name__)
|
||||
|
||||
# load_dotenv()
|
||||
# TG_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN') or settings.TELEGRAM_BOT_TOKEN
|
||||
|
||||
# User = get_user_model()
|
||||
|
||||
|
||||
# # @shared_task(bind=True, name="documents.send_document_to_recipients")
|
||||
# def send_document_to_recipients(document_id):
|
||||
# try:
|
||||
# doc = Document.objects.get(pk=document_id)
|
||||
# except Document.DoesNotExist:
|
||||
# return
|
||||
|
||||
# # Собираем подписчиков
|
||||
# if doc.sent_to_all:
|
||||
# subs = BotSubscriber.objects.filter(
|
||||
# telegram_id__isnull=False,
|
||||
# user__is_active=True,
|
||||
# )
|
||||
# else:
|
||||
# subs = BotSubscriber.objects.filter(
|
||||
# telegram_id__isnull=False,
|
||||
# user__in=doc.recipients.filter(is_active=True),
|
||||
# )
|
||||
# if not subs.exists():
|
||||
# return
|
||||
|
||||
# # Синхронный телеграм-бот
|
||||
# bot = TgBot(token=TG_TOKEN)
|
||||
|
||||
# # Кнопка «Открыть документы»
|
||||
# kb = InlineKeyboardMarkup(
|
||||
# [[ InlineKeyboardButton("📄 Открыть документы", callback_data="menu_docs") ]]
|
||||
# )
|
||||
|
||||
# # Текст уведомления
|
||||
# text = (
|
||||
# f"🆕 *Новый документ*: {doc.title}\n\n"
|
||||
# f"{doc.description or ''}\n\n"
|
||||
# f"Дата публикации: {doc.uploaded_at.strftime('%d.%m.%Y %H:%M')}"
|
||||
# )
|
||||
|
||||
# # for sub in subs:
|
||||
# # try:
|
||||
# # bot.send_message(
|
||||
# # chat_id=sub.telegram_id,
|
||||
# # text=text,
|
||||
# # parse_mode="Markdown",
|
||||
# # reply_markup=kb,
|
||||
# # disable_web_page_preview=True
|
||||
# # )
|
||||
# # except Exception as e:
|
||||
# # print(e)
|
||||
# # логируем, чтобы понимать, кому не дошло
|
||||
# # self.retry(exc=e, countdown=60, max_retries=3)
|
||||
|
||||
# for sub in subs:
|
||||
# try:
|
||||
# bot.send_message(
|
||||
# chat_id=sub.telegram_id,
|
||||
# text=text,
|
||||
# parse_mode="Markdown",
|
||||
# disable_web_page_preview=True,
|
||||
# reply_markup=kb,
|
||||
# )
|
||||
# except Exception as e:
|
||||
# # Записываем в лог, на каком именно подписчике и с каким текстом упало
|
||||
# logger.error(
|
||||
# f"[Telegram] Ошибка при отправке документу={doc.pk} "
|
||||
# f"пользователю={sub.user.pk} tg_id={sub.telegram_id}: {e}",
|
||||
# exc_info=True
|
||||
# )
|
||||
# else:
|
||||
# logger.info(
|
||||
# f"[Telegram] Уведомление по документу={doc.pk} "
|
||||
# f"успешно отправлено пользователю={sub.user.pk}"
|
||||
# )
|
||||
@ -1,624 +0,0 @@
|
||||
# documents/test_documents_api.py
|
||||
"""
|
||||
Comprehensive pytest API tests для DocumentViewSet
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from rest_framework import status
|
||||
from filer.models import File as FilerFile
|
||||
from documents.models import (
|
||||
Document, DocumentAcknowledgement, DocumentComment
|
||||
)
|
||||
from reversion.models import Version
|
||||
import reversion
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# CRUD TESTS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestDocumentCRUDAPI:
|
||||
"""Тесты CRUD операций для документов"""
|
||||
|
||||
def test_list_documents_authenticated(self, viewer_client, document):
|
||||
"""GET /documents/ - авторизованный пользователь видит свои документы"""
|
||||
response = viewer_client.get('/api/v1/documents/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
# viewer в recipients - должен видеть документ
|
||||
assert len(response.data.get('results', response.data)) >= 1
|
||||
|
||||
def test_list_documents_unauthenticated(self, api_client):
|
||||
"""GET /documents/ - неавторизованный получает 401"""
|
||||
response = api_client.get('/api/v1/documents/')
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_create_document(self, authenticated_client, author_user):
|
||||
"""POST /documents/ - автор может создать документ"""
|
||||
file_obj = SimpleUploadedFile("new_doc.txt", b"Content", content_type="text/plain")
|
||||
|
||||
data = {
|
||||
'title': 'Новый документ',
|
||||
'description': 'Описание',
|
||||
'sent_to_all': True,
|
||||
'file': file_obj,
|
||||
}
|
||||
|
||||
response = authenticated_client.post('/api/v1/documents/', data, format='multipart')
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
assert response.data['title'] == 'Новый документ'
|
||||
|
||||
def test_retrieve_document_as_recipient(self, viewer_client, document):
|
||||
"""GET /documents/{id}/ - получатель может просмотреть"""
|
||||
response = viewer_client.get(f'/api/v1/documents/{document.id}/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data['title'] == document.title
|
||||
|
||||
def test_retrieve_document_as_non_recipient(self, api_client, other_user, document):
|
||||
"""GET /documents/{id}/ - не-получатель не видит документ (404 - отфильтрован queryset)"""
|
||||
api_client.force_authenticate(user=other_user)
|
||||
response = api_client.get(f'/api/v1/documents/{document.id}/')
|
||||
# QuerySet фильтрует недоступные документы, поэтому 404
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
def test_update_document_as_author(self, authenticated_client, document, author_user):
|
||||
"""PATCH /documents/{id}/ - автор может редактировать"""
|
||||
data = {'title': 'Обновленное название'}
|
||||
response = authenticated_client.patch(f'/api/v1/documents/{document.id}/', data)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
# Проверяем через API response (refresh_from_db не работает с FSM protected fields)
|
||||
assert response.data['title'] == 'Обновленное название'
|
||||
|
||||
def test_update_document_as_viewer_forbidden(self, viewer_client, document):
|
||||
"""PATCH /documents/{id}/ - получатель не может редактировать"""
|
||||
data = {'title': 'Попытка изменить'}
|
||||
response = viewer_client.patch(f'/api/v1/documents/{document.id}/', data)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
def test_delete_document_as_author(self, authenticated_client, document):
|
||||
"""DELETE /documents/{id}/ - автор может удалить"""
|
||||
doc_id = document.id
|
||||
response = authenticated_client.delete(f'/api/v1/documents/{doc_id}/')
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
assert not Document.objects.filter(id=doc_id).exists()
|
||||
|
||||
def test_delete_document_as_viewer_forbidden(self, viewer_client, document):
|
||||
"""DELETE /documents/{id}/ - получатель не может удалить"""
|
||||
response = viewer_client.delete(f'/api/v1/documents/{document.id}/')
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FSM WORKFLOW TESTS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestDocumentFSMWorkflow:
|
||||
"""Тесты FSM переходов состояний документа"""
|
||||
|
||||
def test_submit_for_review_transition(self, authenticated_client, document):
|
||||
"""POST submit-for-review: draft → in_review"""
|
||||
assert document.status == Document.Status.DRAFT
|
||||
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/submit-for-review/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data['status'] == Document.Status.IN_REVIEW
|
||||
|
||||
def test_approve_transition(self, authenticated_client, document):
|
||||
"""POST approve: in_review → approved"""
|
||||
document.submit_for_review()
|
||||
document.save()
|
||||
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/approve/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data['status'] == Document.Status.APPROVED
|
||||
|
||||
def test_reject_transition(self, authenticated_client, document):
|
||||
"""POST reject: in_review → rejected"""
|
||||
document.submit_for_review()
|
||||
document.save()
|
||||
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/reject/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data['status'] == Document.Status.REJECTED
|
||||
|
||||
def test_publish_transition(self, authenticated_client, document):
|
||||
"""POST publish: approved → published"""
|
||||
document.submit_for_review()
|
||||
document.approve()
|
||||
document.save()
|
||||
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/publish/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data['status'] == Document.Status.PUBLISHED
|
||||
|
||||
def test_return_to_draft_transition(self, authenticated_client, document):
|
||||
"""POST return-to-draft: in_review → draft"""
|
||||
document.submit_for_review()
|
||||
document.save()
|
||||
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/return-to-draft/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data['status'] == Document.Status.DRAFT
|
||||
|
||||
def test_archive_transition(self, authenticated_client, document):
|
||||
"""POST archive: published → archived"""
|
||||
document.submit_for_review()
|
||||
document.approve()
|
||||
document.publish()
|
||||
document.save()
|
||||
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/archive/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data['status'] == Document.Status.ARCHIVED
|
||||
|
||||
def test_unarchive_transition(self, authenticated_client, document):
|
||||
"""POST unarchive: archived → published"""
|
||||
document.submit_for_review()
|
||||
document.approve()
|
||||
document.publish()
|
||||
document.archive()
|
||||
document.save()
|
||||
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/unarchive/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
assert document.status == Document.Status.PUBLISHED
|
||||
|
||||
def test_invalid_transition_returns_400(self, authenticated_client, document):
|
||||
"""Невалидный переход возвращает 400"""
|
||||
# Пытаемся reject когда status=DRAFT (невалидно)
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/reject/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# VERSIONING TESTS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestDocumentVersioning:
|
||||
"""Тесты версионирования django-reversion"""
|
||||
|
||||
def test_get_versions(self, authenticated_client, document):
|
||||
"""GET /documents/{id}/versions/ - история версий"""
|
||||
# Создаем версии через reversion
|
||||
with reversion.create_revision():
|
||||
document.title = "Version 1"
|
||||
document.save()
|
||||
reversion.set_comment("First change")
|
||||
|
||||
with reversion.create_revision():
|
||||
document.title = "Version 2"
|
||||
document.save()
|
||||
reversion.set_comment("Second change")
|
||||
|
||||
response = authenticated_client.get(
|
||||
f'/api/v1/documents/{document.id}/versions/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
versions = response.data
|
||||
assert len(versions) >= 2
|
||||
|
||||
def test_get_activity_timeline(self, authenticated_client, viewer_user, document):
|
||||
"""GET /documents/{id}/activity/ - timeline активности"""
|
||||
# Создаем ознакомление
|
||||
DocumentAcknowledgement.objects.create(
|
||||
document=document,
|
||||
user=viewer_user
|
||||
)
|
||||
|
||||
# Создаем версию
|
||||
with reversion.create_revision():
|
||||
document.title = "Changed"
|
||||
document.save()
|
||||
|
||||
response = authenticated_client.get(
|
||||
f'/api/v1/documents/{document.id}/activity/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
activity = response.data
|
||||
assert len(activity) > 0
|
||||
|
||||
# Проверяем типы в timeline
|
||||
types = [item['type'] for item in activity]
|
||||
assert 'version' in types or 'acknowledgement' in types
|
||||
|
||||
def test_revert_to_version(self, authenticated_client, document):
|
||||
"""POST /documents/{id}/revert/ - откат к версии"""
|
||||
original_title = "Original Title"
|
||||
|
||||
# Сохраняем первую версию
|
||||
with reversion.create_revision():
|
||||
document.title = original_title
|
||||
document.save()
|
||||
|
||||
# Получаем первую версию
|
||||
versions = Version.objects.get_for_object(document)
|
||||
first_version = versions.first()
|
||||
|
||||
# Меняем документ
|
||||
with reversion.create_revision():
|
||||
document.title = "Modified Title"
|
||||
document.save()
|
||||
|
||||
# Откатываем
|
||||
data = {
|
||||
'version_id': first_version.id,
|
||||
'comment': 'Откат'
|
||||
}
|
||||
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/revert/',
|
||||
data
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.data['title'] == original_title
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# ACKNOWLEDGEMENT TESTS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestDocumentAcknowledgement:
|
||||
"""Тесты ознакомления с документами"""
|
||||
|
||||
def test_acknowledge_document(self, viewer_client, viewer_user, document):
|
||||
"""POST /documents/{id}/acknowledge/ - подтвердить ознакомление"""
|
||||
response = viewer_client.post(
|
||||
f'/api/v1/documents/{document.id}/acknowledge/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
# Проверяем создание записи
|
||||
assert DocumentAcknowledgement.objects.filter(
|
||||
document=document,
|
||||
user=viewer_user
|
||||
).exists()
|
||||
|
||||
def test_acknowledge_non_recipient_forbidden(self, api_client, other_user, document):
|
||||
"""Не-получатель не может ознакомиться"""
|
||||
api_client.force_authenticate(user=other_user)
|
||||
|
||||
response = api_client.post(
|
||||
f'/api/v1/documents/{document.id}/acknowledge/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
def test_get_acknowledgements_list(self, viewer_client, viewer_user, document):
|
||||
"""GET /documents/{id}/acknowledgements/ - список ознакомившихся"""
|
||||
# Создаем ознакомление
|
||||
DocumentAcknowledgement.objects.create(
|
||||
document=document,
|
||||
user=viewer_user
|
||||
)
|
||||
|
||||
response = viewer_client.get(
|
||||
f'/api/v1/documents/{document.id}/acknowledgements/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
acks = response.data
|
||||
assert len(acks) == 1
|
||||
assert acks[0]['user']['id'] == viewer_user.id
|
||||
|
||||
def test_double_acknowledge_prevented(self, viewer_client, viewer_user, document):
|
||||
"""Двойное ознакомление запрещено"""
|
||||
# Первое ознакомление
|
||||
viewer_client.post(f'/api/v1/documents/{document.id}/acknowledge/')
|
||||
|
||||
# Второе ознакомление - должно вернуть ошибку или already acknowledged
|
||||
response = viewer_client.post(
|
||||
f'/api/v1/documents/{document.id}/acknowledge/'
|
||||
)
|
||||
# Может вернуть 200 с сообщением "already acknowledged" или 400
|
||||
# Главное - запись одна
|
||||
count = DocumentAcknowledgement.objects.filter(
|
||||
document=document,
|
||||
user=viewer_user
|
||||
).count()
|
||||
assert count == 1
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# COMMENT TESTS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestDocumentComments:
|
||||
"""Тесты комментариев к документам"""
|
||||
|
||||
def test_create_comment(self, viewer_client, viewer_user, document):
|
||||
"""POST /document-comments/ - создать комментарий"""
|
||||
data = {
|
||||
'text': 'Отличный документ!',
|
||||
'document_id': document.id
|
||||
}
|
||||
|
||||
response = viewer_client.post('/api/v1/document-comments/', data)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
assert DocumentComment.objects.filter(
|
||||
document=document,
|
||||
author=viewer_user
|
||||
).exists()
|
||||
|
||||
def test_create_reply_to_comment(self, authenticated_client, viewer_client, viewer_user, author_user, document):
|
||||
"""Создание ответа на комментарий"""
|
||||
# Создаем родительский комментарий
|
||||
parent = DocumentComment.objects.create(
|
||||
document=document,
|
||||
author=viewer_user,
|
||||
text='Parent comment'
|
||||
)
|
||||
|
||||
# Создаем ответ
|
||||
data = {
|
||||
'text': 'Reply to parent',
|
||||
'document_id': document.id,
|
||||
'parent_id': parent.id
|
||||
}
|
||||
|
||||
response = authenticated_client.post('/api/v1/document-comments/', data)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
reply = DocumentComment.objects.get(parent=parent)
|
||||
assert reply.text == 'Reply to parent'
|
||||
|
||||
def test_get_comment_replies(self, viewer_client, viewer_user, author_user, document):
|
||||
"""GET /comments/{id}/replies/ - получить ответы"""
|
||||
parent = DocumentComment.objects.create(
|
||||
document=document,
|
||||
author=viewer_user,
|
||||
text='Parent'
|
||||
)
|
||||
|
||||
DocumentComment.objects.create(
|
||||
document=document,
|
||||
author=author_user,
|
||||
text='Reply 1',
|
||||
parent=parent
|
||||
)
|
||||
|
||||
DocumentComment.objects.create(
|
||||
document=document,
|
||||
author=author_user,
|
||||
text='Reply 2',
|
||||
parent=parent
|
||||
)
|
||||
|
||||
response = viewer_client.get(f'/api/v1/document-comments/{parent.id}/replies/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.data) == 2
|
||||
|
||||
def test_delete_own_comment(self, viewer_client, viewer_user, document):
|
||||
"""Пользователь может удалить свой комментарий"""
|
||||
comment = DocumentComment.objects.create(
|
||||
document=document,
|
||||
author=viewer_user,
|
||||
text='My comment'
|
||||
)
|
||||
|
||||
response = viewer_client.delete(f'/api/v1/document-comments/{comment.id}/')
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
def test_delete_others_comment_forbidden(self, viewer_client, author_user, document):
|
||||
"""Нельзя удалить чужой комментарий"""
|
||||
comment = DocumentComment.objects.create(
|
||||
document=document,
|
||||
author=author_user,
|
||||
text='Author comment'
|
||||
)
|
||||
|
||||
response = viewer_client.delete(f'/api/v1/document-comments/{comment.id}/')
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# PERMISSIONS TESTS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestDocumentPermissions:
|
||||
"""Тесты системы разрешений"""
|
||||
|
||||
def test_sent_to_all_visible_to_everyone(self, api_client, other_user, document_sent_to_all):
|
||||
"""sent_to_all=True - документ видим всем активным пользователям"""
|
||||
api_client.force_authenticate(user=other_user)
|
||||
|
||||
response = api_client.get(f'/api/v1/documents/{document_sent_to_all.id}/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
def test_private_document_hidden_from_others(self, api_client, other_user, document):
|
||||
"""Приватный документ скрыт от не-получателей (404 - отфильтрован)"""
|
||||
api_client.force_authenticate(user=other_user)
|
||||
|
||||
response = api_client.get(f'/api/v1/documents/{document.id}/')
|
||||
# QuerySet фильтрует недоступные документы
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
def test_department_member_can_view(self, api_client, other_user, document, department):
|
||||
"""Член отдела-получателя может видеть документ"""
|
||||
# Добавляем пользователя в отдел
|
||||
other_user.department = department
|
||||
other_user.save()
|
||||
|
||||
# Добавляем отдел в получатели документа
|
||||
document.departments.add(department)
|
||||
|
||||
api_client.force_authenticate(user=other_user)
|
||||
|
||||
response = api_client.get(f'/api/v1/documents/{document.id}/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
def test_author_can_always_edit(self, authenticated_client, document):
|
||||
"""Автор всегда может редактировать свой документ"""
|
||||
data = {'description': 'Updated by author'}
|
||||
response = authenticated_client.patch(
|
||||
f'/api/v1/documents/{document.id}/',
|
||||
data
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
def test_admin_has_full_access(self, admin_client, document):
|
||||
"""Admin имеет полный доступ"""
|
||||
# Должен видеть документ
|
||||
response = admin_client.get(f'/api/v1/documents/{document.id}/')
|
||||
# Admin может иметь доступ или нет в зависимости от rules
|
||||
assert response.status_code in [status.HTTP_200_OK, status.HTTP_403_FORBIDDEN]
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FOLDER TESTS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestFolderAPI:
|
||||
"""Тесты для папок документов"""
|
||||
|
||||
def test_list_folders(self, authenticated_client, folder):
|
||||
"""GET /folders/ - список папок"""
|
||||
response = authenticated_client.get('/api/v1/folders/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
def test_create_folder(self, authenticated_client):
|
||||
"""POST /folders/ - создание папки"""
|
||||
data = {'name': 'Новая папка'}
|
||||
response = authenticated_client.post('/api/v1/folders/', data)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
def test_get_folder_children(self, authenticated_client, folder, author_user):
|
||||
"""GET /folders/{id}/children/ - дочерние папки"""
|
||||
# Создаем дочернюю папку
|
||||
from filer.models import Folder as FilerFolder
|
||||
child = FilerFolder.objects.create(
|
||||
name='Дочерняя',
|
||||
parent=folder,
|
||||
owner=author_user
|
||||
)
|
||||
|
||||
response = authenticated_client.get(f'/api/v1/folders/{folder.id}/children/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.data) == 1
|
||||
|
||||
def test_get_folder_documents(self, authenticated_client, folder, document, filer_file):
|
||||
"""GET /folders/{id}/documents/ - документы в папке"""
|
||||
# Перемещаем документ в папку
|
||||
document.folder = folder
|
||||
document.save()
|
||||
|
||||
response = authenticated_client.get(f'/api/v1/folders/{folder.id}/documents/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.data) == 1
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# TAG & TYPE TESTS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestDocumentTagsAndTypes:
|
||||
"""Тесты для тегов и типов документов"""
|
||||
|
||||
def test_list_tags(self, authenticated_client, document_tag):
|
||||
"""GET /document-tags/ - список тегов"""
|
||||
response = authenticated_client.get('/api/v1/document-tags/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
def test_create_tag(self, authenticated_client):
|
||||
"""POST /document-tags/ - создание тега"""
|
||||
data = {'name': 'Срочно', 'color': '#FF0000'}
|
||||
response = authenticated_client.post('/api/v1/document-tags/', data)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
def test_list_document_types(self, authenticated_client, document_type):
|
||||
"""GET /document-types/ - список типов"""
|
||||
response = authenticated_client.get('/api/v1/document-types/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
def test_get_documents_by_tag(self, authenticated_client, document, document_tag):
|
||||
"""GET /tags/{id}/documents/ - документы с тегом"""
|
||||
# Добавляем тег к документу (если есть M2M)
|
||||
if hasattr(document, 'tags'):
|
||||
document.tags.add(document_tag)
|
||||
|
||||
response = authenticated_client.get(f'/api/v1/document-tags/{document_tag.id}/documents/')
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# RELATED DOCUMENTS TESTS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.django_db
|
||||
class TestRelatedDocuments:
|
||||
"""Тесты для связанных документов"""
|
||||
|
||||
def test_get_related_documents(self, authenticated_client, document):
|
||||
"""GET /documents/{id}/related/ - связанные документы"""
|
||||
response = authenticated_client.get(
|
||||
f'/api/v1/documents/{document.id}/related/'
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
def test_add_related_document(self, authenticated_client, document, filer_file, author_user):
|
||||
"""POST /documents/{id}/add_related/ - добавить связь"""
|
||||
# Создаем второй документ
|
||||
doc2 = Document.objects.create(
|
||||
title="Связанный документ",
|
||||
file=filer_file,
|
||||
uploaded_by=author_user,
|
||||
sent_to_all=True
|
||||
)
|
||||
|
||||
data = {'document_id': doc2.id}
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/add_related/',
|
||||
data
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
# Проверяем связь
|
||||
assert doc2 in document.related_documents.all()
|
||||
|
||||
def test_remove_related_document(self, authenticated_client, document, filer_file, author_user):
|
||||
"""POST /documents/{id}/remove_related/ - удалить связь"""
|
||||
# Создаем и связываем документ
|
||||
doc2 = Document.objects.create(
|
||||
title="Связанный",
|
||||
file=filer_file,
|
||||
uploaded_by=author_user,
|
||||
sent_to_all=True
|
||||
)
|
||||
document.related_documents.add(doc2)
|
||||
|
||||
data = {'document_id': doc2.id}
|
||||
response = authenticated_client.post(
|
||||
f'/api/v1/documents/{document.id}/remove_related/',
|
||||
data
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
# Проверяем что связь удалена
|
||||
assert doc2 not in document.related_documents.all()
|
||||
@ -1,298 +0,0 @@
|
||||
# documents/tests.py
|
||||
"""
|
||||
Тесты для моделей Document и DocumentAcknowledgement с django-filer
|
||||
"""
|
||||
|
||||
from django.test import TestCase
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from filer.models import File as FilerFile
|
||||
from documents.models import Document, DocumentAcknowledgement
|
||||
from employees.models import Department
|
||||
import rules
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class DocumentModelTest(TestCase):
|
||||
"""Тесты для модели Document"""
|
||||
|
||||
def setUp(self):
|
||||
"""Создание тестовых данных"""
|
||||
self.user = User.objects.create_user(
|
||||
phone_number='+79999999999',
|
||||
email='test1@example.com',
|
||||
first_name='Тест',
|
||||
last_name='Тестов',
|
||||
is_active=True
|
||||
)
|
||||
|
||||
# Создаем тестовый файл через filer
|
||||
test_file = SimpleUploadedFile(
|
||||
"test_document.txt",
|
||||
b"Test content",
|
||||
content_type="text/plain"
|
||||
)
|
||||
|
||||
self.filer_file = FilerFile.objects.create(
|
||||
file=test_file,
|
||||
original_filename="test_document.txt",
|
||||
name="Test Document",
|
||||
owner=self.user
|
||||
)
|
||||
|
||||
def test_create_document(self):
|
||||
"""Создание документа с filer файлом"""
|
||||
document = Document.objects.create(
|
||||
title="Тестовый документ",
|
||||
file=self.filer_file,
|
||||
description="Описание тестового документа",
|
||||
uploaded_by=self.user,
|
||||
sent_to_all=True
|
||||
)
|
||||
|
||||
self.assertEqual(document.title, "Тестовый документ")
|
||||
self.assertEqual(document.file, self.filer_file)
|
||||
self.assertEqual(document.uploaded_by, self.user)
|
||||
self.assertTrue(document.sent_to_all)
|
||||
self.assertIsNotNone(document.uploaded_at)
|
||||
|
||||
def test_file_properties(self):
|
||||
"""Проверка свойств файла (размер, расширение)"""
|
||||
document = Document.objects.create(
|
||||
title="Тестовый документ",
|
||||
file=self.filer_file,
|
||||
uploaded_by=self.user
|
||||
)
|
||||
|
||||
self.assertGreater(document.file_size, 0)
|
||||
self.assertEqual(document.file_extension, 'txt')
|
||||
|
||||
def test_sent_to_all(self):
|
||||
"""Документ с sent_to_all=True"""
|
||||
document = Document.objects.create(
|
||||
title="Документ для всех",
|
||||
file=self.filer_file,
|
||||
uploaded_by=self.user,
|
||||
sent_to_all=True
|
||||
)
|
||||
|
||||
self.assertTrue(document.sent_to_all)
|
||||
self.assertEqual(document.departments.count(), 0)
|
||||
self.assertEqual(document.recipients.count(), 0)
|
||||
|
||||
def test_department_recipients(self):
|
||||
"""Документ для конкретных отделов"""
|
||||
department = Department.objects.create(
|
||||
name="Тестовый отдел"
|
||||
)
|
||||
|
||||
document = Document.objects.create(
|
||||
title="Документ для отдела",
|
||||
file=self.filer_file,
|
||||
uploaded_by=self.user,
|
||||
sent_to_all=False
|
||||
)
|
||||
document.departments.add(department)
|
||||
|
||||
self.assertFalse(document.sent_to_all)
|
||||
self.assertEqual(document.departments.count(), 1)
|
||||
self.assertIn(department, document.departments.all())
|
||||
|
||||
def test_individual_recipients(self):
|
||||
"""Документ для конкретных получателей"""
|
||||
recipient = User.objects.create_user(
|
||||
phone_number='+79999999998',
|
||||
email='recipient@example.com',
|
||||
first_name='Получатель',
|
||||
last_name='Тестовый',
|
||||
is_active=True
|
||||
)
|
||||
|
||||
document = Document.objects.create(
|
||||
title="Документ для получателя",
|
||||
file=self.filer_file,
|
||||
uploaded_by=self.user,
|
||||
sent_to_all=False
|
||||
)
|
||||
document.recipients.add(recipient)
|
||||
|
||||
self.assertFalse(document.sent_to_all)
|
||||
self.assertEqual(document.recipients.count(), 1)
|
||||
self.assertIn(recipient, document.recipients.all())
|
||||
|
||||
|
||||
class DocumentAcknowledgementModelTest(TestCase):
|
||||
"""Тесты для модели DocumentAcknowledgement"""
|
||||
|
||||
def setUp(self):
|
||||
"""Создание тестовых данных"""
|
||||
self.user = User.objects.create_user(
|
||||
phone_number='+79999999999',
|
||||
email='test@example.com',
|
||||
first_name='Тест',
|
||||
last_name='Тестов',
|
||||
is_active=True
|
||||
)
|
||||
|
||||
test_file = SimpleUploadedFile(
|
||||
"test_document.txt",
|
||||
b"Test content",
|
||||
content_type="text/plain"
|
||||
)
|
||||
|
||||
filer_file = FilerFile.objects.create(
|
||||
file=test_file,
|
||||
original_filename="test_document.txt",
|
||||
name="Test Document",
|
||||
owner=self.user
|
||||
)
|
||||
|
||||
self.document = Document.objects.create(
|
||||
title="Тестовый документ",
|
||||
file=filer_file,
|
||||
uploaded_by=self.user,
|
||||
sent_to_all=True
|
||||
)
|
||||
|
||||
def test_create_acknowledgement(self):
|
||||
"""Создание ознакомления с документом"""
|
||||
ack = DocumentAcknowledgement.objects.create(
|
||||
document=self.document,
|
||||
user=self.user
|
||||
)
|
||||
|
||||
self.assertEqual(ack.document, self.document)
|
||||
self.assertEqual(ack.user, self.user)
|
||||
self.assertIsNotNone(ack.acknowledged_at)
|
||||
|
||||
def test_unique_acknowledgement(self):
|
||||
"""Один пользователь может ознакомиться только один раз"""
|
||||
DocumentAcknowledgement.objects.create(
|
||||
document=self.document,
|
||||
user=self.user
|
||||
)
|
||||
|
||||
# Попытка создать второе ознакомление должна вызвать ошибку
|
||||
with self.assertRaises(Exception):
|
||||
DocumentAcknowledgement.objects.create(
|
||||
document=self.document,
|
||||
user=self.user
|
||||
)
|
||||
|
||||
def test_acknowledgement_count(self):
|
||||
"""Подсчет ознакомлений"""
|
||||
user2 = User.objects.create_user(
|
||||
phone_number='+79999999998',
|
||||
email='test2@example.com',
|
||||
first_name='Тест2',
|
||||
last_name='Тестов2',
|
||||
is_active=True
|
||||
)
|
||||
|
||||
DocumentAcknowledgement.objects.create(
|
||||
document=self.document,
|
||||
user=self.user
|
||||
)
|
||||
DocumentAcknowledgement.objects.create(
|
||||
document=self.document,
|
||||
user=user2
|
||||
)
|
||||
|
||||
self.assertEqual(self.document.acknowledgements.count(), 2)
|
||||
|
||||
|
||||
class DocumentRulesTest(TestCase):
|
||||
"""Тесты для правил доступа django-rules"""
|
||||
|
||||
def setUp(self):
|
||||
"""Создание тестовых данных"""
|
||||
self.owner = User.objects.create_user(
|
||||
phone_number='+79999999999',
|
||||
email='owner@example.com',
|
||||
first_name='Владелец',
|
||||
last_name='Документа',
|
||||
is_active=True
|
||||
)
|
||||
|
||||
self.other_user = User.objects.create_user(
|
||||
phone_number='+79999999998',
|
||||
email='other@example.com',
|
||||
first_name='Другой',
|
||||
last_name='Пользователь',
|
||||
is_active=True
|
||||
)
|
||||
|
||||
test_file = SimpleUploadedFile(
|
||||
"test_document.txt",
|
||||
b"Test content",
|
||||
content_type="text/plain"
|
||||
)
|
||||
|
||||
filer_file = FilerFile.objects.create(
|
||||
file=test_file,
|
||||
original_filename="test_document.txt",
|
||||
name="Test Document",
|
||||
owner=self.owner
|
||||
)
|
||||
|
||||
self.document = Document.objects.create(
|
||||
title="Тестовый документ",
|
||||
file=filer_file,
|
||||
uploaded_by=self.owner,
|
||||
sent_to_all=False # Не для всех
|
||||
)
|
||||
|
||||
def test_owner_can_view(self):
|
||||
"""Владелец может просматривать свой документ"""
|
||||
self.assertTrue(
|
||||
rules.test_rule('documents.view_document', self.owner, self.document)
|
||||
)
|
||||
|
||||
def test_other_user_cannot_view(self):
|
||||
"""Другой пользователь не может просматривать чужой документ (если не в списке)"""
|
||||
self.assertFalse(
|
||||
rules.test_rule('documents.view_document', self.other_user, self.document)
|
||||
)
|
||||
|
||||
def test_sent_to_all_allows_view(self):
|
||||
"""Документ с sent_to_all=True доступен всем активным пользователям"""
|
||||
self.document.sent_to_all = True
|
||||
self.document.save()
|
||||
|
||||
self.assertTrue(
|
||||
rules.test_rule('documents.view_document', self.other_user, self.document)
|
||||
)
|
||||
|
||||
def test_recipient_can_view(self):
|
||||
"""Получатель из списка может просматривать документ"""
|
||||
self.document.recipients.add(self.other_user)
|
||||
|
||||
self.assertTrue(
|
||||
rules.test_rule('documents.view_document', self.other_user, self.document)
|
||||
)
|
||||
|
||||
def test_owner_can_edit(self):
|
||||
"""Владелец может редактировать свой документ"""
|
||||
self.assertTrue(
|
||||
rules.test_rule('documents.change_document', self.owner, self.document)
|
||||
)
|
||||
|
||||
def test_other_user_cannot_edit(self):
|
||||
"""Другой пользователь не может редактировать чужой документ"""
|
||||
self.assertFalse(
|
||||
rules.test_rule('documents.change_document', self.other_user, self.document)
|
||||
)
|
||||
|
||||
def test_owner_can_delete(self):
|
||||
"""Владелец может удалять свой документ"""
|
||||
self.assertTrue(
|
||||
rules.test_rule('documents.delete_document', self.owner, self.document)
|
||||
)
|
||||
|
||||
def test_other_user_cannot_delete(self):
|
||||
"""Другой пользователь не может удалять чужой документ"""
|
||||
self.assertFalse(
|
||||
rules.test_rule('documents.delete_document', self.other_user, self.document)
|
||||
)
|
||||
|
||||
@ -1,10 +0,0 @@
|
||||
# documents/urls.py
|
||||
from django.urls import path
|
||||
from .views import DocumentView, acknowledge_document
|
||||
|
||||
app_name = 'documents'
|
||||
|
||||
urlpatterns = [
|
||||
path('', DocumentView.as_view(), name='document_list'),
|
||||
path('ack/<int:pk>/', acknowledge_document, name='acknowledge'),
|
||||
]
|
||||
@ -1,435 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple
|
||||
|
||||
from api.client import get_api_client
|
||||
from api.decorators import require_api_auth
|
||||
from django.contrib import messages
|
||||
from django.contrib.auth.decorators import login_required
|
||||
from django.contrib.auth.models import AbstractBaseUser
|
||||
from django.contrib.auth.mixins import LoginRequiredMixin
|
||||
from django.http import HttpRequest, HttpResponse
|
||||
from django.shortcuts import redirect
|
||||
from django.urls import reverse
|
||||
from django.utils.decorators import method_decorator
|
||||
from django.utils.dateparse import parse_datetime
|
||||
from django.utils.timezone import localtime
|
||||
from django.views.decorators.csrf import csrf_protect
|
||||
from django.views.generic import TemplateView
|
||||
|
||||
|
||||
# ==============================
|
||||
# Вспомогательные функции
|
||||
# ==============================
|
||||
|
||||
|
||||
def _as_bool(value: Any) -> bool:
|
||||
"""Нормализует булево значение из строк HTML-форм.
|
||||
|
||||
Args:
|
||||
value: Строка/булево/None.
|
||||
|
||||
Returns:
|
||||
bool: True, если значение похоже на истину ('1', 'true', 'on', 'yes').
|
||||
|
||||
Raises:
|
||||
ValueError: Если значение имеет неожиданный тип, не приводимый к строке.
|
||||
"""
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
try:
|
||||
s = str(value).strip().lower()
|
||||
except Exception as exc: # pragma: no cover - защитный блок
|
||||
raise ValueError("Unsupported boolean value") from exc
|
||||
return s in {"1", "true", "on", "yes"}
|
||||
|
||||
|
||||
def _repeat_field_pairs(field: str, values: Iterable[int | str]) -> List[Tuple[str, str]]:
|
||||
"""Готовит список повторяемых пар (field, value) для multipart-формы.
|
||||
|
||||
Args:
|
||||
field (str): Имя поля.
|
||||
values (Iterable[int|str]): Значения (могут быть пустыми/None — игнорируются).
|
||||
|
||||
Returns:
|
||||
list[tuple[str, str]]: Повторяемые пары для передачи в API-клиент.
|
||||
|
||||
Raises:
|
||||
TypeError: Если field пустой или values неитерируемый.
|
||||
"""
|
||||
if not field:
|
||||
raise TypeError("field must be a non-empty string")
|
||||
out: List[Tuple[str, str]] = []
|
||||
for v in values or []:
|
||||
if v is None:
|
||||
continue
|
||||
s = str(v).strip()
|
||||
if s:
|
||||
out.append((field, s))
|
||||
return out
|
||||
|
||||
|
||||
def _api_unpack(resp: Any) -> Tuple[bool, Dict[str, Any], int]:
|
||||
"""Безопасно извлекает признаки успеха, JSON и статус из ответа API.
|
||||
|
||||
Args:
|
||||
resp: Объект ответа клиентской обёртки (ожидаются атрибуты .ok, .json, .status).
|
||||
|
||||
Returns:
|
||||
tuple: (ok, data, status)
|
||||
|
||||
Raises:
|
||||
AttributeError: Если у ответа нет ожидаемых атрибутов.
|
||||
"""
|
||||
ok: bool = getattr(resp, "ok")
|
||||
status: int = getattr(resp, "status")
|
||||
data: Dict[str, Any] = getattr(resp, "json") or {}
|
||||
return ok, data, status
|
||||
|
||||
|
||||
def _parse_page_payload(
|
||||
payload: Any,
|
||||
) -> Tuple[List[Dict[str, Any]], Optional[str], Optional[str], Optional[int]]:
|
||||
"""Нормализует ответ API к (results, next, previous, count).
|
||||
|
||||
Поддерживает форматы:
|
||||
- пагинация DRF: {"count": N, "next": URL, "previous": URL,
|
||||
"results": [...]}
|
||||
- сырой список: [ ... ]
|
||||
|
||||
Args:
|
||||
payload (Any): Десериализованный JSON из API.
|
||||
|
||||
Returns:
|
||||
Tuple: (results, next_url, prev_url, count)
|
||||
"""
|
||||
# 1) Непагинированный список
|
||||
if isinstance(payload, list):
|
||||
results = _format_datetime_fields(payload)
|
||||
return results, None, None, len(results)
|
||||
|
||||
# 2) Словарь (пагинированный или нет)
|
||||
if isinstance(payload, dict):
|
||||
if "results" in payload:
|
||||
raw = payload.get("results") or []
|
||||
results = _format_datetime_fields(list(raw))
|
||||
return (
|
||||
results,
|
||||
payload.get("next"),
|
||||
payload.get("previous"),
|
||||
payload.get("count")
|
||||
)
|
||||
# словарь, но не страница — пустой список
|
||||
return [], None, None, None
|
||||
|
||||
# 3) Непонятный формат
|
||||
return [], None, None, None
|
||||
|
||||
|
||||
def _user_can_manage_documents(user: AbstractBaseUser) -> bool:
|
||||
"""Проверяет, есть ли у пользователя административные/модельные права на Document.
|
||||
|
||||
Args:
|
||||
user (AbstractBaseUser): Пользователь.
|
||||
|
||||
Returns:
|
||||
bool: True, если is_staff или есть одно из прав add/change/delete/view.
|
||||
"""
|
||||
if getattr(user, "is_staff", False):
|
||||
return True
|
||||
model = "document"
|
||||
app_label = "documents"
|
||||
for act in ("view", "add", "change", "delete"):
|
||||
if user.has_perm(f"{app_label}.{act}_{model}"):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _fmt_dt(val, fmt: str = "%d.%m.%Y %H:%M") -> str | None:
|
||||
"""Превращает ISO-строку/дату в локализованную строку.
|
||||
|
||||
Args:
|
||||
val: ISO строка даты или datetime объект.
|
||||
fmt: Формат вывода даты.
|
||||
|
||||
Returns:
|
||||
str | None: Отформатированная строка или None.
|
||||
"""
|
||||
if not val:
|
||||
return None
|
||||
# если уже datetime — ок, если строка — парсим
|
||||
dt = parse_datetime(val) if isinstance(val, str) else val
|
||||
if dt is None:
|
||||
return val # на всякий случай оставим как есть
|
||||
try:
|
||||
return localtime(dt).strftime(fmt) # уважим таймзону пользователя
|
||||
except Exception:
|
||||
return val
|
||||
|
||||
|
||||
def _format_datetime_fields(
|
||||
items: List[Any],
|
||||
fields: tuple[str, ...] = ("uploaded_at", "created_at", "updated_at"),
|
||||
) -> List[Any]:
|
||||
"""Форматирует поля дат в списке объектов.
|
||||
|
||||
Args:
|
||||
items: Список объектов (словарей).
|
||||
fields: Кортеж имен полей с датами.
|
||||
|
||||
Returns:
|
||||
List[Any]: Список с отформатированными датами.
|
||||
"""
|
||||
out: List[Any] = []
|
||||
for it in items:
|
||||
if isinstance(it, Mapping):
|
||||
d = dict(it) # копия, не мутируем исходник
|
||||
for f in fields:
|
||||
if f in d:
|
||||
d[f] = _fmt_dt(d.get(f))
|
||||
out.append(d)
|
||||
else:
|
||||
out.append(it)
|
||||
return out
|
||||
|
||||
|
||||
def _err_msg(payload: Any, fallback: str = "") -> str:
|
||||
"""Извлекает сообщение об ошибке из ответа API.
|
||||
|
||||
Args:
|
||||
payload: Полезная нагрузка ответа API.
|
||||
fallback: Сообщение по умолчанию.
|
||||
|
||||
Returns:
|
||||
str: Сообщение об ошибке.
|
||||
"""
|
||||
# dict: detail / non_field_errors / первое поле со строкой
|
||||
if isinstance(payload, dict):
|
||||
if isinstance(payload.get("detail"), (str, list)):
|
||||
d = payload["detail"]
|
||||
return d if isinstance(d, str) else "; ".join(map(str, d))
|
||||
if isinstance(payload.get("non_field_errors"), list):
|
||||
return "; ".join(map(str, payload["non_field_errors"]))
|
||||
# первая строка из значений полей
|
||||
for v in payload.values():
|
||||
if isinstance(v, list) and v and isinstance(v[0], str):
|
||||
return "; ".join(map(str, v))
|
||||
if isinstance(v, str):
|
||||
return v
|
||||
# list: склеить
|
||||
if isinstance(payload, list):
|
||||
return "; ".join(map(str, payload))
|
||||
# str или что-то иное
|
||||
return str(payload or fallback)
|
||||
|
||||
|
||||
def _current_url_with(request: HttpRequest, **new_params: Any) -> str:
|
||||
"""Возвращает URL текущего пути с подменой/удалением query-параметров.
|
||||
|
||||
Args:
|
||||
request (HttpRequest): Текущий запрос.
|
||||
**new_params: Пары ключ-значение; None удаляет параметр.
|
||||
|
||||
Returns:
|
||||
str: Абсолютный путь с новой query-строкой.
|
||||
"""
|
||||
q = request.GET.copy()
|
||||
for k, v in new_params.items():
|
||||
if v is None and k in q:
|
||||
del q[k]
|
||||
elif v is not None:
|
||||
q[k] = str(v)
|
||||
qs = q.urlencode()
|
||||
return f"{request.path}?{qs}" if qs else request.path
|
||||
|
||||
|
||||
def _filter_for_user(items: Sequence[Dict[str, Any]], user_id: Optional[int]) -> List[Dict[str, Any]]:
|
||||
"""Фильтрует документы в область «Мои».
|
||||
|
||||
Документ считается «моим», если:
|
||||
- он отправлен всем (`sent_to_all == true`), или
|
||||
- в `recipients[].id` присутствует `user_id`, или
|
||||
- в `departments` есть отдел, где пользователь является активным сотрудником, или
|
||||
- пользователь является загрузившим документ (`uploaded_by.id == user_id`).
|
||||
|
||||
Args:
|
||||
items (Sequence[dict]): Полученный список документов из API.
|
||||
user_id (int|None): Идентификатор текущего пользователя.
|
||||
|
||||
Returns:
|
||||
list[dict]: Отфильтрованный список.
|
||||
|
||||
Raises:
|
||||
ValueError: Если user_id не задан (None).
|
||||
"""
|
||||
if user_id is None:
|
||||
raise ValueError("user_id is required for filtering")
|
||||
result: List[Dict[str, Any]] = []
|
||||
for it in items:
|
||||
if not isinstance(it, dict):
|
||||
continue
|
||||
|
||||
# 1. Документы для всех
|
||||
if it.get("sent_to_all"):
|
||||
result.append(it)
|
||||
continue
|
||||
|
||||
# 2. Документы, загруженные текущим пользователем
|
||||
uploaded_by = it.get("uploaded_by")
|
||||
if uploaded_by and isinstance(uploaded_by, dict):
|
||||
try:
|
||||
if int(uploaded_by.get("id")) == user_id:
|
||||
result.append(it)
|
||||
continue
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
# 3. Проверяем recipients
|
||||
recips = it.get("recipients") or []
|
||||
try:
|
||||
if any(int(r.get("id")) == user_id for r in recips if isinstance(r, dict) and r.get("id") is not None):
|
||||
result.append(it)
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 4. Проверяем departments
|
||||
depts = it.get("departments") or []
|
||||
try:
|
||||
# Если в документе есть отдел, нужно проверить членство
|
||||
# Но API возвращает только id и name отдела, без списка сотрудников
|
||||
# Поэтому нужно загрузить эту информацию отдельно или доверять backend'у
|
||||
# Для упрощения, если есть departments - считаем что пользователь может видеть
|
||||
# (реальная проверка происходит на backend при создании уведомлений)
|
||||
if depts:
|
||||
# TODO: добавить проверку членства в отделе через отдельный API запрос
|
||||
# Пока просто пропускаем - backend сам решит, нужно ли показывать
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
return result
|
||||
|
||||
|
||||
# ==============================
|
||||
# Вьюхи
|
||||
# ==============================
|
||||
|
||||
@method_decorator([login_required, require_api_auth, csrf_protect], name="dispatch")
|
||||
class DocumentView(LoginRequiredMixin, TemplateView):
|
||||
"""Страница списка документов с SSR + API для CRUD.
|
||||
|
||||
GET:
|
||||
- Загружает список документов через API (GET /api/v1/documents/).
|
||||
- Понимает фильтры: scope=mine|all, ack_status=acked|not_acked.
|
||||
- Для scope=mine загружает список ознакомленных документов.
|
||||
|
||||
Контекст шаблона:
|
||||
documents, acked_ids, next_url, prev_url, count, scope,
|
||||
show_admin_controls, can_manage_documents, filters, perms,
|
||||
api_document_list_url, api_document_detail_base.
|
||||
"""
|
||||
|
||||
template_name = "documents/document_list.html"
|
||||
|
||||
def get_context_data(self, **kwargs: Any) -> Dict[str, Any]:
|
||||
"""Формирует контекст страницы списка, фильтров и пагинации."""
|
||||
ctx = super().get_context_data(**kwargs)
|
||||
request = self.request
|
||||
user = request.user
|
||||
|
||||
# --- Определяем права ---
|
||||
can_manage = _user_can_manage_documents(user)
|
||||
show_admin_controls = can_manage
|
||||
|
||||
# --- Читаем фильтры из GET ---
|
||||
scope = (request.GET.get("scope") or "mine").strip().lower()
|
||||
if scope not in ("mine", "all"):
|
||||
scope = "mine"
|
||||
|
||||
# Если нет прав управления - принудительно mine
|
||||
if not can_manage:
|
||||
scope = "mine"
|
||||
|
||||
ack_status = (request.GET.get("ack_status") or "").strip()
|
||||
|
||||
# --- Формируем контекст без загрузки документов ---
|
||||
# Документы загружаются через AJAX
|
||||
ctx.update({
|
||||
"documents": [], # Пустой список - загрузка через JS
|
||||
"acked_ids": set(),
|
||||
"scope": scope,
|
||||
"show_admin_controls": show_admin_controls,
|
||||
"can_manage_documents": can_manage,
|
||||
"filters": {
|
||||
"ack_status": ack_status,
|
||||
},
|
||||
"perms": {
|
||||
"documents": {
|
||||
"add_document": (
|
||||
user.has_perm("documents.add_document") or
|
||||
user.is_staff
|
||||
),
|
||||
"change_document": (
|
||||
user.has_perm("documents.change_document") or
|
||||
user.is_staff
|
||||
),
|
||||
"delete_document": (
|
||||
user.has_perm("documents.delete_document") or
|
||||
user.is_staff
|
||||
),
|
||||
}
|
||||
},
|
||||
"api_document_list_url": reverse("api:v1:documents-list"),
|
||||
"api_document_detail_base": reverse("api:v1:documents-list"),
|
||||
})
|
||||
|
||||
return ctx
|
||||
|
||||
|
||||
@login_required
|
||||
@require_api_auth
|
||||
def acknowledge_document(request: HttpRequest, pk: int) -> HttpResponse:
|
||||
"""Отмечает документ «ознакомленным» через API и перенаправляет на файл.
|
||||
|
||||
Args:
|
||||
request (HttpRequest): Текущий запрос.
|
||||
pk (int): Идентификатор документа.
|
||||
|
||||
Returns:
|
||||
HttpResponse: Редирект на файл документа или список документов при ошибке.
|
||||
|
||||
Raises:
|
||||
ValueError: Если pk имеет неверный формат.
|
||||
"""
|
||||
try:
|
||||
doc_id = int(pk)
|
||||
except (TypeError, ValueError) as exc:
|
||||
messages.error(request, "Некорректный идентификатор документа.")
|
||||
raise ValueError("pk must be an integer") from exc
|
||||
|
||||
# Получаем документ для редиректа на файл
|
||||
from documents.models import Document
|
||||
try:
|
||||
doc = Document.objects.get(id=doc_id)
|
||||
file_url = doc.file.url if doc.file else None
|
||||
except Document.DoesNotExist:
|
||||
messages.error(request, "Документ не найден.")
|
||||
return redirect(reverse("documents:document_list"))
|
||||
|
||||
# Отмечаем ознакомление через API
|
||||
api = get_api_client(request)
|
||||
resp = api.post(f"v1/documents/{doc_id}/acknowledge/")
|
||||
ok, data, _status = _api_unpack(resp)
|
||||
if ok:
|
||||
if data.get("already"):
|
||||
messages.info(request, "Вы уже отмечали ознакомление с этим документом.")
|
||||
else:
|
||||
messages.success(request, "Ознакомление с документом отмечено.")
|
||||
else:
|
||||
messages.error(request, data.get("detail") or "Не удалось отметить документ.")
|
||||
|
||||
# Редирект на файл или обратно на список
|
||||
if file_url:
|
||||
return redirect(file_url)
|
||||
else:
|
||||
return redirect(reverse("documents:document_list"))
|
||||
@ -27,7 +27,6 @@ def serve_service_worker(request):
|
||||
|
||||
urlpatterns = [
|
||||
path("admin/", admin.site.urls),
|
||||
path("documents/", include("documents.urls", namespace="documents")),
|
||||
path("api/", include(("api.urls", "api"), namespace="api")),
|
||||
# Service Worker должен быть в корне для правильного scope
|
||||
path("sw.js", serve_service_worker, name="sw"),
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user