TelegramBotAdmin/wbb/services/message_cleanup.py

from __future__ import annotations

import asyncio
import logging
import re
from functools import wraps
from typing import Any
from weakref import WeakKeyDictionary

GROUP_INTERACTION_TTL_SECONDS = 10

_LOGGER = logging.getLogger(__name__)
_scheduled_messages: set[tuple[int, int]] = set()
_cleanup_tasks: set[asyncio.Task[Any]] = set()
_active_interactions: WeakKeyDictionary[asyncio.Task[Any], tuple[int, int]] = WeakKeyDictionary()
_TRACKED_SEND_METHODS = (
    "send_message",
    "send_photo",
    "send_document",
    "send_video",
    "send_animation",
    "send_audio",
    "send_voice",
    "send_video_note",
    "send_sticker",
    "send_contact",
    "send_location",
    "send_venue",
    "send_poll",
    "send_dice",
    "send_media_group",
    "send_cached_media",
    "copy_message",
    "copy_media_group",
    "forward_messages",
)


def _message_key(message: Any) -> tuple[int, int] | None:
    chat = getattr(message, "chat", None)
    message_id = getattr(message, "id", None)
    if not chat or not message_id:
        return None
    try:
        return int(chat.id), int(message_id)
    except (TypeError, ValueError):
        return None


def _is_group_message(message: Any) -> bool:
    chat = getattr(message, "chat", None)
    if not chat:
        return False
    chat_type = getattr(chat, "type", None)
    chat_type_value = str(getattr(chat_type, "value", chat_type) or "").lower()
    if chat_type_value:
        return chat_type_value in {"group", "supergroup"}
    try:
        return int(chat.id) < 0
    except (TypeError, ValueError):
        return False


def _current_task() -> asyncio.Task[Any] | None:
    try:
        return asyncio.current_task()
    except RuntimeError:
        return None


async def _delete_later(message: Any, delay_seconds: int) -> None:
    key = _message_key(message)
    if key is None:
        return
    try:
        await asyncio.sleep(max(1, int(delay_seconds)))
        try:
            await message.delete(revoke=True)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            _LOGGER.warning(
                "群交互消息自动删除失败 chat_id=%s message_id=%s: %s",
                key[0],
                key[1],
                exc,
            )
    finally:
        _scheduled_messages.discard(key)


def schedule_group_cleanup(
    *messages: Any,
    delay_seconds: int = GROUP_INTERACTION_TTL_SECONDS,
) -> None:
    for message in messages:
        key = _message_key(message)
        if key is None or not _is_group_message(message):
            continue
        if key in _scheduled_messages:
            continue
        _scheduled_messages.add(key)
        task = asyncio.create_task(
            _delete_later(message, delay_seconds),
            name=f"group-interaction-cleanup:{key[0]}:{key[1]}",
        )
        _cleanup_tasks.add(task)
        task.add_done_callback(_cleanup_tasks.discard)


async def reply_group_interaction(
    message: Any,
    text: str,
    **kwargs: Any,
) -> Any:
    """Reply to a group interaction and explicitly clean up both messages."""
    response = await message.reply(text, **kwargs)
    schedule_group_cleanup(message, response)
    return response


def is_group_cleanup_scheduled(message: Any) -> bool:
    key = _message_key(message)
    return bool(key and key in _scheduled_messages)


def clear_current_group_interaction() -> None:
    task = _current_task()
    if task is not None:
        _active_interactions.pop(task, None)


def is_group_interaction_candidate(
    message: Any, *, menu_texts: set[str] | frozenset[str] = frozenset()
) -> bool:
    if not _is_group_message(message):
        return False
    text = str(
        getattr(message, "text", None) or getattr(message, "caption", None) or ""
    ).strip()
    folded = text.casefold()
    if text.startswith("/") or folded in {"@admin", "@admins"}:
        return True
    if folded.startswith(("@admin ", "@admins ")) or text in menu_texts:
        return True
    return bool(
        re.fullmatch(
            r"(?:抽奖|开奖|中奖名单|签到|积分|积分排行|流水|封禁|解封|禁言|解除禁言|禁言\d+分钟|警告|删除消息|删除|删|踢出|踢|飞)",
            text,
        )
    )


def begin_group_message_interaction(
    message: Any,
    *,
    menu_texts: set[str] | frozenset[str] = frozenset(),
    delay_seconds: int = GROUP_INTERACTION_TTL_SECONDS,
) -> bool:
    clear_current_group_interaction()
    if not is_group_interaction_candidate(message, menu_texts=menu_texts):
        return False
    task = _current_task()
    if task is None:
        return False
    _active_interactions[task] = (int(message.chat.id), delay_seconds)
    schedule_group_cleanup(message, delay_seconds=delay_seconds)
    return True


def begin_group_callback_interaction(
    query: Any, *, delay_seconds: int = GROUP_INTERACTION_TTL_SECONDS
) -> bool:
    clear_current_group_interaction()
    message = getattr(query, "message", None)
    if (
        not message
        or not _is_group_message(message)
        or not is_group_cleanup_scheduled(message)
    ):
        return False
    task = _current_task()
    if task is None:
        return False
    _active_interactions[task] = (int(message.chat.id), delay_seconds)
    return True


def track_current_group_interaction(
    result: Any,
    *,
    reply_to_message_id: Any = None,
) -> None:
    if reply_to_message_id is None:
        return
    task = _current_task()
    if task is None:
        return
    context = _active_interactions.get(task)
    if context is None:
        return
    chat_id, delay_seconds = context
    messages = result if isinstance(result, (list, tuple)) else (result,)
    for message in messages:
        key = _message_key(message)
        if key and key[0] == chat_id and _is_group_message(message):
            schedule_group_cleanup(message, delay_seconds=delay_seconds)


def install_group_interaction_tracking(client: Any) -> None:
    if getattr(client, "_wbb_group_interaction_tracking", False):
        return
    for method_name in _TRACKED_SEND_METHODS:
        original = getattr(client, method_name, None)
        if not callable(original):
            continue

        @wraps(original)
        async def tracked_send(*args: Any, _original=original, **kwargs: Any):
            result = await _original(*args, **kwargs)
            track_current_group_interaction(
                result,
                reply_to_message_id=kwargs.get("reply_to_message_id"),
            )
            return result

        setattr(client, method_name, tracked_send)
    client._wbb_group_interaction_tracking = True