149 lines
4.8 KiB
Python
149 lines
4.8 KiB
Python
"""
|
||
Утилиты для модуля communications.
|
||
Переиспользуемые функции для API, WebSocket и других компонентов.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
import logging
|
||
|
||
from django.utils import timezone
|
||
from django.utils.dateparse import parse_datetime
|
||
|
||
from .models import Chat, ChatMembership
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _coerce_ts(val: str | None) -> datetime.datetime:
|
||
"""
|
||
Принимает миллисекунды/секунды с эпохи или ISO-дату.
|
||
Возвращает aware-дату (UTC). Фоллбек — timezone.now().
|
||
|
||
Args:
|
||
val: Строка с timestamp (ms/sec) или ISO-дата
|
||
|
||
Returns:
|
||
datetime: Aware datetime объект в UTC
|
||
|
||
Examples:
|
||
>>> _coerce_ts("1640000000000") # миллисекунды
|
||
datetime.datetime(2021, 12, 20, 13, 33, 20, tzinfo=UTC)
|
||
|
||
>>> _coerce_ts("1640000000") # секунды
|
||
datetime.datetime(2021, 12, 20, 13, 33, 20, tzinfo=UTC)
|
||
|
||
>>> _coerce_ts("2021-12-20T13:33:20Z") # ISO
|
||
datetime.datetime(2021, 12, 20, 13, 33, 20, tzinfo=UTC)
|
||
"""
|
||
if not val:
|
||
return timezone.now()
|
||
|
||
# Сначала пробуем число (sec/ms)
|
||
try:
|
||
iv = int(val)
|
||
if iv > 10**12: # пришли миллисекунды
|
||
iv //= 1000
|
||
return datetime.datetime.fromtimestamp(iv, tz=datetime.UTC)
|
||
except (OSError, OverflowError, TypeError, ValueError):
|
||
pass
|
||
|
||
# Потом пробуем ISO-строку
|
||
d = parse_datetime(val)
|
||
if d is None:
|
||
return timezone.now()
|
||
if timezone.is_naive(d):
|
||
return timezone.make_aware(d, timezone=timezone.get_current_timezone())
|
||
return d
|
||
|
||
|
||
def _membership_or_legacy_participant(chat: Chat, user) -> bool:
|
||
membership_state = (
|
||
ChatMembership.objects.filter(chat=chat, user=user)
|
||
.values_list("is_active", flat=True)
|
||
.first()
|
||
)
|
||
if membership_state is not None:
|
||
return membership_state
|
||
return chat.participants.filter(pk=user.pk).exists()
|
||
|
||
|
||
def _context_participant_access(chat: Chat, user) -> bool | None:
|
||
try:
|
||
return chat.get_participants().filter(pk=user.pk).exists()
|
||
except Exception as error: # noqa: BLE001 - configurable resolver boundary
|
||
logger.warning(
|
||
"Chat access participant resolution failed",
|
||
extra={
|
||
"event": "chat.access.participant_resolution_failed",
|
||
"chat_id": chat.id,
|
||
"user_id": user.id,
|
||
"chat_type": chat.type,
|
||
"error_type": type(error).__name__,
|
||
},
|
||
exc_info=True,
|
||
)
|
||
return None
|
||
|
||
|
||
def _group_chat_access(chat: Chat, user) -> bool:
|
||
if chat.context_object_id:
|
||
resolved = _context_participant_access(chat, user)
|
||
if resolved is not None:
|
||
return resolved
|
||
return _membership_or_legacy_participant(chat, user)
|
||
|
||
|
||
def _comments_chat_access(chat: Chat, user) -> bool:
|
||
if (chat.flags or {}).get("show_in_messages") is True:
|
||
return True
|
||
if chat.context_object_id:
|
||
return _context_participant_access(chat, user) is True
|
||
return _membership_or_legacy_participant(chat, user)
|
||
|
||
|
||
_CHAT_ACCESS_CHECKERS = {
|
||
"private": _membership_or_legacy_participant,
|
||
"group": _group_chat_access,
|
||
"channel": _membership_or_legacy_participant,
|
||
"announcement": _membership_or_legacy_participant,
|
||
"comments": _comments_chat_access,
|
||
}
|
||
|
||
|
||
def user_can_access_chat(chat: Chat, user) -> bool:
|
||
"""
|
||
Проверяет, имеет ли пользователь доступ к чату.
|
||
|
||
Правила доступа по типам чатов:
|
||
- global: доступ всем
|
||
- private: активное ChatMembership
|
||
- group: активное ChatMembership (или context-based resolver)
|
||
- channel/announcement: include_all_users или активное ChatMembership
|
||
- comments: context-based доступ через get_participants()
|
||
|
||
Args:
|
||
chat: Объект Chat
|
||
user: Объект User
|
||
|
||
Returns:
|
||
bool: True если пользователь имеет доступ
|
||
|
||
Note:
|
||
Использует прямые запросы к БД в обход prefetch cache
|
||
"""
|
||
if not chat or not user:
|
||
return False
|
||
|
||
# Создатель является владельцем ресурса и не должен терять доступ из-за
|
||
# рассинхронизации legacy participants/ChatMembership.
|
||
if chat.created_by_id == user.pk:
|
||
return True
|
||
|
||
if chat.type == "global" or chat.include_all_users:
|
||
return True
|
||
|
||
checker = _CHAT_ACCESS_CHECKERS.get(chat.type)
|
||
return checker(chat, user) if checker else False
|