karkas/src/karkas_blocks/karkas_blocks/standard/statistics/main.py

from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict

from aiogram import BaseMiddleware

from .db.tables import ChatStats, Messages, UserStats

if TYPE_CHECKING:
    from aiogram.types import Message, TelegramObject


async def update_chat_stats(event: "Message"):
    await ChatStats.insert(
        ChatStats(chat_id=event.chat.id, date=event.date, messages_count=1)
    ).on_conflict(
        action="DO UPDATE",
        values=[
            ChatStats.date,
            (ChatStats.messages_count, ChatStats.messages_count + 1),
        ],
    ).run()


async def update_user_stats(event: "Message"):
    await UserStats.insert(
        UserStats(
            key=f"{event.chat.id}-{event.from_user.id}",
            chat_id=event.chat.id,
            user_id=event.from_user.id,
            date=event.date,
            messages_count=1,
        )
    ).on_conflict(
        action="DO UPDATE",
        values=[
            UserStats.date,
            (UserStats.messages_count, UserStats.messages_count + 1),
        ],
    ).run()


async def save_messages(event: "Message"):
    await Messages.insert(
        Messages(
            key=f"{event.chat.id}-{event.message_id}",
            chat_id=event.chat.id,
            message_id=event.message_id,
            sender_id=event.from_user.id,
            answer_to_message_id=(
                event.reply_to_message.message_id if event.reply_to_message else None
            ),
            message_text=event.text,
        )
    )


class StatisticsMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[["TelegramObject", Dict[str, Any]], Awaitable[Any]],
        event: "TelegramObject",
        data: Dict[str, Any],
    ) -> Any:

        await update_chat_stats(event)
        await update_user_stats(event)
        await save_messages(event)

        result = await handler(event, data)

        return result