zhenxun_bot/zhenxun/utils/platform.py

from typing import Awaitable, Callable, Literal, Set

import nonebot
from nonebot.adapters import Bot
from nonebot.adapters.discord import Bot as DiscordBot
from nonebot.adapters.dodo import Bot as DodoBot
from nonebot.adapters.kaiheila import Bot as KaiheilaBot
from nonebot.adapters.onebot.v11 import Bot as v11Bot
from nonebot.adapters.onebot.v12 import Bot as v12Bot
from nonebot.utils import is_coroutine_callable
from nonebot_plugin_saa import (
    MessageFactory,
    TargetDoDoChannel,
    TargetKaiheilaChannel,
    TargetQQGroup,
    Text,
)

from zhenxun.models.friend_user import FriendUser
from zhenxun.models.group_console import GroupConsole
from zhenxun.services.log import logger


class PlatformManage:

    @classmethod
    async def update_group(cls, bot: Bot) -> int:
        """更新群组信息

        参数:
            bot: Bot

        返回:
            int: 更新个数
        """
        create_list = []
        group_list, platform = await cls.get_group_list(bot)
        if group_list:
            exists_group_list = await GroupConsole.all().values_list(
                "group_id", "channel_id"
            )
            for group in group_list:
                group.platform = platform
                if (group.group_id, group.channel_id) not in exists_group_list:
                    create_list.append(group)
                    logger.debug(
                        "群聊信息更新成功",
                        "更新群信息",
                        target=f"{group.group_id}:{group.channel_id}",
                    )
        if create_list:
            await GroupConsole.bulk_create(create_list, 10)
        return len(create_list)

    @classmethod
    def get_platform(cls, bot: Bot) -> str | None:
        """获取平台

        参数:
            bot: Bot

        返回:
            str | None: 平台
        """
        if isinstance(bot, (v11Bot, v12Bot)):
            return "qq"
        if isinstance(bot, DodoBot):
            return "dodo"
        if isinstance(bot, KaiheilaBot):
            return "kaiheila"
        return None

    @classmethod
    async def get_group_list(cls, bot: Bot) -> tuple[list[GroupConsole], str]:
        """获取群组列表

        参数:
            bot: Bot

        返回:
            list[GroupConsole]: 群组列表
        """
        if isinstance(bot, v11Bot):
            group_list = await bot.get_group_list()
            return [
                GroupConsole(
                    group_id=str(g["group_id"]),
                    group_name=g["group_name"],
                    max_member_count=g["max_member_count"],
                    member_count=g["member_count"],
                )
                for g in group_list
            ], "qq"
        if isinstance(bot, v12Bot):
            group_list = await bot.get_group_list()
            return [
                GroupConsole(
                    group_id=g.group_id,  # type: ignore
                    user_name=g.group_name,  # type: ignore
                )
                for g in group_list
            ], "qq"
        if isinstance(bot, DodoBot):
            island_list = await bot.get_island_list()
            source_id_list = [
                (g.island_source_id, g.island_name)
                for g in island_list
                if g.island_source_id
            ]
            group_list = []
            for id, name in source_id_list:
                channel_list = await bot.get_channel_list(island_source_id=id)
                group_list.append(GroupConsole(group_id=id, group_name=name))
                group_list += [
                    GroupConsole(
                        group_id=id, group_name=c.channel_name, channel_id=c.channel_id
                    )
                    for c in channel_list
                ]
            return group_list, "dodo"
        if isinstance(bot, KaiheilaBot):
            group_list = []
            guilds = await bot.guild_list()
            if guilds.guilds:
                for guild_id, name in [(g.id_, g.name) for g in guilds.guilds if g.id_]:
                    view = await bot.guild_view(guild_id=guild_id)
                    group_list.append(GroupConsole(group_id=guild_id, group_name=name))
                    if view.channels:
                        group_list += [
                            GroupConsole(
                                group_id=guild_id, group_name=c.name, channel_id=c.id_
                            )
                            for c in view.channels
                            if c.type != 0
                        ]
            return group_list, "kaiheila"
        if isinstance(bot, DiscordBot):
            # TODO: discord群组列表
            pass
        return [], ""

    @classmethod
    async def update_friend(cls, bot: Bot) -> int:
        """更新好友信息

        参数:
            bot: Bot

        返回:
            int: 更新个数
        """
        create_list = []
        friend_list, platform = await cls.get_friend_list(bot)
        if friend_list:
            user_id_list = await FriendUser.all().values_list("user_id", flat=True)
            for friend in friend_list:
                friend.platform = platform
                if friend.user_id not in user_id_list:
                    create_list.append(friend)
        if create_list:
            await FriendUser.bulk_create(create_list, 10)
        return len(create_list)

    @classmethod
    async def get_friend_list(cls, bot: Bot) -> tuple[list[FriendUser], str]:
        """获取好友列表

        参数:
            bot: Bot

        返回:
            list[FriendUser]: 好友列表
        """
        if isinstance(bot, v11Bot):
            friend_list = await bot.get_friend_list()
            return [
                FriendUser(user_id=str(f["user_id"]), user_name=f["nickname"])
                for f in friend_list
            ], "qq"
        if isinstance(bot, v12Bot):
            friend_list = await bot.get_friend_list()
            return [
                FriendUser(
                    user_id=f.user_id,  # type: ignore
                    user_name=f.user_displayname or f.user_remark or f.user_name,  # type: ignore
                )
                for f in friend_list
            ], "qq"
        if isinstance(bot, DodoBot):
            # TODO: dodo好友列表
            pass
        if isinstance(bot, KaiheilaBot):
            # TODO: kaiheila好友列表
            pass
        if isinstance(bot, DiscordBot):
            # TODO: discord好友列表
            pass
        return [], ""

    @classmethod
    def get_target(cls, bot: Bot, group_id: str | None, channel_id: str | None):
        """获取发生Target

        参数:
            bot: Bot
            group_id: 群组id
            channel_id: 频道id

        返回:
            target: 对应平台Target
        """
        target = None
        if isinstance(bot, (v11Bot, v12Bot)):
            if group_id:
                target = TargetQQGroup(group_id=int(group_id))
        if channel_id:
            if isinstance(bot, DodoBot):
                target = TargetDoDoChannel(channel_id=channel_id)
            elif isinstance(bot, KaiheilaBot):
                target = TargetKaiheilaChannel(channel_id=channel_id)
        return target


