Corp/backend/communications/mentions.py

import json
import re
import unicodedata
from collections.abc import Iterable

from django.contrib.auth import get_user_model
from django.db.models import F, Q, QuerySet

User = get_user_model()

MAX_MENTIONS_PER_MESSAGE = 50


def mention_display_name(user) -> str:
    full_name = " ".join(
        part.strip()
        for part in (user.last_name, user.first_name, user.patronymic)
        if part and part.strip()
    )
    return (full_name or user.username or user.email or str(user.pk)).strip()


def _normalize_search_value(value) -> str:
    return unicodedata.normalize("NFKC", str(value or "")).casefold()


def filter_mentionable_users(users: Iterable, query: str) -> list:
    """Match every query token against any employee identity field."""
    tokens = [
        _normalize_search_value(token) for token in query.split() if token.strip()
    ]
    if not tokens:
        return list(users)

    matches = []
    for user in users:
        fields = (
            _normalize_search_value(user.last_name),
            _normalize_search_value(user.first_name),
            _normalize_search_value(user.patronymic),
            _normalize_search_value(user.email),
            _normalize_search_value(user.username),
        )
        if all(any(token in field for field in fields) for token in tokens):
            matches.append(user)
    return matches


def get_mentionable_users(chat, *, author=None) -> QuerySet:
    """Return active users who can see the chat, including legacy memberships."""
    if chat.type == "global" or chat.include_all_users:
        users = User.objects.visible_in_ui().filter(is_active=True)
    else:
        users = (
            User.objects.visible_in_ui()
            .filter(
                Q(chat_memberships__chat=chat, chat_memberships__is_active=True)
                | Q(chats=chat)
            )
            .filter(is_active=True)
            .distinct()
        )

    if author is not None:
        users = users.exclude(pk=author.pk)
    return users


def parse_mentioned_user_ids(raw_value) -> list[int]:
    if raw_value in (None, ""):
        return []

    values = raw_value
    if isinstance(raw_value, str):
        try:
            values = json.loads(raw_value)
        except json.JSONDecodeError:
            values = raw_value.split(",")
    if not isinstance(values, Iterable) or isinstance(values, (str, bytes, dict)):
        values = [values]

    result = []
    seen = set()
    for value in values:
        try:
            user_id = int(value)
        except (TypeError, ValueError):
            continue
        if user_id <= 0 or user_id in seen:
            continue
        seen.add(user_id)
        result.append(user_id)
        if len(result) >= MAX_MENTIONS_PER_MESSAGE:
            break
    return result


def build_message_mentions(chat, author, content: str, user_ids) -> list[dict]:
    """Validate selected users and resolve their visible mention ranges."""
    normalized_ids = parse_mentioned_user_ids(user_ids)
    if not content or not normalized_ids:
        return []

    users_by_id = {
        user.id: user
        for user in get_mentionable_users(chat, author=author).filter(
            id__in=normalized_ids
        )
    }
    mentions = []
    occupied_ranges: list[tuple[int, int]] = []

    for user_id in normalized_ids:
        user = users_by_id.get(user_id)
        if user is None:
            continue
        label = mention_display_name(user)
        if not label:
            continue
        pattern = re.compile(
            rf"(?<![\w@])@{re.escape(label)}(?=$|[\s.,!?;:)\]}}])",
            flags=re.IGNORECASE,
        )
        for match in pattern.finditer(content):
            start, end = match.span()
            if any(
                start < used_end and end > used_start
                for used_start, used_end in occupied_ranges
            ):
                continue
            occupied_ranges.append((start, end))
            mentions.append(
                {
                    "user_id": user.id,
                    "name": label,
                    "start": start,
                    "end": end,
                }
            )

    return sorted(mentions, key=lambda mention: mention["start"])


def mentioned_user_ids(mentions) -> set[int]:
    result = set()
    for mention in mentions or []:
        try:
            result.add(int(mention.get("user_id")))
        except (AttributeError, TypeError, ValueError):
            continue
    return result


def update_unread_mention_counters(
    message,
    *,
    added_user_ids=(),
    removed_user_ids=(),
) -> None:
    """Keep per-chat mention badges correct when a message is edited."""
    from .models import ChatReadState

    added_ids = set(parse_mentioned_user_ids(added_user_ids))
    removed_ids = set(parse_mentioned_user_ids(removed_user_ids)) - added_ids

    if removed_ids:
        ChatReadState.objects.filter(
            chat_id=message.chat_id,
            user_id__in=removed_ids,
            unread_mention_count__gt=0,
        ).update(unread_mention_count=F("unread_mention_count") - 1)

    if not added_ids:
        return

    existing_ids = set(
        ChatReadState.objects.filter(
            chat_id=message.chat_id,
            user_id__in=added_ids,
        ).values_list("user_id", flat=True)
    )
    if existing_ids:
        ChatReadState.objects.filter(
            chat_id=message.chat_id,
            user_id__in=existing_ids,
        ).update(unread_mention_count=F("unread_mention_count") + 1)

    missing_ids = added_ids - existing_ids
    if missing_ids:
        ChatReadState.objects.bulk_create(
            [
                ChatReadState(
                    chat_id=message.chat_id,
                    user_id=user_id,
                    unread_mention_count=1,
                )
                for user_id in missing_ids
            ],
            ignore_conflicts=True,
        )