liuwei_x_x/abot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增点歌插件。

Browse Source
This commit is contained in:
liuwei
2025-03-18 15:45:07 +08:00
parent adfa3505a5
commit ede83d6c94
8 changed files with 417 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
plugins/message_summary/main.py
View File

@@ -41,13 +41,10 @@ class MessageSummaryPlugin(MessagePluginInterface):
def initialize(self, context: Dict[str, Any]) -> bool:
"""初始化插件"""
try:
# 从插件配置中获取API密钥和URL
api_config = self._config.get("api", {})
self._api_key = api_config.get("api_key", "app-McGLzBhBjeBCSEi7n83MtuTo")
self._api_url = api_config.get("api_url", "http://192.168.2.240/v1/chat-messages")
self.all_contacts = context["all_contacts"]
self.message_storage = MessageStorage()
self.LOG.info(f"初始化 {self.name} 插件成功")
@@ -96,11 +93,13 @@ class MessageSummaryPlugin(MessagePluginInterface):
if gbm and gbm.get_group_permission(group_id, Feature.AI_CAPABILITY) == PermissionStatus.DISABLED:
return True, None
# 从消息历史中获取群聊记录
chat_content = self.message_storage.get_messages(group_id, self.all_contacts)
all_contacts: dict = message.get("all_contacts")
chat_content = self.message_storage.get_messages(group_id, all_contacts)
if len(chat_content) < 100:
return False, None
# 生成总结
summary, image_path = self._generate_summary(chat_content, self.all_contacts.get(group_id, group_id))
summary, image_path = self._generate_summary(chat_content, all_contacts.get(group_id, group_id))
# 发送总结结果
wcf = message.get("wcf")

1
plugins/music/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .main import MusicPlugin

8
plugins/music/config.toml Normal file
View File

@@ -0,0 +1,8 @@
[Music]
enable = true
command = ["点歌", "音乐", "音乐点播", "点播音乐", "音乐点歌"]
command-format = """
-----Bot-----
🎵点歌指令：
点歌 歌曲名
"""

290
plugins/music/main.py Normal file
View File

