feat✨: add bt

zhenxun/plugins/ai/__init__.py

@@ -16,33 +16,6 @@ from zhenxun.utils.depends import UserName
 
 from .data_source import get_chat_result, hello, no_result
 
-__zx_plugin_name__ = "AI"
-__plugin_usage__ = f"""
-usage：
-    与{NICKNAME}普普通通的对话吧！
-"""
-__plugin_version__ = 0.1
-__plugin_author__ = "HibiKier"
-__plugin_settings__ = {
-    "level": 5,
-    "cmd": ["Ai", "ai", "AI", "aI"],
-}
-__plugin_configs__ = {
-    "TL_KEY": {"value": [], "help": "图灵Key", "type": List[str]},
-    "ALAPI_AI_CHECK": {
-        "value": False,
-        "help": "是否检测青云客骂娘回复",
-        "default_value": False,
-        "type": bool,
-    },
-    "TEXT_FILTER": {
-        "value": ["鸡", "口交"],
-        "help": "文本过滤器，将敏感词更改为*",
-        "default_value": [],
-        "type": List[str],
-    },
-}
-
 __plugin_meta__ = PluginMetadata(
     name="AI",
     description="屑Ai",

zhenxun/plugins/bt/__init__.py

@@ -0,0 +1,78 @@
+from asyncio.exceptions import TimeoutError
+
+from httpx import ConnectTimeout
+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.utils import PluginExtraData, RegisterConfig
+from zhenxun.services.log import logger
+from zhenxun.utils.rules import ensure_private
+
+from .data_source import get_bt_info
+
+__plugin_meta__ = PluginMetadata(
+    name="磁力搜索",
+    description="bt(磁力搜索)[仅支持私聊，懂的都懂]",
+    usage="""
+    * 拒绝反冲斗士！ *
+    指令:
+        bt [关键词] ?[页数]
+        示例：bt 钢铁侠
+        示例：bt 钢铁侠 3
+    """.strip(),
+    extra=PluginExtraData(
+        author="HibiKier",
+        version="0.1",
+        configs=[
+            RegisterConfig(
+                key="BT_MAX_NUM",
+                value=10,
+                help="单次BT搜索返回最大消息数量",
+                default_value=10,
+                type=int,
+            ),
+        ],
+    ).dict(),
+)
+
+
+_matcher = on_alconna(
+    Alconna("bt", Args["keyword", str]["page?", int]),
+    rule=ensure_private,
+    priority=5,
+    block=True,
+)
+
+
+@_matcher.handle()
+async def _(
+    session: EventSession,
+    arparma: Arparma,
+    keyword: str,
+    page: Match[int],
+):
+    send_flag = False
+    try:
+        async for title, type_, create_time, file_size, link in get_bt_info(
+            keyword, page.result if page.available else 1
+        ):
+            await Text(
+                f"标题：{title}\n"
+                f"类型：{type_}\n"
+                f"创建时间：{create_time}\n"
+                f"文件大小：{file_size}\n"
+                f"种子：{link}"
+            ).send()
+            send_flag = True
+    except (TimeoutError, ConnectTimeout):
+        await Text(f"搜索 {keyword} 超时...").finish()
+    except Exception as e:
+        logger.error(f"bt 错误", arparma.header_result, session=session, e=e)
+        await Text(f"bt 其他未知错误..").finish()
+    if not send_flag:
+        await Text(f"{keyword} 未搜索到...").send()
+    logger.info(
+        f"BT搜索 {keyword} 第 {page} 页", arparma.header_result, session=session
+    )

zhenxun/plugins/bt/data_source.py

@@ -0,0 +1,54 @@
+from bs4 import BeautifulSoup
+
+from zhenxun.configs.config import Config
+from zhenxun.utils.http_utils import AsyncHttpx, AsyncPlaywright
+
+url = "http://www.eclzz.ink"
+
+
+async def get_bt_info(keyword: str, page: int):
+    """获取资源信息
+
+    参数:
+        keyword: 关键词
+        page: 页数
+    """
+    global url
+    text = (await AsyncHttpx.get(f"{url}/s/{keyword}_rel_{page}.html", timeout=30)).text
+    if "301 Moved Permanently" in text:
+        async with AsyncPlaywright.new_page() as _page:
+            await _page.goto(url)
+            url = _page.url
+        text = (
+            await AsyncHttpx.get(f"{url}/s/{keyword}_rel_{page}.html", timeout=30)
+        ).text
+    if "大约0条结果" in text:
+        return
+    soup = BeautifulSoup(text, "lxml")
+    item_lst = soup.find_all("div", {"class": "search-item"})
+    bt_max_num = Config.get_config("bt", "BT_MAX_NUM") or 10
+    bt_max_num = bt_max_num if bt_max_num < len(item_lst) else len(item_lst)
+    for item in item_lst[:bt_max_num]:
+        divs = item.find_all("div")
+        title = (
+            str(divs[0].find("a").text).replace("<em>", "").replace("</em>", "").strip()
+        )
+        spans = divs[2].find_all("span")
+        type_ = spans[0].text
+        create_time = spans[1].find("b").text
+        file_size = spans[2].find("b").text
+        link = await get_download_link(divs[0].find("a")["href"])
+        yield title, type_, create_time, file_size, link
+
+
+async def get_download_link(_url: str) -> str | None:
+    """获取资源下载地址
+
+    参数:
+        _url: 链接
+    """
+    text = (await AsyncHttpx.get(f"{url}{_url}")).text
+    soup = BeautifulSoup(text, "lxml")
+    if fd := soup.find("a", {"id": "down-url"}):
+        return fd["href"]  # type: ignore
+    return None