Merge branch 'main' of https://github.com/wling-art/U1bot

src/plugins/ncm/__init__.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 from typing import Tuple, Any, Union
 
 from nonebot import on_regex, on_command, on_message
@@ -40,9 +38,10 @@ async def song_is_open(event: Union[GroupMessageEvent, PrivateMessageEvent]) ->
     if isinstance(event, GroupMessageEvent):
         if info := setting.search(Q["group_id"] == event.group_id):
             return info[0]["song"]
-        setting.insert({"group_id": event.group_id, "song": False, "list": False})
+        setting.insert({"group_id": event.group_id,
+                       "song": False, "list": False})
         return False
-    elif isinstance(event, PrivateMessageEvent):
+    if isinstance(event, PrivateMessageEvent):
         if info := setting.search(Q["user_id"] == event.user_id):
             return info[0]["song"]
         setting.insert({"user_id": event.user_id, "song": True, "list": True})
@@ -56,25 +55,24 @@ async def playlist_is_open(
         info = setting.search(Q["group_id"] == event.group_id)
         if info:
             return info[0]["list"]
-        else:
-            setting.insert({"group_id": event.group_id, "song": False, "list": False})
-            return False
-    elif isinstance(event, PrivateMessageEvent):
+        setting.insert({"group_id": event.group_id,
+                       "song": False, "list": False})
+        return False
+    if isinstance(event, PrivateMessageEvent):
         info = setting.search(Q["user_id"] == event.user_id)
         if info:
             return info[0]["list"]
-        else:
-            setting.insert({"user_id": event.user_id, "song": True, "list": True})
-            return True
+        setting.insert({"user_id": event.user_id,
+                       "song": True, "list": True})
+        return True
 
 
 async def check_search() -> bool:
     info = setting.search(Q["global"] == "search")
     if info:
         return info[0]["value"]
-    else:
-        setting.insert({"global": "search", "value": True})
-        return True
+    setting.insert({"global": "search", "value": True})
+    return True
 
 
 async def music_set_rule(event: Union[GroupMessageEvent, PrivateMessageEvent]) -> bool:
@@ -86,7 +84,6 @@ async def music_set_rule(event: Union[GroupMessageEvent, PrivateMessageEvent]) -
 
 
 async def music_reply_rule(event: Union[GroupMessageEvent, PrivateMessageEvent]):
-    # logger.info(event.get_plaintext())
     return event.reply and event.get_plaintext().strip() == "下载"
 
 
@@ -95,7 +92,8 @@ async def music_reply_rule(event: Union[GroupMessageEvent, PrivateMessageEvent])
 """功能设置"""
 music_regex = on_regex(r"(song|url)\?id=([0-9]+)(|&)", priority=2, block=False)
 """歌曲id识别"""
-playlist_regex = on_regex(r"playlist\?id=([0-9]+)(|&)", priority=2, block=False)
+playlist_regex = on_regex(
+    r"playlist\?id=([0-9]+)(|&)", priority=2, block=False)
 """歌单识别"""
 music_reply = on_message(priority=2, rule=Rule(music_reply_rule), block=False)
 """回复下载"""
@@ -117,10 +115,10 @@ async def receive_song(
 ):
     _id = await nncm.search_song(keyword=song.extract_plain_text(), limit=1)
     message_id = await bot.send(
-        event=event, message=Message(MessageSegment.music(type_="163", id_=_id))
+        event=event, message=Message(
+            MessageSegment.music(type_="163", id_=_id))
     )
     nncm.get_song(message_id=message_id["message_id"], nid=_id)
-    # try:
 
     # except ActionFailed as e:
     #    logger.error(e.info)
@@ -185,28 +183,34 @@ async def set_receive(
                     if mold in TRUE:
                         info[0]["song"] = True
                         info[0]["list"] = True
-                        setting.update(info[0], Q["group_id"] == event.group_id)
+                        setting.update(
+                            info[0], Q["group_id"] == event.group_id)
                         msg = "已开启自动下载功能"
                         await bot.send(
-                            event=event, message=Message(MessageSegment.text(msg))
+                            event=event, message=Message(
+                                MessageSegment.text(msg))
                         )
                     elif mold in FALSE:
                         info[0]["song"] = False
                         info[0]["list"] = False
-                        setting.update(info[0], Q["group_id"] == event.group_id)
+                        setting.update(
+                            info[0], Q["group_id"] == event.group_id)
                         msg = "已关闭自动下载功能"
                         await bot.send(
-                            event=event, message=Message(MessageSegment.text(msg))
+                            event=event, message=Message(
+                                MessageSegment.text(msg))
                         )
                     logger.debug(f"用户<{event.sender.nickname}>执行操作成功")
                 else:
                     if mold in TRUE:
                         setting.insert(
-                            {"group_id": event.group_id, "song": True, "list": True}
+                            {"group_id": event.group_id,
+                                "song": True, "list": True}
                         )
                     elif mold in FALSE:
                         setting.insert(
-                            {"group_id": event.group_id, "song": False, "list": False}
+                            {"group_id": event.group_id,
+                                "song": False, "list": False}
                         )
             elif isinstance(event, PrivateMessageEvent):
                 info = setting.search(Q["user_id"] == event.user_id)
@@ -218,15 +222,17 @@ async def set_receive(
                         setting.update(info[0], Q["user_id"] == event.user_id)
                         msg = "已开启下载功能"
                         await bot.send(
-                            event=event, message=Message(MessageSegment.text(msg))
+                            event=event, message=Message(
+                                MessageSegment.text(msg))
                         )
                     elif mold in FALSE:
                         info[0]["song"] = False
                         info[0]["list"] = False
                         setting.update(info[0], Q["user_id"] == event.user_id)
                         msg = "已关闭下载功能"
                         await bot.send(
-                            event=event, message=Message(MessageSegment.text(msg))
+                            event=event, message=Message(
+                                MessageSegment.text(msg))
                         )
                     logger.debug(f"用户<{event.sender.nickname}>执行操作成功")
                 else:

src/plugins/ncm/data_source.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 import asyncio
 import re
 import time
@@ -38,7 +36,8 @@
 cmd = list(nonebot.get_driver().config.command_start)[0]
 
 
-class NcmLoginFailedException(Exception): pass
+class NcmLoginFailedException(Exception):
+    pass
 
 
 # ============主类=============
@@ -61,7 +60,8 @@ def load_user(session: str):
 
     def login(self) -> bool:
         try:
-            self.api.login.LoginViaCellphone(phone=ncm_config.ncm_phone, password=ncm_config.ncm_password)
+            self.api.login.LoginViaCellphone(
+                phone=ncm_config.ncm_phone, password=ncm_config.ncm_password)
             self.get_user_info()
             return True
         except Exception as e:
@@ -74,25 +74,28 @@ def login(self) -> bool:
             return False
 
     def get_user_info(self) -> str:
-        message: str = f"欢迎您网易云用户:{GetCurrentSession().nickname} [{GetCurrentSession().uid}]";
+        message: str = f"欢迎您网易云用户:{GetCurrentSession().nickname} [{GetCurrentSession().uid}]"
         logger.success(message)
         self.save_user(DumpSessionAsString(GetCurrentSession()))
         return message
 
     def get_phone_login(self):
         phone = ncm_config.ncm_phone
         ctcode = int(ncm_config.ncm_ctcode)
-        result = self.api.login.SetSendRegisterVerifcationCodeViaCellphone(cell=phone, ctcode=ctcode)
+        result = self.api.login.SetSendRegisterVerifcationCodeViaCellphone(
+            cell=phone, ctcode=ctcode)
         if not result.get('code', 0) == 200:
             logger.error(result)
         else:
             logger.success('已发送验证码,输入验证码:')
         while True:
             captcha = int(input())
-            verified = self.api.login.GetRegisterVerifcationStatusViaCellphone(phone, captcha, ctcode)
+            verified = self.api.login.GetRegisterVerifcationStatusViaCellphone(
+                phone, captcha, ctcode)
             if verified.get('code', 0) == 200:
                 break
-        result = self.api.login.LoginViaCellphone(phone, captcha=captcha, ctcode=ctcode)
+        result = self.api.login.LoginViaCellphone(
+            phone, captcha=captcha, ctcode=ctcode)
         self.get_user_info()
 
     def get_qrcode(self):
@@ -122,12 +125,14 @@ def get_qrcode(self):
 
     def detail_names(self, ids: List[int]) -> List[str]:
         songs: list = self.api.track.GetTrackDetail(song_ids=ids)["songs"]
-        detail = [(data["name"] + "-" + ",".join([names["name"] for names in data["ar"]])) for data in songs]
+        detail = [(data["name"] + "-" + ",".join([names["name"]
+                   for names in data["ar"]])) for data in songs]
         return detail
 
     @logger.catch()
     def get_detail(self, ids: List[int]):
-        data: list = self.api.track.GetTrackAudio(song_ids=ids, bitrate=ncm_config.ncm_bitrate * 1000)["data"]
+        data: list = self.api.track.GetTrackAudio(
+            song_ids=ids, bitrate=ncm_config.ncm_bitrate * 1000)["data"]
         names: list = self.detail_names(ids)
         for i in range(len(ids)):
             data[i]['ncm_name'] = names[i]
@@ -143,7 +148,8 @@ async def music_check(self, nid: Union[int, List[int]], event: Union[GroupMessag
                 info = music.search(Q["id"] == i)
                 if info:
                     try:
-                        tasks.append(asyncio.create_task(self.upload_data_file(event=event, data=info[0])))
+                        tasks.append(asyncio.create_task(
+                            self.upload_data_file(event=event, data=info[0])))
                         del_nid.append(i)
                     except Exception:
                         continue
@@ -154,7 +160,8 @@ async def music_check(self, nid: Union[int, List[int]], event: Union[GroupMessag
             info = music.search(Q["id"] == nid)
             if info:
                 try:
-                    tasks.append(asyncio.create_task(self.upload_data_file(event=event, data=info[0])))
+                    tasks.append(asyncio.create_task(
+                        self.upload_data_file(event=event, data=info[0])))
                     return
                 except Exception as e:
                     if isinstance(e, ActionFailed) and e.info.get("retcode") != 10003:
@@ -168,7 +175,8 @@ async def music_check(self, nid: Union[int, List[int]], event: Union[GroupMessag
             await self.start_upload(ids=nid, event=event)
 
     async def search_song(self, keyword: str, limit: int = 1) -> int:  # 搜索歌曲
-        res = self.api.cloudsearch.GetSearchResult(keyword=keyword, stype=SONG, limit=limit)
+        res = self.api.cloudsearch.GetSearchResult(
+            keyword=keyword, stype=SONG, limit=limit)
         logger.debug(f"搜索歌曲{keyword},返回结果:{res}")
         if "result" in res.keys():
             data = res["result"]["songs"]
@@ -178,10 +186,12 @@ async def search_song(self, keyword: str, limit: int = 1) -> int:  # 搜索歌
             return data[0]["id"]
 
     async def search_user(self, keyword: str, limit: int = 1):  # 搜索用户
-        self.api.cloudsearch.GetSearchResult(keyword=keyword, stype=USER, limit=limit)
+        self.api.cloudsearch.GetSearchResult(
+            keyword=keyword, stype=USER, limit=limit)
 
     async def search_playlist(self, keyword: str, limit: int = 1):  # 搜索歌单
-        self.api.cloudsearch.GetSearchResult(keyword=keyword, stype=PLAYLIST, limit=limit)
+        self.api.cloudsearch.GetSearchResult(
+            keyword=keyword, stype=PLAYLIST, limit=limit)
 
     @staticmethod
     def check_message(message_id: int):
@@ -261,13 +271,6 @@ async def upload_private_file(user_id: int, file: str, name: str):
                 await bot.send_private_msg(user_id=user_id, message=Message(MessageSegment.text(
                     "[ERROR]  文件上传失败\r\n[原因]  上传超时(一般来说还在传,建议等待五分钟)")))
 
-    # @run_sync
-    # def get_zip(self, lid: int, filenames: list):
-    #     zip_file_new = f'{lid}.zip'
-    #     with zipfile.ZipFile(str(Path.cwd().joinpath("music").joinpath(zip_file_new)), 'w', zipfile.ZIP_DEFLATED) as z:
-    #         for f in filenames:
-    #             z.write(str(f), f.name)
-    #     return zip_file_new
     async def upload(self, data: dict, fr: str, event: Union[GroupMessageEvent, PrivateMessageEvent]):
         if data["code"] == 404:
             logger.error("未从网易云读取到下载地址")

src/plugins/nonebot_plugin_addFriend/utils.py

@@ -13,7 +13,7 @@ def filterFriend(comment, type, allowTextList):
 
 
 async def parseMsg(commandText, resMsg, font_size=32, isText=1):
-    if type(resMsg) == list:
+    if type(resMsg) is list:
         temp = ''.join(str(item)+'\n' for item in resMsg)
         temp = temp.replace("'", "").replace('"', '')
         resMsg = temp

src/plugins/nonebot_plugin_heweather/weather_data.py

@@ -84,11 +84,10 @@ async def _get_city_id(self, api_type: str = "lookup"):
         logger.debug(res)
         if res["code"] == "404":
             raise CityNotFoundError()
-        elif res["code"] != "200":
+        if res["code"] != "200":
             raise APIError(f'错误! 错误代码: {res["code"]}{self.__reference}')
-        else:
-            self.city_name = res["location"][0]["name"]
-            return res["location"][0]["id"]
+        self.city_name = res["location"][0]["name"]
+        return res["location"][0]["id"]
 
     def _data_validate(self):
         if self.now.code != "200" or self.daily.code != "200":

src/plugins/today_in_history/__init__.py

@@ -50,7 +50,7 @@ async def _(event: MessageEvent):
 
 # api处理->json
 def text_handle(text: str) -> json:
-    text = text.replace("<\/a>", "")
+    text = text.replace(r"<\/a>", "")
     text = text.replace("\n", "")
 
     # 去除html标签

src/plugins/waifu/__init__.py

@@ -260,7 +260,6 @@ async def check_divorce_rule(event):
 
 # 分手
 if waifu_cd_bye > -1:
-    global cd_bye
     cd_bye = {}
     bye = on_command(
         "离婚",