diff --git a/plugins/music_163/config_default.py b/plugins/music_163/config_default.py deleted file mode 100644 index d2389c2e86ab705def5b9b580f5dcb2b6fac47dc..0000000000000000000000000000000000000000 --- a/plugins/music_163/config_default.py +++ /dev/null @@ -1,8 +0,0 @@ -API_URL = "http://127.0.0.1:3000" - -SEARCH_LIMIT = 10 - -LOGIN_MODE = "phone" # "phone" or "email" 其他则不登录 -PHONE = "" -EMAIL = "" -PASSWORD = "" \ No newline at end of file diff --git a/plugins/music_163/music_163.py b/plugins/music_163/music_163.py deleted file mode 100644 index 24d5fb3eb7091f459173bda896c082f6849e651a..0000000000000000000000000000000000000000 --- a/plugins/music_163/music_163.py +++ /dev/null @@ -1,124 +0,0 @@ -from register import command -from global_vars import CONFIG -import threading -import urllib.parse,urllib.request,urllib.error -import json -import base64 -from util import print_log - -config = CONFIG[__name__] - -def plugin(): - return { - "author":"Antares", - "version":1.0, - "description":"网易云点歌" - } - -def login(): - if config.LOGIN_MODE == "phone": - url = config.API_URL + f"/login/cellphone?phone={config.PHONE}&password={config.PASSWORD}" - elif config.LOGIN_MODE == "email": - url = config.API_URL + f"/login?email={config.EMAIL}&password={config.PASSWORD}" - else: - return - - with urllib.request.urlopen(url) as f: - data = json.JSONDecoder().decode(f.read().decode("utf8")) - - if data['code'] == 200: - print_log("登陆成功!") - else: - print_log("登陆失败！请检查账号密码！") - - -def search_music(key:str) -> dict: - url = config.API_URL + f"/search?keywords={urllib.parse.quote(key)}&limit={config.SEARCH_LIMIT}" - with urllib.request.urlopen(url) as f: - data = json.JSONDecoder().decode(f.read().decode("utf8")) - if 'result' in data: - return data['result'] - else: - return {'songCount':0} - -def check_music(music_id:int) -> bool: - url = config.API_URL + f"/check/music?id={music_id}" - try: - with urllib.request.urlopen(url) as f: - data = json.JSONDecoder().decode(f.read().decode("utf8")) - except urllib.error.HTTPError as err: - print_log(err) - return False - return data['success'] - -def get_music_url(music_id:int) -> str: - url = config.API_URL + f"/song/url?id={music_id}&br=320000" - with urllib.request.urlopen(url) as f: - data = json.JSONDecoder().decode(f.read().decode("utf8")) - return data['data'][0]['url'] - -@command(name="music",help="网易云音乐点歌") -def music(bot,context,args): - while args[-1] == "": - del args[-1] - - def handle(): - raw = False - link = False - if args [-1] == "raw": - raw = True - del args[-1] - elif args[-1] == "link": - link = True - del args[-1] - - music_id = -1 - if args[1] == "id": - try: - music_id = int(args[2]) - except ValueError as err: - print_log(err) - bot.send(context,"请输入正确的id") - return - login() - - if music_id != -1: - if not check_music(music_id): - bot.send(context,"id对应的音乐不存在或无版权") - return - else: - if raw : - bot.send(context,f"[CQ:music,type=163,id={music_id}]") - else: - url = get_music_url(music_id) - if link: - bot.send(context, url) - else: - bot.send(context,f"[CQ:record,file={url}]") - return - - key = " ".join(args[1:]) - data = search_music(key) - - if data['songCount'] == 0: - bot.send(context,"您搜索的歌曲不存在") - return - - for item in data['songs']: - music_id = item['id'] - if check_music(music_id): - if raw: - bot.send(context,f"[CQ:music,type=163,id={music_id}]") - else: - url = get_music_url(music_id) - if link: - bot.send(context,url) - else: - bot.send(context,f"[CQ:record,file={url}]") - return - - bot.send(context,"您搜索的歌曲不存在或无版权") - - threading.Thread(target=handle).start() - - \ No newline at end of file diff --git a/plugins_new/music_163/plugin.py b/plugins_new/music_163/plugin.py new file mode 100644 index 0000000000000000000000000000000000000000..d6e4ea7b4c0f3b20be3a5dd294e1e3a1bcde5026 --- /dev/null +++ b/plugins_new/music_163/plugin.py @@ -0,0 +1,159 @@ +from common.plugin import Plugin +from common.config_loader import ConfigBase +from common.datatypes import PluginMeta +from common.countdown_bot import CountdownBot +from common.loop import TimeTuple +from common.command import ChatType +from common.event import GroupMessageEvent +import aiohttp +import urllib +from typing import Dict, List + + +class Music163Config(ConfigBase): + # https://github.com/Binaryify/NeteaseCloudMusicApi + API_URL = "http://127.0.0.1:3000" + SEARCH_LIMIT = 10 + LOGIN_MODE = "phone" # "phone" or "email" 其他则不登录 + PHONE = "" + EMAIL = "" + PASSWORD = "" + + +class Music163Plugin(Plugin): + + async def login(self) -> bool: + if self.config.LOGIN_MODE == "phone": + async with self.aioclient.get(f"{self.config.API_URL}/login/cellphone", params={ + "phone": self.config.PHONE, + "password": self.config.PASSWORD + }) as resp: + result = await resp.json() + return result["code"] == 200 + elif self.config.LOGIN_MODE == "email": + async with self.aioclient.get(f"{self.config.API_URL}/login", params={ + "email": self.config.EMAIL, + "password": self.config.PASSWORD + }) as resp: + result = await resp.json() + return result["code"] == 200 + else: + return False + + async def check_login_status(self) -> bool: + if self.config.LOGIN_MODE != "phone" and self.config.LOGIN_MODE != "email": + return True + async with self.aioclient.get(f"{self.config.API_URL}/login/refresh") as resp: + result = await resp.json() + return result["code"] == 200 + + async def check_music_available(self, music_id: int) -> bool: + async with self.aioclient.get(f"{self.config.API_URL}/check/music", params={"id": music_id}) as resp: + result = await resp.json() + return result["success"] + + async def get_music_url(self, music_id: int) -> str: + async with self.aioclient.get(f"{self.config.API_URL}/song/url", params={ + "id": music_id, + "br": 320000 + }) as resp: + result = await resp.json() + return result['data'][0]['url'] + + async def search_music(self, key_words: str) -> List[dict]: + async with self.aioclient.get(f"{self.config.API_URL}/search", params={ + "keywords": key_words, + "limit": self.config.SEARCH_LIMIT + }) as resp: + result = await resp.json() + if result["code"] != 200 or result["result"]["songCount"] == 0: + return [] + else: + return result["result"]["songs"] + + async def command_music(self, plugin, args: List[str], raw_string: str, context, evt: GroupMessageEvent): + while not args[-1]: + del args[-1] + raw = False + link = False + if args[-1] == "raw": + raw = True + del args[-1] + elif args[-1] == "link": + link = True + del args[-1] + + if not await self.check_login_status(): + if not await self.login(): + await self.bot.client_async.send("网易云账号登陆失败！") + + if args[0] == "id": + try: + query_id = int(args[1]) + except ValueError: + await self.bot.client_async.send(context, "请输入正确的id") + return + + if not await self.check_music_available(query_id): + await self.bot.client_async.send(context, "id对应的音乐不存在或无版权") + return + + if raw: + await self.bot.client_async.send(context, f"[CQ:music,type=163,id={query_id}]") + else: + music_url = await self.get_music_url(query_id) + if not music_url: + await self.bot.client_async.send(context, "无法取得音乐链接，请检查是否为VIP歌曲") + return + if link: + await self.bot.client_async.send(context, music_url) + else: + await self.bot.client_async.send(context, f"[CQ:record,file={music_url}]") + return + + key_words = " ".join(args) + musics_list = await self.search_music(key_words) + + for music in musics_list: + music_id = music['id'] + if await self.check_music_available(music_id): + if raw: + await self.bot.client_async.send( + context, f"[CQ:music,type=163,id={music_id}]") + else: + music_url = await self.get_music_url(music_id) + if not music_url: + continue + if link: + await self.bot.client_async.send(context, music_url) + else: + await self.bot.client_async.send(context, f"[CQ:record,file={music_url}]") + return + + await self.bot.client_async.send(context, "您搜索的歌曲可能不存在、无版权或为VIP歌曲") + + def on_enable(self): + self.aioclient = aiohttp.ClientSession() + self.bot: CountdownBot + self.config: Music163Config + self.register_command_wrapped( + command_name="music", + command_handler=self.command_music, + help_string="网易云音乐查询 | music [歌名] [raw/link](可选) | music id [歌曲id] [raw/link](可选)", + chats=ChatType.all(), + is_async=True + ) + + +def get_plugin_class(): + return Music163Plugin + + +def get_config_class(): + return Music163Config + + +def get_plugin_meta(): + return PluginMeta( + "Antares", 2.0, "网易云音乐推歌" + )