TelegramBotAdmin/wbb/services/member_identity.py

from __future__ import annotations

import asyncio
from collections import defaultdict
from typing import Any

from wbb import app, log
from wbb.utils.dbadmin import (
    get_managed_chat_settings,
    get_recent_chat_member,
    record_member_identity_change,
    upsert_recent_chat_member,
)
from wbb.utils.dbpoints import update_point_account_identity

_identity_locks: defaultdict[tuple[int, int], asyncio.Lock] = defaultdict(asyncio.Lock)


def display_name(first_name: str | None, last_name: str | None) -> str:
    return " ".join(value.strip() for value in (first_name, last_name) if value and value.strip())


def user_display_name(user: Any) -> str:
    return (
        display_name(getattr(user, "first_name", None), getattr(user, "last_name", None))
        or (
            f"@{user.username}"
            if getattr(user, "username", None)
            else str(user.id)
        )
    )


def _snapshot(*, username: str | None, first_name: str | None, last_name: str | None) -> dict[str, Any]:
    return {
        "username": username,
        "first_name": first_name,
        "last_name": last_name,
        "display_name": display_name(first_name, last_name),
    }


def _username_label(value: str | None) -> str:
    return f"@{value}" if value else "未设置"


async def observe_member_identity(*, chat_id: int, user: Any) -> dict[str, Any] | None:
    user_id = int(user.id)
    username = getattr(user, "username", None)
    first_name = getattr(user, "first_name", None)
    last_name = getattr(user, "last_name", None)
    is_bot = bool(getattr(user, "is_bot", False))

    async with _identity_locks[(int(chat_id), user_id)]:
        previous = await get_recent_chat_member(chat_id, user_id)
        await upsert_recent_chat_member(
            chat_id=chat_id,
            user_id=user_id,
            username=username,
            first_name=first_name,
            last_name=last_name,
            is_bot=is_bot,
        )
        await update_point_account_identity(
            chat_id=chat_id,
            user_id=user_id,
            username=username,
            first_name=first_name,
            last_name=last_name,
            display_name=display_name(first_name, last_name),
        )
        if not previous or is_bot:
            return None

        settings = await get_managed_chat_settings(chat_id)
        monitor = settings.get("identity_monitor") or {}
        if not monitor.get("enabled"):
            return None

        before = _snapshot(
            username=previous.get("username"),
            first_name=previous.get("first_name"),
            last_name=previous.get("last_name"),
        )
        after = _snapshot(
            username=username,
            first_name=first_name,
            last_name=last_name,
        )
        changed_fields = [
            field
            for field in ("username", "display_name")
            if before[field] != after[field]
        ]
        if not changed_fields:
            return None

        event = await record_member_identity_change(
            chat_id=chat_id,
            user_id=user_id,
            before=before,
            after=after,
            changed_fields=changed_fields,
        )
        if monitor.get("notify_in_chat"):
            details = []
            if "display_name" in changed_fields:
                details.append(
                    f"昵称：{before['display_name'] or '未设置'} -> "
                    f"{after['display_name'] or '未设置'}"
                )
            if "username" in changed_fields:
                details.append(
                    f"用户名：{_username_label(before['username'])} -> "
                    f"{_username_label(after['username'])}"
                )
            try:
                await app.send_message(
                    chat_id,
                    "成员资料发生变化\n"
                    f"用户 ID：{user_id}\n"
                    + "\n".join(details),
                )
            except Exception as exc:
                log.error(f"群 {chat_id} 成员 {user_id} 资料变更提醒发送失败：{exc}")
        return event


async def observe_message_member_identities(
    *, chat_id: int, message: Any
) -> list[dict[str, Any]]:
    users: dict[int, Any] = {}
    sender = getattr(message, "from_user", None)
    if sender:
        users[int(sender.id)] = sender
    for member in getattr(message, "new_chat_members", None) or []:
        users[int(member.id)] = member

    changes: list[dict[str, Any]] = []
    for user in users.values():
        event = await observe_member_identity(chat_id=chat_id, user=user)
        if event:
            changes.append(event)
    return changes