@@ -0,0 +1,290 @@
import logging
import requests
import lz4.block as lb
from typing import Dict, Any, List, Optional, Tuple
from wcferry import Wcf
from plugin_common.message_plugin_interface import MessagePluginInterface
from plugin_common.plugin_interface import PluginStatus
from robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager
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 initialize(self, context: Dict[str, Any]) -> bool:
"""初始化插件"""
self.LOG = logging.getLogger(f"Plugin.{self.name}")
self.LOG.info(f"正在初始化 {self.name} 插件...")
# 保存上下文对象
self.wcf = context.get("wcf")
self.event_system = context.get("event_system")
self.message_util = context.get("message_util")
# 加载配置
self.load_config()
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
def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""处理消息"""
content = str(message.get("content", "")).strip()
command = content.split(" ")[0]
sender = message.get("sender")
roomid = message.get("roomid", "")
wcf: Wcf = message.get("wcf")
gbm: GroupBotManager = message.get("gbm")
# 检查命令格式
if len(content.split(" ")) == 1:
wcf.send_text(f"-----Bot-----\n❌命令格式错误！\n{self.command_format}",
(roomid if roomid else sender), sender)
return True, "命令格式错误"
# 检查权限
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"):
wcf.send_text(f"-----Bot-----\n❌未找到歌曲：{user_song_name}",
(roomid if roomid else sender), sender)
return True, "未找到歌曲"
# 发送音乐
self._send_music_message(wcf, song_info, roomid or sender)
return True, "发送成功"
except Exception as e:
self.LOG.error(f"处理音乐请求出错: {e}")
wcf.send_text(f"-----Bot-----\n❌请求出错：{e}",
(roomid if roomid else sender), sender)
return True, f"处理出错: {e}"
def _search_song(self, song_name: str) -> Dict[str, Any]:
"""搜索歌曲信息"""
try:
# 尝试QQ音乐API
short_play_api = f"https://qqmusic.qqovo.cn/getSearchByKey?key={song_name}&page=1&limit=1"
fallback_api = f"https://www.hhlqilongzhu.cn/api/dg_wyymusic.php?gm={song_name}&n=1&num=1&type=json"
response = requests.get(short_play_api)
if response.status_code == 400:
response = requests.get(fallback_api)
if response.status_code != 200:
self.LOG.error(f"API 请求失败，状态码: {response.status_code}")
return {}
json_data = response.json()
result = json_data.get('response', {}).get('data', {}).get('song', {}).get('list', [])
if not result:
# 尝试备用API
response = requests.get(fallback_api)
if response.status_code == 200:
music_response = response.json()
return {
"song_name": music_response.get('title', ''),
"singer_name": music_response.get('singer', ''),
"play_url": music_response.get('music_url', ''),
"singer_pic": music_response.get('cover', ''),
"data_url": music_response.get('link', '')
}
return {}
# 解析QQ音乐结果
first_song = result[0]
song_name = first_song.get('songname', '')
song_mid = first_song.get('songmid', '')
first_singer_name = first_song.get('singer', [{}])[0].get('name', '')
zhida_singer = json_data.get('response', {}).get('data', {}).get('zhida', {}).get('zhida_singer', {})
singer_pic = zhida_singer.get('singerPic', '') if zhida_singer else None
# 获取播放链接
music_play_api = f"https://qqmusic.qqovo.cn/getMusicPlay?songmid={song_mid}&quality=m4a"
music_response = requests.get(music_play_api)
if music_response.status_code == 400:
# 尝试备用API
response = requests.get(fallback_api)
if response.status_code == 200:
music_response = response.json()
return {
"song_name": music_response.get('title', song_name),
"singer_name": music_response.get('singer', first_singer_name),
"play_url": music_response.get('music_url', ''),
"singer_pic": music_response.get('cover', singer_pic),
"data_url": music_response.get('link', '')
}
return {}
if music_response.status_code != 200:
self.LOG.error(f"获取播放链接失败，状态码: {music_response.status_code}")
return {}
music_data = music_response.json()
play_url = music_data.get('data', {}).get('playUrl', {}).get(song_mid, {}).get('url', '')
if not play_url:
# 尝试备用API
response = requests.get(fallback_api)
if response.status_code == 200:
music_response = response.json()
return {
"song_name": music_response.get('title', song_name),
"singer_name": music_response.get('singer', first_singer_name),
"play_url": music_response.get('music_url', ''),
"singer_pic": music_response.get('cover', singer_pic),
"data_url": music_response.get('link', '')
}
return {}
data_url = f"https://y.qq.com/n/ryqq/songDetail/{song_mid}"
return {
"song_name": song_name,
"singer_name": first_singer_name,
"play_url": play_url,
"singer_pic": singer_pic,
"data_url": data_url
}
except Exception as e:
self.LOG.error(f"搜索歌曲出错: {e}")
return {}
def _send_music_message(self, wcf, 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"""
<?xml version="1.0"?>
<msg>
<appmsg appid="wx8dd6ecd81906fd84" sdkver="0">
<title>{song_name}</title>
<des>{singer_name}\n❤Bot-祝您天天开心❤</des>
<action>view</action>
<type>3</type>
<showtype>0</showtype>
<content />
<url>{data_url}</url>
<dataurl>{play_url}</dataurl>
<lowurl/>
<lowdataurl/>
<recorditem />
<thumburl />
<messageaction />
<laninfo />
<extinfo />
<sourceusername />
<sourcedisplayname />
<commenturl />
<appattach>
<totallen>0</totallen>
<attachid />
<emoticonmd5></emoticonmd5>
<fileext />
<aeskey></aeskey>
</appattach>
<webviewshared>
<publisherId />
<publisherReqId>0</publisherReqId>
</webviewshared>
<weappinfo>
<pagepath />
<username />
<appid />
<appservicetype>0</appservicetype>
</weappinfo>
<websearch />
<songalbumurl>{singer_pic}</songalbumurl>
</appmsg>
<scene>0</scene>
<appinfo>
<version>49</version>
<appname></appname>
</appinfo>
<commenturl />
</msg>"""
# 修改消息数据库里面的消息content内容
text_bytes = xml_message.encode('utf-8')
compressed_data = lb.compress(text_bytes, store_size=False).hex()
data = wcf.query_sql('MSG0.db', "SELECT * FROM MSG where type = 49 limit 1")
wcf.query_sql('MSG0.db',
f"""UPDATE MSG SET CompressContent = x'{compressed_data}', BytesExtra=x'', type=49, SubType=3,
IsSender=0, TalkerId=2 WHERE MsgSvrID={data[0]['MsgSvrID']}"""
)
result = wcf.forward_msg(data[0]["MsgSvrID"], receiver)
self.LOG.info(f"点歌发送结果: {result}")
return True
except Exception as e:
self.LOG.error(f"发送音乐消息出错: {e}")
return False