abot/plugins/music/main.py

import base64

from loguru import logger
import requests
from typing import Dict, Any, List, Optional, Tuple

from base.plugin_common.message_plugin_interface import MessagePluginInterface
from base.plugin_common.plugin_interface import PluginStatus
from utils.decorator.plugin_decorators import plugin_stats_decorator
from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager
from utils.decorator.points_decorator import plugin_points_cost
from wechat_ipad import WechatAPIClient

import aiohttp

from wechat_ipad.models.appmsg_xml import MUSIC_XML


class MusicPlugin(MessagePluginInterface):
    """音乐点播插件"""

    @property
    def name(self) -> str:
        return "音乐点播"

    @property
    def version(self) -> str:
        return "1.0.0"

    @property
    def description(self) -> str:
        return "提供音乐点播功能，支持QQ音乐和网易云音乐"

    @property
    def author(self) -> str:
        return "Trae AI"

    @property
    def command_prefix(self) -> Optional[str]:
        return ""  # 不需要前缀，直接匹配命令

    @property
    def commands(self) -> List[str]:
        return self._commands

    def __init__(self):
        super().__init__()

    def initialize(self, context: Dict[str, Any]) -> bool:
        """初始化插件"""
        self.LOG = logger
        self.LOG.info(f"正在初始化 {self.name} 插件...")

        # 保存上下文对象
        self.event_system = context.get("event_system")

        self._commands = self._config.get("Music", {}).get("command", ["点歌", "音乐"])
        self.command_format = self._config.get("Music", {}).get("command-format", "点歌 歌曲名")
        self.enable = self._config.get("Music", {}).get("enable", True)

        self.LOG.info(f"[{self.name}] 插件初始化完成，指令：{self._commands}")
        return True

    def start(self) -> bool:
        """启动插件"""
        self.LOG.info(f"[{self.name}] 插件已启动")
        self.status = PluginStatus.RUNNING
        return True

    def stop(self) -> bool:
        """停止插件"""
        self.LOG.info(f"[{self.name}] 插件已停止")
        self.status = PluginStatus.STOPPED
        return True

    def can_process(self, message: Dict[str, Any]) -> bool:
        """检查是否可以处理该消息"""
        if not self.enable:
            return False

        content = str(message.get("content", "")).strip()
        command = content.split(" ")[0]

        return command in self._commands

    @plugin_stats_decorator(plugin_name="音乐点播")
    @plugin_points_cost(2, "音乐点播消耗积分", Feature.MUSIC)
    async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
        """处理消息"""
        content = str(message.get("content", "")).strip()
        self.LOG.debug(f"插件执行： {self.name}：{content}")
        command = content.split(" ")[0]
        sender = message.get("sender")
        roomid = message.get("roomid", "")
        gbm: GroupBotManager = message.get("gbm")
        bot: WechatAPIClient = message.get("bot")

        # 检查命令格式
        if len(content.split(" ")) == 1:
            await bot.send_text_message((roomid if roomid else sender), f"❌命令格式错误！\n{self.command_format}"
                                        , sender)
            return False, "命令格式错误"

        # 检查权限
        if roomid and gbm.get_group_permission(roomid, Feature.MUSIC) == PermissionStatus.DISABLED:
            return False, "没有权限"

        # 提取歌曲名
        user_song_name = content[len(command):].strip()

        try:
            # 搜索歌曲
            song_info = self._search_song(user_song_name)
            if not song_info or not song_info.get("play_url"):
                await bot.send_text_message((roomid if roomid else sender), f"❌未找到歌曲：{user_song_name}", sender)
                return False, "未找到歌曲"

            # 发送音乐
            await self._send_music_message(bot, song_info, roomid or sender)
            return True, "发送成功"

        except Exception as e:
            self.LOG.error(f"处理音乐请求出错: {e}")
            return False, f"处理出错: {e}"

    def _search_song(self, song_name: str) -> Dict[str, Any]:
        """搜索歌曲信息"""
        try:
            # 尝试QQ音乐API
            fallback_api = f"https://www.hhlqilongzhu.cn/api/joox/juhe_music.php?msg={song_name}&type=json&n=1"
            response = requests.get(fallback_api)

            if response.status_code != 200:
                self.LOG.error(f"API 请求失败，状态码: {response.status_code}")
                return {}

            json_data = response.json().get("data")
            return {
                "song_name": json_data.get('title', ''),
                "singer_name": json_data.get('singer', ''),
                "play_url": json_data.get('url', ''),
                "singer_pic": json_data.get('cover', ''),
                "data_url": json_data.get('link', '')
            }

        except Exception as e:
            self.LOG.error(f"搜索歌曲出错: {e}")
            return {}

    async def url_to_base64(self, play_url: str):
        async with aiohttp.ClientSession() as session:
            async with session.get(play_url) as resp:
                resp.raise_for_status()
                data = await resp.read()
                b64_str = base64.b64encode(data).decode('utf-8')
                return b64_str

    async def _send_music_message(self, bot: WechatAPIClient, song_info: Dict[str, Any], receiver: str) -> bool:
        """发送音乐消息"""
        try:
            song_name = song_info.get("song_name", "")
            singer_name = song_info.get("singer_name", "")
            play_url = song_info.get("play_url", "")
            singer_pic = song_info.get("singer_pic", "")
            data_url = song_info.get("data_url", "")
            xml_message = f"{MUSIC_XML}".format(song_name=song_name, singer_name=singer_name, play_url=play_url,
                                                data_url=data_url, singer_pic=singer_pic)

            self.LOG.info(f"发送音乐消息：{xml_message}")
            res = await bot.send_app_message(wxid=receiver, xml=xml_message, type=0)
            self.LOG.info(f"发送音乐消息 res:{res}")
            return True

        except Exception as e:
            self.LOG.error(f"发送音乐消息出错: {e}")
            return False