feat✨: search_anime

zhenxun/plugins/search_anime/__init__.py

@@ -0,0 +1,67 @@
+from nonebot.adapters import Bot
+from nonebot.plugin import PluginMetadata
+from nonebot_plugin_alconna import Alconna, Args, Arparma, Match, on_alconna
+from nonebot_plugin_saa import Text
+from nonebot_plugin_session import EventSession
+
+from zhenxun.configs.config import Config
+from zhenxun.configs.utils import BaseBlock, PluginExtraData, RegisterConfig
+from zhenxun.services.log import logger
+
+from .data_source import from_anime_get_info
+
+__plugin_meta__ = PluginMetadata(
+    name="搜番",
+    description="找不到想看的动漫吗？",
+    usage="""
+    搜索动漫资源
+    指令：
+        搜番  [番剧名称或者关键词]
+        示例：搜番 刀剑神域
+    """.strip(),
+    extra=PluginExtraData(
+        author="HibiKier",
+        version="0.1",
+        menu_type="一些工具",
+        limits=[BaseBlock(result="搜索还未完成，不要重复触发！")],
+        configs=[
+            RegisterConfig(
+                key="SEARCH_ANIME_MAX_INFO",
+                value=20,
+                help="搜索动漫返回的最大数量",
+                default_value=20,
+                type=10,
+            )
+        ],
+    ).dict(),
+)
+
+_matcher = on_alconna(Alconna("搜番", Args["name?", str]), priority=5, block=True)
+
+
+@_matcher.handle()
+async def _(name: Match[str]):
+    if name.available:
+        _matcher.set_path_arg("name", name.result)
+
+
+@_matcher.got_path("name", prompt="是不是少了番名？")
+async def _(session: EventSession, arparma: Arparma, name: str):
+    gid = session.id3 or session.id2
+    await Text(f"开始搜番 {name}...").send()
+    anime_report = await from_anime_get_info(
+        name,
+        Config.get_config("search_anime", "SEARCH_ANIME_MAX_INFO"),
+    )
+    if anime_report:
+        if isinstance(anime_report, str):
+            await Text(anime_report).finish()
+        await Text("\n\n".join(anime_report)).send()
+        logger.info(
+            f"搜索番剧 {name} 成功: {anime_report}",
+            arparma.header_result,
+            session=session,
+        )
+    else:
+        logger.info(f"未找到番剧 {name}...")
+        await Text(f"未找到番剧 {name}（也有可能是超时，再尝试一下？）").send()

zhenxun/plugins/search_anime/data_source.py

@@ -0,0 +1,53 @@
+import time
+from urllib import parse
+
+import feedparser
+from lxml import etree
+
+from zhenxun.services.log import logger
+from zhenxun.utils.http_utils import AsyncHttpx
+
+
+async def from_anime_get_info(key_word: str, max_: int) -> str | list[str]:
+    s_time = time.time()
+    url = "https://share.dmhy.org/topics/rss/rss.xml?keyword=" + parse.quote(key_word)
+    try:
+        repass = await get_repass(url, max_)
+    except Exception as e:
+        logger.error(f"发生了一些错误 {type(e)}", e=e)
+        return "发生了一些错误！"
+    repass.insert(0, f"搜索 {key_word} 结果（耗时 {int(time.time() - s_time)} 秒）：\n")
+    return repass
+
+
+async def get_repass(url: str, max_: int) -> list[str]:
+    put_line = []
+    text = (await AsyncHttpx.get(url)).text
+    d = feedparser.parse(text)
+    max_ = (
+        max_
+        if max_ < len([e.link for e in d.entries])
+        else len([e.link for e in d.entries])
+    )
+    url_list = [e.link for e in d.entries][:max_]
+    for u in url_list:
+        try:
+            text = (await AsyncHttpx.get(u)).text
+            html = etree.HTML(text)  # type: ignore
+            magent = html.xpath('.//a[@id="a_magnet"]/text()')[0]
+            title = html.xpath(".//h3/text()")[0]
+            item = html.xpath('//div[@class="info resource-info right"]/ul/li')
+            class_a = (
+                item[0]
+                .xpath("string(.)")[5:]
+                .strip()
+                .replace("\xa0", "")
+                .replace("\t", "")
+            )
+            size = item[3].xpath("string(.)")[5:].strip()
+            put_line.append(
+                "【{}】| {}\n【{}】| {}".format(class_a, title, size, magent)
+            )
+        except Exception as e:
+            logger.error(f"搜番发生错误", e=e)
+    return put_line