v2.6.18 (#426)

* fix(config): 新增配置项 `debug` ，用来控制是否开启 debug 模式 (fix #413)

* fix(forward): 修正合并转发相关逻辑 (fixes #417)

* fix(sending): 修正消息发送相关逻辑 (fixes #414)

* refactor(sub): 为添加订阅相关逻辑加上防呆机制 (resolve #424)

* build(deps): 去除没用到的依赖 `typing-extensions`

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ELF_RSS"
-version = "2.6.17"
+version = "2.6.18"
 description = "QQ机器人 RSS订阅 插件，订阅源建议选择 RSSHub"
 authors = ["Quan666 <[email protected]>"]
 license = "GPL-3.0-only"
@@ -16,6 +16,7 @@ keywords = ["nonebot", "nonebot2", "rss" ,"elf" ,"rsshub"]
 python = "^3.8.3"
 aiohttp = {extras = ["speedups"], version = "^3.8.4"}
 arrow = "^1.2.3"
+async-timeout = "^4.0.2"
 bbcode = "^1.1.0"
 cachetools = "^5.3.0"
 emoji = "^2.2.0"
@@ -34,7 +35,6 @@ pyquery = "^2.0.0"
 python-qbittorrent = "^0.4.3"
 tenacity = "^8.2.2"
 tinydb = "^4.7.1"
-typing-extensions = "^4.5.0"
 yarl = "^1.9.1"
 
 [tool.poetry.dev-dependencies]

requirements.txt

@@ -1,5 +1,6 @@
 aiohttp[speedups]~=3.8.4
 arrow~=1.2.3
+async-timeout~=4.0.2
 bbcode~=1.1.0
 cachetools~=5.3.0
 emoji~=2.2.0
@@ -19,5 +20,4 @@ pyquery~=2.0.0
 python-qbittorrent~=0.4.3
 tenacity~=8.2.2
 tinydb~=4.7.1
-typing-extensions~=4.5.0
 yarl~=1.9.1

src/plugins/ELF_RSS2/__init__.py

@@ -14,7 +14,7 @@
 require("nonebot_plugin_apscheduler")
 from nonebot_plugin_apscheduler import scheduler  # noqa
 
-VERSION = "2.6.17"
+VERSION = "2.6.18"
 
 __plugin_meta__ = PluginMetadata(
     name="ELF_RSS",

src/plugins/ELF_RSS2/command/add_dy.py

@@ -57,10 +57,13 @@ async def handle_rss_add(
     event: MessageEvent, name_and_url: str = ArgPlainText("RSS_ADD")
 ) -> None:
     try:
-        name, url = name_and_url.split(" ")
+        name, url = name_and_url.strip().split(" ")
     except ValueError:
         await RSS_ADD.reject(prompt)
         return
+    if not name or not url:
+        await RSS_ADD.reject(prompt)
+        return
 
     if _ := Rss.get_one_by_name(name):
         await RSS_ADD.finish(f"已存在订阅名为 {name} 的订阅")

src/plugins/ELF_RSS2/command/rsshub_add.py

@@ -37,12 +37,16 @@
 
 @RSSHUB_ADD.handle()
 async def handle_first_receive(matcher: Matcher, args: Message = CommandArg()) -> None:
-    if args.extract_plain_text():
+    if args.extract_plain_text().strip():
         matcher.set_arg("route", args)
 
 
 @RSSHUB_ADD.got("name", prompt="请输入要订阅的订阅名")
 async def handle_feed_name(name: str = ArgPlainText("name")) -> None:
+    if not name.strip():
+        await RSSHUB_ADD.reject("订阅名不能为空，请重新输入")
+        return
+
     if _ := Rss.get_one_by_name(name=name):
         await RSSHUB_ADD.reject(f"已存在名为 {name} 的订阅，请重新输入")
 
@@ -51,6 +55,10 @@ async def handle_feed_name(name: str = ArgPlainText("name")) -> None:
 async def handle_rsshub_routes(
     state: T_State, route: str = ArgPlainText("route")
 ) -> None:
+    if not route.strip():
+        await RSSHUB_ADD.reject("路由名不能为空，请重新输入")
+        return
+
     rsshub_url = URL(config.rsshub)
     # 对本机部署的 RSSHub 不使用代理
     local_host = [
@@ -71,7 +79,7 @@ async def handle_rsshub_routes(
             rsshub_routes = await resp.json()
 
     if route not in rsshub_routes["data"]:
-        await RSSHUB_ADD.reject("没有这个路由，请重新输入")
+        await RSSHUB_ADD.reject(f"未找到名为 {route} 的 RSSHub 路由，请重新输入")
     else:
         route_list = state["route_list"] = rsshub_routes["data"][route]["routes"]
         if len(route_list) > 1:

src/plugins/ELF_RSS2/config.py

@@ -21,6 +21,7 @@ class Config:
     limit: int = 200
     max_length: int = 1024  # 正文长度限制，防止消息太长刷屏，以及消息过长发送失败的情况
     enable_boot_message: bool = True  # 是否启用启动时的提示消息推送
+    debug: bool = False  # 是否开启 debug 模式，开启后会打印更多的日志信息，同时检查更新时不会使用缓存,便于调试
 
     zip_size: int = 2 * 1024
     gif_zip_size: int = 6 * 1024

src/plugins/ELF_RSS2/my_trigger.py

@@ -4,6 +4,7 @@
 from apscheduler.executors.pool import ProcessPoolExecutor, ThreadPoolExecutor
 from apscheduler.triggers.cron import CronTrigger
 from apscheduler.triggers.interval import IntervalTrigger
+from async_timeout import timeout
 from nonebot import require
 from nonebot.log import logger
 
@@ -13,14 +14,14 @@
 require("nonebot_plugin_apscheduler")
 from nonebot_plugin_apscheduler import scheduler  # noqa
 
-wait_for = 5 * 60
-
 
 # 检测某个rss更新
 async def check_update(rss: Rss) -> None:
     logger.info(f"{rss.name} 检查更新")
     try:
-        await asyncio.wait_for(rss_parsing.start(rss), timeout=wait_for)
+        wait_for = 5 * 60 if re.search(r"[_*/,-]", rss.time) else int(rss.time) * 60
+        async with timeout(wait_for):
+            await rss_parsing.start(rss)
     except asyncio.TimeoutError:
         logger.error(f"{rss.name} 检查更新超时，结束此次任务!")
 

src/plugins/ELF_RSS2/parsing/__init__.py

@@ -267,18 +267,42 @@ async def handle_date(item: Dict[str, Any]) -> str:
     return f"日期：{date.format('YYYY年MM月DD日 HH:mm:ss')}"
 
 
+# 发送消息
+@ParsingBase.append_handler(parsing_type="after")
+async def handle_message(
+    rss: Rss,
+    state: Dict[str, Any],
+    item: Dict[str, Any],
+    item_msg: str,
+) -> str:
+    if rss.send_forward_msg:
+        return ""
+
+    # 发送消息并写入文件
+    await handle_send_msgs(rss=rss, messages=[item_msg], items=[item], state=state)
+    return ""
+
+
 @ParsingBase.append_after_handler()
 async def after_handler(rss: Rss, state: Dict[str, Any]) -> Dict[str, Any]:
-    # 发送消息并写入文件
-    await handle_send_msgs(
-        rss=rss, messages=state["messages"], items=state["items"], state=state
-    )
+    if rss.send_forward_msg:
+        # 发送消息并写入文件
+        await handle_send_msgs(
+            rss=rss, messages=state["messages"], items=state["items"], state=state
+        )
 
-    message_count = len(state["messages"])
     db = state["tinydb"]
+    new_data_length = len(state["new_data"])
+    cache_json_manage(db, new_data_length)
+
+    message_count = len(state["change_data"])
+    success_count = message_count - state["error_count"]
 
-    if state["success_count"] > 0:
-        logger.info(f"{rss.name} 新消息推送完毕，共计：{message_count}")
+    if message_count > 10 and len(state["messages"]) == 10:
+        return {}
+
+    if success_count > 0:
+        logger.info(f"{rss.name} 新消息推送完毕，共计：{success_count}/{message_count}")
     elif message_count > 0:
         logger.error(f"{rss.name} 新消息推送失败，共计：{message_count}")
     else:
@@ -287,8 +311,6 @@ async def after_handler(rss: Rss, state: Dict[str, Any]) -> Dict[str, Any]:
     if conn := state["conn"]:
         conn.close()
 
-    new_data_length = len(state["new_data"])
-    cache_json_manage(db, new_data_length)
     db.close()
 
     return {}

src/plugins/ELF_RSS2/parsing/handle_images.py

@@ -206,27 +206,25 @@ async def handle_img_combo(url: str, img_proxy: bool, rss: Optional[Rss] = None)
         返回当前图片的CQ码,以base64格式编码发送
         如获取图片失败将会提示图片走丢了
     """
-    content = await download_image(url, img_proxy)
-    if content:
+    if content := await download_image(url, img_proxy):
         if rss is not None and rss.download_pic:
             _url = URL(url)
             logger.debug(f"正在保存图片: {url}")
             try:
                 save_image(content=content, file_url=_url, rss=rss)
             except Exception as e:
-                logger.warning(e)
-                logger.warning("在保存图片到本地时出现错误")
-        resize_content = await zip_pic(url, content)
-        if img_base64 := get_pic_base64(resize_content):
-            return f"[CQ:image,file=base64://{img_base64}]"
-    return f"\n图片走丢啦: {url}\n"
+                logger.warning(f"在保存图片到本地时出现错误\nE:{repr(e)}")
+        if resize_content := await zip_pic(url, content):
+            if img_base64 := get_pic_base64(resize_content):
+                return f"[CQ:image,file=base64://{img_base64}]"
+    return f"\n图片走丢啦 链接：[{url}]\n"
 
 
-async def handle_img_combo_with_content(gif_url: str, content: bytes) -> str:
-    resize_content = await zip_pic(gif_url, content)
-    if img_base64 := get_pic_base64(resize_content):
-        return f"[CQ:image,file=base64://{img_base64}]"
-    return "\n图片走丢啦\n"
+async def handle_img_combo_with_content(url: str, content: bytes) -> str:
+    if resize_content := await zip_pic(url, content):
+        if img_base64 := get_pic_base64(resize_content):
+            return f"[CQ:image,file=base64://{img_base64}]"
+    return f"\n图片走丢啦 链接：[{url}]\n" if url else "\n图片走丢啦\n"
 
 
 # 处理图片、视频

src/plugins/ELF_RSS2/parsing/parsing_rss.py

@@ -3,11 +3,10 @@
 from typing import Any, Callable, Dict, List, Optional, Tuple
 
 from tinydb import TinyDB
-from tinydb.middlewares import CachingMiddleware
-from tinydb.storages import JSONStorage
 
 from ..config import DATA_PATH
 from ..rss_class import Rss
+from ..utils import partition_list
 
 
 # 订阅器启动的时候将解析器注册到rss实例类？，避免每次推送时再匹配
@@ -55,6 +54,7 @@ class ParsingBase:
         "source": [],
         "date": [],
         "torrent": [],
+        "after": [],  # item的最后处理，此处调用消息截取、发送
     }
 
     """
@@ -188,7 +188,6 @@ async def start(self, rss_name: str, new_rss: Dict[str, Any]) -> None:
         _file = DATA_PATH / f"{Rss.handle_name(rss_name)}.json"
         db = TinyDB(
             _file,
-            storage=CachingMiddleware(JSONStorage),  # type: ignore
             encoding="utf-8",
             sort_keys=True,
             indent=4,
@@ -201,6 +200,7 @@ async def start(self, rss_name: str, new_rss: Dict[str, Any]) -> None:
                 "change_data": [],  # 更新的消息列表
                 "conn": None,  # 数据库连接
                 "tinydb": db,  # 缓存 json
+                "error_count": 0,  # 消息发送失败计数
             }
         )
         self.state, _ = await _run_handlers(self.before_handler, self.rss, self.state)
@@ -213,26 +213,30 @@ async def start(self, rss_name: str, new_rss: Dict[str, Any]) -> None:
                 "items": [],
             }
         )
-        for item in self.state["change_data"]:
-            item_msg = ""
-            for handler_list in self.handler.values():
-                # 用于保存上一次处理结果
-                tmp = ""
-                tmp_state = {"continue": True}  # 是否继续执行后续处理
-
-                # 某一个内容的处理如正文，传入原文与上一次处理结果，此次处理完后覆盖
-                _, tmp = await _run_handlers(
-                    handler_list,
-                    self.rss,
-                    self.state,
-                    item=item,
-                    item_msg=item_msg,
-                    tmp=tmp,
-                    tmp_state=tmp_state,
-                )
-                item_msg += tmp
-            self.state["messages"].append(item_msg)
-            self.state["items"].append(item)
-
-        # 最后处理
-        self.state, _ = await _run_handlers(self.after_handler, self.rss, self.state)
+        if change_data := self.state["change_data"]:
+            for parted_item_list in partition_list(change_data, 10):
+                for item in parted_item_list:
+                    item_msg = ""
+                    for handler_list in self.handler.values():
+                        # 用于保存上一次处理结果
+                        tmp = ""
+                        tmp_state = {"continue": True}  # 是否继续执行后续处理
+
+                        # 某一个内容的处理如正文，传入原文与上一次处理结果，此次处理完后覆盖
+                        _, tmp = await _run_handlers(
+                            handler_list,
+                            self.rss,
+                            self.state,
+                            item=item,
+                            item_msg=item_msg,
+                            tmp=tmp,
+                            tmp_state=tmp_state,
+                        )
+                        item_msg += tmp
+                    self.state["messages"].append(item_msg)
+                    self.state["items"].append(item)
+
+                _, _ = await _run_handlers(self.after_handler, self.rss, self.state)
+                self.state["messages"] = self.state["items"] = []
+        else:
+            _, _ = await _run_handlers(self.after_handler, self.rss, self.state)