LiteyukiBot/liteyuki/plugins/liteyuki_statistics/stat_monitors.py

import time

from nonebot import require
from nonebot.message import event_postprocessor

from liteyuki.utils.base.data import Database, LiteModel
from liteyuki.utils.base.ly_typing import v11, satori

from liteyuki.utils.base.ly_typing import T_Bot, T_MessageEvent

from .common import MessageEventModel, msg_db
from ...utils import satori_utils

require("nonebot_plugin_alconna")


@event_postprocessor
async def general_event_monitor(bot: T_Bot, event: T_MessageEvent):
    if isinstance(bot, satori.Bot):
        return await satori_event_monitor(bot, event)
    else:
        return await onebot_v11_event_monitor(bot, event)


async def onebot_v11_event_monitor(bot: v11.Bot, event: v11.MessageEvent):
    if event.message_type == "group":
        event: v11.GroupMessageEvent
        group_id = str(event.group_id)
    else:
        group_id = ""

    mem = MessageEventModel(
        time=int(time.time()),
        bot_id=bot.self_id,
        adapter="onebot.v11",
        group_id=group_id,
        user_id=str(event.user_id),

        message_id=str(event.message_id),

        message=event.message,
        message_text=event.raw_message,
        message_type=event.message_type,
    )
    msg_db.save(mem)


async def satori_event_monitor(bot: satori.Bot, event: satori.MessageEvent):
    if event.guild is not None:
        event: satori.MessageEvent
        group_id = str(event.guild.id)
    else:
        group_id = ""

    mem = MessageEventModel(
        time=int(time.time()),
        bot_id=bot.self_id,
        adapter="satori",
        group_id=group_id,
        user_id=str(event.user.id),

        message_id=str(event.message.id),

        message=event.message,
        message_text=event.message.content,
        message_type=satori_utils.get_message_type(event),
    )
    msg_db.save(mem)