async def broadcast_group(
    message: str | MessageFactory,
    bot: Bot | list[Bot] | None = None,
    bot_id: str | Set[str] | None = None,
    ignore_group: Set[int] | None = None,
    check_func: Callable[[str, str | None], Awaitable] | None = None,
    log_cmd: str | None = None,
    platform: Literal["qq", "dodo", "kaiheila"] | None = None,
):
    """获取所有Bot或指定Bot对象广播群聊

    Args:
        message: 广播消息内容
        bot: 指定bot对象.
        bot_id: 指定bot id.
        ignore_group: 忽略群聊列表.
        check_func: 发送前对群聊检测方法，判断是否发送.
        log_cmd: 日志标记.
        platform: 指定平台
    """
    if platform and platform not in ["qq", "dodo", "kaiheila"]:
        raise ValueError("指定平台不支持")
    if not message:
        raise ValueError("群聊广播消息不能为空")
    bot_dict = nonebot.get_bots()
    bot_list: list[Bot] = []
    if bot:
        if isinstance(bot, list):
            bot_list = bot
        else:
            bot_list.append(bot)
    elif bot_id:
        _bot_id_list = bot_id
        if isinstance(bot_id, str):
            _bot_id_list = [bot_id]
        for id_ in _bot_id_list:
            if bot_id in bot_dict:
                bot_list.append(bot_dict[bot_id])
            else:
                logger.warning(f"Bot:{id_} 对象未连接或不存在")
    else:
        bot_list = list(bot_dict.values())
    _used_group = []
    for _bot in bot_list:
        try:
            if platform and platform != PlatformManage.get_platform(_bot):
                continue
            group_list, _ = await PlatformManage.get_group_list(_bot)
            if group_list:
                for group in group_list:
                    key = f"{group.group_id}:{group.channel_id}"
                    try:
                        if (
                            ignore_group
                            and (
                                group.group_id in ignore_group
                                or group.channel_id in ignore_group
                            )
                        ) or key in _used_group:
                            continue
                        is_continue = False
                        if check_func:
                            if is_coroutine_callable(check_func):
                                is_continue = not await check_func(
                                    group.group_id, group.channel_id
                                )
                            else:
                                is_continue = not check_func(
                                    group.group_id, group.channel_id
                                )
                        if is_continue:
                            continue
                        target = PlatformManage.get_target(
                            _bot, group.group_id, group.channel_id
                        )
                        if target:
                            _used_group.append(key)
                            message_list = message
                            if isinstance(message, str):
                                message_list = MessageFactory([Text(message)])
                            await MessageFactory(message_list).send_to(target, _bot)
                            logger.debug("发送成功", log_cmd, target=key)
                        else:
                            logger.warning("target为空", log_cmd, target=key)
                    except Exception as e:
                        logger.error("发送失败", log_cmd, target=key, e=e)
        except Exception as e:
            logger.error(f"Bot: {_bot.self_id} 获取群聊列表失败", command=log_cmd, e=e)