zhenxun_bot/zhenxun/utils/platform.py

import random
from typing import Awaitable, Callable, Literal, Set

import httpx
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_alconna.uniseg import Receipt, Target, UniMessage
from pydantic import BaseModel

from zhenxun.models.friend_user import FriendUser
from zhenxun.models.group_console import GroupConsole
from zhenxun.services.log import logger
from zhenxun.utils.exception import NotFindSuperuser
from zhenxun.utils.message import MessageUtils


class UserData(BaseModel):

    name: str
    """昵称"""
    card: str | None = None
    """名片/备注"""
    user_id: str
    """用户id"""
    group_id: str | None = None
    """群组id"""
    role: str | None = None
    """角色"""
    avatar_url: str | None = None
    """头像url"""
    join_time: int | None = None
    """加入时间"""


class PlatformUtils:

    @classmethod
    async def ban_user(cls, bot: Bot, user_id: str, group_id: str, duration: int):
        """禁言

        参数:
            bot: Bot
            user_id: 用户id
            group_id: 群组id
            duration: 禁言时长(分钟)
        """
        if isinstance(bot, v11Bot):
            await bot.set_group_ban(
                group_id=int(group_id),
                user_id=int(user_id),
                duration=duration * 60,
            )

    @classmethod
    async def send_superuser(
        cls,
        bot: Bot,
        message: UniMessage | str,
        superuser_id: str | None = None,
    ) -> Receipt | None:
        """发送消息给超级用户

        参数:
            bot: Bot
            message: 消息
            superuser_id: 指定超级用户id.

        异常:
            NotFindSuperuser: 未找到超级用户id

        返回:
            Receipt | None: Receipt
        """
        if not superuser_id:
            platform = cls.get_platform(bot)
            platform_superusers = bot.config.PLATFORM_SUPERUSERS.get(platform) or []
            if not platform_superusers:
                raise NotFindSuperuser()
            superuser_id = random.choice(platform_superusers)
        if isinstance(message, str):
            message = MessageUtils.build_message(message)
        return await cls.send_message(bot, superuser_id, None, message)

    @classmethod
    async def get_group_member_list(cls, bot: Bot, group_id: str) -> list[UserData]:
        """获取群组/频道成员列表

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

        返回:
            list[UserData]: 用户数据列表
        """
        if isinstance(bot, v11Bot):
            if member_list := await bot.get_group_member_list(group_id=int(group_id)):
                return [
                    UserData(
                        name=user["nickname"],
                        card=user["card"],
                        user_id=user["user_id"],
                        group_id=user["group_id"],
                        role=user["role"],
                        join_time=user["join_time"],
                    )
                    for user in member_list
                ]
        if isinstance(bot, v12Bot):
            if member_list := await bot.get_group_member_list(group_id=group_id):
                return [
                    UserData(
                        name=user["user_name"],
                        card=user["user_displayname"],
                        user_id=user["user_id"],
                        group_id=group_id,
                    )
                    for user in member_list
                ]
        if isinstance(bot, DodoBot):
            if result_data := await bot.get_member_list(
                island_source_id=group_id, page_size=100, max_id=0
            ):
                max_id = result_data.max_id
                result_list = result_data.list
                data_list = []
                while max_id == 100:
                    result_data = await bot.get_member_list(
                        island_source_id=group_id, page_size=100, max_id=0
                    )
                    result_list += result_data.list
                    max_id = result_data.max_id
                for user in result_list:
                    data_list.append(
                        UserData(
                            name=user.nick_name,
                            card=user.personal_nick_name,
                            avatar_url=user.avatar_url,
                            user_id=user.dodo_source_id,
                            group_id=user.island_source_id,
                            join_time=int(user.join_time.timestamp()),
                        )
                    )
                return data_list
        if isinstance(bot, KaiheilaBot):
            if result_data := await bot.guild_userList(guild_id=group_id):
                if result_data.users:
                    data_list = []
                    for user in result_data.users:
                        second = None
                        if user.joined_at:
                            second = int(user.joined_at / 1000)
                        data_list.append(
                            UserData(
                                name=user.nickname or "",
                                avatar_url=user.avatar,
                                user_id=user.id_,  # type: ignore
                                group_id=group_id,
                                join_time=second,
                            )
                        )
                    return data_list
        if isinstance(bot, DiscordBot):
            # TODO: discord获取用户
            pass
        return []

    @classmethod
    async def get_user(
        cls, bot: Bot, user_id: str, group_id: str | None = None
    ) -> UserData | None:
        """获取用户信息

        参数:
            bot: Bot
            user_id: 用户id
            group_id: 群组/频道id.

        返回:
            UserData | None: 用户数据
        """
        if isinstance(bot, v11Bot):
            if group_id:
                if user := await bot.get_group_member_info(
                    group_id=int(group_id), user_id=int(user_id)
                ):
                    return UserData(
                        name=user["nickname"],
                        card=user["card"],
                        user_id=user["user_id"],
                        group_id=user["group_id"],
                        role=user["role"],
                        join_time=user["join_time"],
                    )
            else:
                if friend_list := await bot.get_friend_list():
                    for f in friend_list:
                        if f["user_id"] == int(user_id):
                            return UserData(
                                name=f["nickname"],
                                card=f["remark"],
                                user_id=f["user_id"],
                            )
        if isinstance(bot, v12Bot):
            if group_id:
                if user := await bot.get_group_member_info(
                    group_id=group_id, user_id=user_id
                ):
                    return UserData(
                        name=user["user_name"],
                        card=user["user_displayname"],
                        user_id=user["user_id"],
                        group_id=group_id,
                    )
            else:
                if friend_list := await bot.get_friend_list():
                    for f in friend_list:
                        if f["user_id"] == int(user_id):
                            return UserData(
                                name=f["user_name"],
                                card=f["user_remark"],
                                user_id=f["user_id"],
                            )
        if isinstance(bot, DodoBot):
            if group_id:
                if user := await bot.get_member_info(
                    island_source_id=group_id, dodo_source_id=user_id
                ):
                    return UserData(
                        name=user.nick_name,
                        card=user.personal_nick_name,
                        avatar_url=user.avatar_url,
                        user_id=user.dodo_source_id,
                        group_id=user.island_source_id,
                        join_time=int(user.join_time.timestamp()),
                    )
            else:
                # TODO: DoDo个人数据
                pass
        if isinstance(bot, KaiheilaBot):
            if group_id:
                if user := await bot.user_view(guild_id=group_id, user_id=user_id):
                    second = None
                    if user.joined_at:
                        second = int(user.joined_at / 1000)
                    return UserData(
                        name=user.nickname or "",
                        avatar_url=user.avatar,
                        user_id=user_id,
                        group_id=group_id,
                        join_time=second,
                    )
            else:
                # TODO: kaiheila用户详情
                pass
        if isinstance(bot, DiscordBot):
            # TODO: discord获取用户
            pass
        return None

    @classmethod
    async def get_user_avatar(cls, user_id: str, platform: str) -> bytes | None:
        """快捷获取用户头像

        参数:
            user_id: 用户id
            platform: 平台
        """
        if platform == "qq":
            url = f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=160"
            async with httpx.AsyncClient() as client:
                for _ in range(3):
                    try:
                        return (await client.get(url)).content
                    except Exception as e:
                        logger.error(
                            "获取用户头像错误",
                            "Util",
                            target=user_id,
                            platform=platform,
                        )
        else:
            pass
        return None

    @classmethod
    async def get_group_avatar(cls, gid: str, platform: str) -> bytes | None:
        """快捷获取用群头像

        参数:
            gid: 群组id
            platform: 平台
        """
        if platform == "qq":
            url = f"http://p.qlogo.cn/gh/{gid}/{gid}/640/"
            async with httpx.AsyncClient() as client:
                for _ in range(3):
                    try:
                        return (await client.get(url)).content
                    except Exception as e:
                        logger.error(
                            "获取群头像错误", "Util", target=gid, platform=platform
                        )
        else:
            pass
        return None

    @classmethod
    async def send_message(
        cls,
        bot: Bot,
        user_id: str | None,
        group_id: str | None,
        message: str | UniMessage,
    ) -> Receipt | None:
        """发送消息

        参数:
            bot: Bot
            user_id: 用户id
            group_id: 群组id或频道id
            message: 消息文本

        返回:
            Receipt | None: 是否发送成功
        """
        if target := cls.get_target(bot, user_id, group_id):
            send_message = (
                MessageUtils.build_message(message)
                if isinstance(message, str)
                else message
            )
            return await send_message.send(target=target, bot=bot)
        return None

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

        参数:
            bot: Bot

        返回:
            int: 更新个数
        """
        create_list = []
        update_list = []
        group_list, platform = await cls.get_group_list(bot)
        if group_list:
            db_group = await GroupConsole.all()
            db_group_id = [(group.group_id, group.channel_id) for group in db_group]
            for group in group_list:
                group.platform = platform
                if (group.group_id, group.channel_id) not in db_group_id:
                    create_list.append(group)
                    logger.debug(
                        "群聊信息更新成功",
                        "更新群信息",
                        target=f"{group.group_id}:{group.channel_id}",
                    )
                else:
                    _group = [
                        g
                        for g in db_group
                        if g.group_id == group.group_id
                        and g.channel_id == group.channel_id
                    ][0]
                    _group.group_name = group.group_name
                    _group.max_member_count = group.max_member_count
                    _group.member_count = group.member_count
                    update_list.append(_group)
        if create_list:
            await GroupConsole.bulk_create(create_list, 10)
        if group_list:
            await GroupConsole.bulk_update(
                update_list, ["group_name", "max_member_count", "member_count"], 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"
        # if isinstance(bot, DiscordBot):
        #     return "discord"
        return None

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

        参数:
            bot: Bot

        返回:
            tuple[list[GroupConsole], str]: 群组列表, 平台
        """
        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,
        user_id: str | None = None,
        group_id: str | None = None,
        channel_id: str | None = None,
    ):
        """获取发生Target

        参数:
            bot: Bot
            user_id: 用户id
            group_id: 频道id或群组id
            channel_id: 频道id

        返回:
            target: 对应平台Target
        """
        target = None
        if isinstance(bot, (v11Bot, v12Bot)):
            if group_id:
                target = Target(group_id)
            elif user_id:
                target = Target(user_id, private=True)
        elif isinstance(bot, (DodoBot, KaiheilaBot)):
            if group_id and channel_id:
                target = Target(channel_id, parent_id=group_id, channel=True)
            elif user_id:
                target = Target(user_id, private=True)
        return target


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

    参数:
        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 != PlatformUtils.get_platform(_bot):
                continue
            group_list, _ = await PlatformUtils.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_run = False
                        if check_func:
                            if is_coroutine_callable(check_func):
                                is_run = await check_func(group.group_id)
                            else:
                                is_run = check_func(group.group_id)
                        if not is_run:
                            continue
                        target = PlatformUtils.get_target(
                            _bot, None, group.group_id, group.channel_id
                        )
                        if target:
                            _used_group.append(key)
                            message_list = message
                            await MessageUtils.build_message(message_list).send(
                                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)