Merge pull request #325 from NekoAria/2.0

优化逻辑和样式，适配插件元数据 (#325)

.github/dependabot.yml

@@ -7,7 +7,7 @@ version: 2
 updates:
   - package-ecosystem: "pip" # See documentation for possible values
     directory: "/" # Location of package manifests
-    target-branch: "2.7"
+    target-branch: "2.0"
     schedule:
       interval: "daily"
 

bot.py

@@ -5,8 +5,7 @@
 app = nonebot.get_asgi()
 driver = nonebot.get_driver()
 driver.register_adapter(OneBotV11Adapter)
-config = driver.config
 nonebot.load_plugins("src/plugins")
 
 if __name__ == "__main__":
-    nonebot.run(app="bot:app")
+    nonebot.run(app="__mp_main__:app")

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ELF_RSS"
-version = "2.6.3"
+version = "2.6.4"
 description = "QQ机器人 RSS订阅 插件，订阅源建议选择 RSSHub"
 authors = ["Quan666 <[email protected]>"]
 license = "GPL-3.0-only"

src/plugins/ELF_RSS2/__init__.py

@@ -1,16 +1,24 @@
 import asyncio
 
 import arrow
-import nonebot
 from nonebot import on_metaevent, require
-from nonebot.adapters.onebot.v11 import Event, LifecycleMetaEvent
+from nonebot.adapters.onebot.v11 import Bot, Event, LifecycleMetaEvent
 from nonebot.log import logger
+from nonebot.plugin import PluginMetadata
 
 from . import command
 from . import my_trigger as tr
 from .config import DATA_PATH, config
 from .rss_class import Rss
 
+VERSION = "2.6.4"
+__plugin_meta__ = PluginMetadata(
+    name="ELF_RSS",
+    description="QQ机器人 RSS订阅 插件，订阅源建议选择 RSSHub",
+    usage="https://github.com/Quan666/ELF_RSS",
+    extra={"author": "Quan666 <[email protected]>", "version": VERSION},
+)
+
 scheduler = require("nonebot_plugin_apscheduler").scheduler
 START_TIME = arrow.now()
 
@@ -26,17 +34,14 @@ async def check_first_connect(event: Event) -> bool:
 
 # 启动时发送启动成功信息
 @start_metaevent.handle()
-async def start() -> None:
-    bot = nonebot.get_bot()
+async def start(bot: Bot) -> None:
 
     # 启动后检查 data 目录，不存在就创建
     if not DATA_PATH.is_dir():
         DATA_PATH.mkdir()
 
     boot_message = (
-        f"Version: {config.version}\n"
-        "Author：Quan666\n"
-        "https://github.com/Quan666/ELF_RSS"
+        f"Version: v{VERSION}\nAuthor：Quan666\nhttps://github.com/Quan666/ELF_RSS"
     )
 
     rss_list = Rss.read_rss()  # 读取list

src/plugins/ELF_RSS2/command/add_dy.py

@@ -1,8 +1,7 @@
 import re
-from typing import Optional
 
 from nonebot import on_command
-from nonebot.adapters.onebot.v11 import Event, GroupMessageEvent, Message
+from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message, MessageEvent
 from nonebot.adapters.onebot.v11.permission import GROUP_ADMIN, GROUP_OWNER
 from nonebot.matcher import Matcher
 from nonebot.params import ArgPlainText, CommandArg
@@ -41,7 +40,7 @@ async def handle_first_receive(matcher: Matcher, args: Message = CommandArg()) -
 
 @RSS_ADD.got("RSS_ADD", prompt=prompt)
 async def handle_rss_add(
-    event: Event, name_and_url: str = ArgPlainText("RSS_ADD")
+    event: MessageEvent, name_and_url: str = ArgPlainText("RSS_ADD")
 ) -> None:
 
     try:
@@ -55,36 +54,28 @@ async def handle_rss_add(
         await RSS_ADD.reject(prompt)
         return
 
-    user_id = event.get_user_id()
-    group_id = None
-    guild_channel_id = None
+    await add_feed(name, url, event)
 
-    if isinstance(event, GroupMessageEvent):
-        group_id = event.group_id
-    elif isinstance(event, GuildMessageEvent):
-        guild_channel_id = f"{str(event.guild_id)}@{str(event.channel_id)}"
 
+async def add_feed(
+    name: str,
+    url: str,
+    event: MessageEvent,
+) -> None:
     rss = Rss()
     rss.name = name
     rss.url = url
-    await add_feed(rss, user_id, group_id, guild_channel_id)
-
-
-async def add_feed(
-    rss: Rss,
-    user_id: Optional[str],
-    group_id: Optional[int],
-    guild_channel_id: Optional[str],
-) -> None:
-    if guild_channel_id:
-        rss.add_user_or_group(guild_channel=guild_channel_id)
+    if isinstance(event, GuildMessageEvent):
+        rss.add_user_or_group(
+            guild_channel=f"{str(event.guild_id)}@{str(event.channel_id)}"
+        )
         await tr.add_job(rss)
         await RSS_ADD.finish("👏 订阅到当前子频道成功！")
-    elif group_id:
-        rss.add_user_or_group(group=str(group_id))
+    elif isinstance(event, GroupMessageEvent):
+        rss.add_user_or_group(group=str(event.group_id))
         await tr.add_job(rss)
         await RSS_ADD.finish("👏 订阅到当前群组成功！")
     else:
-        rss.add_user_or_group(user=user_id)
+        rss.add_user_or_group(user=event.get_user_id())
         await tr.add_job(rss)
         await RSS_ADD.finish("👏 订阅到当前账号成功！")

src/plugins/ELF_RSS2/command/change_dy.py

@@ -2,7 +2,7 @@
 from typing import Any, List, Optional
 
 from nonebot import on_command
-from nonebot.adapters.onebot.v11 import Event, GroupMessageEvent, Message
+from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message, MessageEvent
 from nonebot.adapters.onebot.v11.permission import GROUP_ADMIN, GROUP_OWNER
 from nonebot.log import logger
 from nonebot.matcher import Matcher
@@ -156,7 +156,7 @@ def handle_change_list(
 
 @RSS_CHANGE.got("RSS_CHANGE", prompt=prompt)
 async def handle_rss_change(
-    event: Event, change_info: str = ArgPlainText("RSS_CHANGE")
+    event: MessageEvent, change_info: str = ArgPlainText("RSS_CHANGE")
 ) -> None:
     group_id = event.group_id if isinstance(event, GroupMessageEvent) else None
     guild_channel_id = (

src/plugins/ELF_RSS2/command/del_dy.py

@@ -1,5 +1,5 @@
 from nonebot import on_command
-from nonebot.adapters.onebot.v11 import Event, GroupMessageEvent, Message
+from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message, MessageEvent
 from nonebot.adapters.onebot.v11.permission import GROUP_ADMIN, GROUP_OWNER
 from nonebot.matcher import Matcher
 from nonebot.params import ArgPlainText, CommandArg
@@ -28,7 +28,7 @@ async def handle_first_receive(matcher: Matcher, args: Message = CommandArg()) -
 
 @RSS_DELETE.got("RSS_DELETE", prompt="输入要删除的订阅名")
 async def handle_rss_delete(
-    event: Event, rss_name: str = ArgPlainText("RSS_DELETE")
+    event: MessageEvent, rss_name: str = ArgPlainText("RSS_DELETE")
 ) -> None:
     group_id = None
     guild_channel_id = None

src/plugins/ELF_RSS2/command/rsshub_add.py

@@ -2,14 +2,13 @@
 
 import aiohttp
 from nonebot import on_command
-from nonebot.adapters.onebot.v11 import Event, GroupMessageEvent, Message
+from nonebot.adapters.onebot.v11 import Message, MessageEvent
 from nonebot.adapters.onebot.v11.permission import GROUP_ADMIN, GROUP_OWNER
 from nonebot.matcher import Matcher
 from nonebot.params import ArgPlainText, CommandArg
 from nonebot.permission import SUPERUSER
 from nonebot.rule import to_me
 from nonebot.typing import T_State
-from nonebot_plugin_guild_patch import GuildMessageEvent
 from yarl import URL
 
 from ..config import config
@@ -31,7 +30,7 @@
 @RSSHUB_ADD.handle()
 async def handle_first_receive(matcher: Matcher, args: Message = CommandArg()) -> None:
     if args.extract_plain_text():
-        matcher.set_arg("router", args)
+        matcher.set_arg("route", args)
 
 
 @RSSHUB_ADD.got("name", prompt="请输入要订阅的订阅名")
@@ -40,9 +39,9 @@ async def handle_feed_name(name: str = ArgPlainText("name")) -> None:
         await RSSHUB_ADD.reject(f"已存在名为 {name} 的订阅，请重新输入")
 
 
-@RSSHUB_ADD.got("router", prompt="请输入要订阅的 RSSHub 路由名")
+@RSSHUB_ADD.got("route", prompt="请输入要订阅的 RSSHub 路由名")
 async def handle_rsshub_routes(
-    state: T_State, route: str = ArgPlainText("router")
+    state: T_State, route: str = ArgPlainText("route")
 ) -> None:
     rsshub_url = URL(config.rsshub)
     # 对本机部署的 RSSHub 不使用代理
@@ -66,14 +65,12 @@ async def handle_rsshub_routes(
     if route not in rsshub_routes["data"]:
         await RSSHUB_ADD.reject("没有这个路由，请重新输入")
     else:
-        route_list = rsshub_routes["data"][route]["routes"]
-        state["route_list"] = route_list
+        route_list = state["route_list"] = rsshub_routes["data"][route]["routes"]
         if len(route_list) > 1:
             await RSSHUB_ADD.send(
                 "请输入序号来选择要订阅的 RSSHub 路由：\n"
                 + "\n".join(
-                    f"{index + 1}. {__route}"
-                    for index, __route in enumerate(route_list)
+                    f"{index + 1}. {_route}" for index, _route in enumerate(route_list)
                 )
             )
         else:
@@ -99,7 +96,7 @@ async def handle_route_index(
 
 @RSSHUB_ADD.got("route_args")
 async def handle_route_args(
-    event: Event,
+    event: MessageEvent,
     state: T_State,
     name: str = ArgPlainText("name"),
     route_index: str = ArgPlainText("route_index"),
@@ -111,16 +108,4 @@ async def handle_route_args(
         if len(i.strip("#")) > 0:
             feed_url += f"/{i}"
 
-    user_id = event.get_user_id()
-    group_id = None
-    guild_channel_id = None
-
-    if isinstance(event, GroupMessageEvent):
-        group_id = event.group_id
-    elif isinstance(event, GuildMessageEvent):
-        guild_channel_id = f"{str(event.guild_id)}@{str(event.channel_id)}"
-
-    rss = Rss()
-    rss.name = name
-    rss.url = feed_url
-    await add_feed(rss, user_id, group_id, guild_channel_id)
+    await add_feed(name, feed_url, event)

src/plugins/ELF_RSS2/command/show_all.py

@@ -1,7 +1,7 @@
 import re
 
 from nonebot import on_command
-from nonebot.adapters.onebot.v11 import Event, GroupMessageEvent, Message
+from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message, MessageEvent
 from nonebot.adapters.onebot.v11.permission import GROUP_ADMIN, GROUP_OWNER
 from nonebot.params import CommandArg
 from nonebot.permission import SUPERUSER
@@ -22,7 +22,9 @@
 
 
 @RSS_SHOW_ALL.handle()
-async def handle_rss_show_all(event: Event, args: Message = CommandArg()) -> None:
+async def handle_rss_show_all(
+    event: MessageEvent, args: Message = CommandArg()
+) -> None:
     search_keyword = args.extract_plain_text().strip()
 
     group_id = None

src/plugins/ELF_RSS2/command/show_dy.py

@@ -1,7 +1,7 @@
 from typing import List
 
 from nonebot import on_command
-from nonebot.adapters.onebot.v11 import Event, GroupMessageEvent, Message
+from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message, MessageEvent
 from nonebot.adapters.onebot.v11.permission import GROUP_ADMIN, GROUP_OWNER
 from nonebot.params import CommandArg
 from nonebot.permission import SUPERUSER
@@ -36,7 +36,7 @@ def handle_rss_list(rss_list: List[Rss]) -> str:
 
 # 不带订阅名称默认展示当前群组或账号的订阅，带订阅名称就显示该订阅的
 @RSS_SHOW.handle()
-async def handle_rss_show(event: Event, args: Message = CommandArg()) -> None:
+async def handle_rss_show(event: MessageEvent, args: Message = CommandArg()) -> None:
     rss_name = args.extract_plain_text().strip()
 
     user_id = event.get_user_id()

src/plugins/ELF_RSS2/command/upload_group_file.py

@@ -2,9 +2,9 @@
 
 from nonebot import on_command
 from nonebot.adapters.onebot.v11 import (
-    Event,
     GroupMessageEvent,
     Message,
+    MessageEvent,
     PrivateMessageEvent,
 )
 from nonebot.params import CommandArg
@@ -22,7 +22,9 @@
 
 
 @upload_group_file.handle()
-async def handle_first_receive(event: Event, message: Message = CommandArg()) -> None:
+async def handle_first_receive(
+    event: MessageEvent, message: Message = CommandArg()
+) -> None:
     if isinstance(event, PrivateMessageEvent):
         await upload_group_file.finish("请在群聊中使用该命令")
     elif isinstance(event, GroupMessageEvent):

src/plugins/ELF_RSS2/config.py

@@ -1,18 +1,18 @@
 from pathlib import Path
-from typing import Any, List, Optional
+from typing import List, Optional
 
 from nonebot import get_driver
 from nonebot.config import BaseConfig
 from nonebot.log import logger
-from pydantic import AnyHttpUrl, Extra
+from pydantic import AnyHttpUrl
 
 DATA_PATH = Path.cwd() / "data"
 JSON_PATH = DATA_PATH / "rss.json"
 
 
 class ELFConfig(BaseConfig):
     class Config:
-        extra = Extra.allow
+        extra = "allow"
 
     rss_proxy: Optional[str] = None
     rsshub: AnyHttpUrl = "https://rsshub.app"  # type: ignore
@@ -37,14 +37,6 @@ class Config:
     down_status_msg_group: Optional[List[int]] = None
     down_status_msg_date: int = 10
 
-    version: str = ""
-
-    def __getattr__(self, name: str) -> Any:
-        data = self.dict()
-        return next(
-            (v for k, v in data.items() if k.casefold() == name.casefold()), None
-        )
-
 
 config = ELFConfig(**get_driver().config.dict())
 logger.debug(f"RSS Config loaded: {config!r}")

src/plugins/ELF_RSS2/parsing/check_update.py

@@ -1,9 +1,9 @@
 import hashlib
+from contextlib import suppress
 from email.utils import parsedate_to_datetime
 from typing import Any, Dict, List
 
 import arrow
-from arrow import Arrow
 from tinydb import Query, TinyDB
 
 
@@ -36,11 +36,9 @@ def check_update(db: TinyDB, new: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     return to_send_list
 
 
-def get_item_date(item: Dict[str, Any]) -> Arrow:
+def get_item_date(item: Dict[str, Any]) -> arrow.Arrow:
     if date := item.get("published", item.get("updated")):
-        try:
+        with suppress(Exception):
             date = parsedate_to_datetime(date)
-        except TypeError:
-            pass
         return arrow.get(date)
     return arrow.now()