feat✨: 微博热搜

zhenxun/plugins/coser.py

@@ -4,7 +4,7 @@ from typing import Tuple
 from nonebot.adapters import Bot
 from nonebot.params import RegexGroup
 from nonebot.plugin import PluginMetadata
-from nonebot_plugin_alconna import Alconna, Args, Arparma, Match, on_alconna
+from nonebot_plugin_alconna import Alconna, Args, Arparma, on_alconna
 from nonebot_plugin_saa import Image, Text
 from nonebot_plugin_session import EventSession
 

zhenxun/plugins/wbtop/__init__.py

@@ -0,0 +1,56 @@
+from nonebot.plugin import PluginMetadata
+from nonebot_plugin_alconna import Alconna, Args, Arparma, Match, on_alconna
+from nonebot_plugin_saa import Image, Text
+from nonebot_plugin_session import EventSession
+
+from zhenxun.configs.path_config import IMAGE_PATH
+from zhenxun.configs.utils import PluginExtraData
+from zhenxun.services.log import logger
+from zhenxun.utils.http_utils import AsyncPlaywright
+
+from .data_source import get_hot_image
+
+__plugin_meta__ = PluginMetadata(
+    name="微博热搜",
+    description="刚买完瓜，在吃瓜现场",
+    usage="""
+    指令：
+        微博热搜：发送实时热搜
+        微博热搜 [id]：截图该热搜页面
+        示例：微博热搜 5
+    """.strip(),
+    extra=PluginExtraData(
+        author="HibiKier & yajiwa",
+        version="0.1",
+    ).dict(),
+)
+
+
+_matcher = on_alconna(Alconna("微博热搜", Args["idx?", int]), priority=5, block=True)
+
+
+@_matcher.handle()
+async def _(session: EventSession, arparma: Arparma, idx: Match[int]):
+    result, data_list = await get_hot_image()
+    if isinstance(result, str):
+        await Text(result).finish(reply=True)
+    if idx.available:
+        _idx = idx.result
+        url = data_list[_idx - 1]["url"]
+        file = IMAGE_PATH / "temp" / f"wbtop_{session.id1}.png"
+        img = await AsyncPlaywright.screenshot(
+            url,
+            file,
+            "#pl_feed_main",
+            wait_time=12,
+        )
+        if img:
+            await Image(file).send()
+            logger.info(
+                f"查询微博热搜 Id: {_idx}", arparma.header_result, session=session
+            )
+        else:
+            await Text("获取图片失败...").send()
+    else:
+        await Image(result.pic2bytes()).send()
+        logger.info(f"查询微博热搜", arparma.header_result, session=session)

zhenxun/plugins/wbtop/data_source.py

@@ -0,0 +1,63 @@
+from zhenxun.configs.path_config import IMAGE_PATH
+from zhenxun.services.log import logger
+from zhenxun.utils.http_utils import AsyncHttpx
+from zhenxun.utils.image_utils import BuildImage
+
+URL = "https://weibo.com/ajax/side/hotSearch"
+
+
+async def get_data() -> list | str:
+    """获取数据
+
+    返回:
+        list | str: 数据或消息
+    """
+    data_list = []
+    for _ in range(3):
+        try:
+            response = await AsyncHttpx.get(URL, timeout=20)
+            if response.status_code == 200:
+                data_json = response.json()["data"]["realtime"]
+                for item in data_json:
+                    if "is_ad" in item:
+                        """广告跳过"""
+                        continue
+                    data = {
+                        "hot_word": item["note"],
+                        "hot_word_num": str(item["num"]),
+                        "url": "https://s.weibo.com/weibo?q=%23" + item["word"] + "%23",
+                    }
+                    data_list.append(data)
+                if not data:
+                    return "没有搜索到..."
+                return data_list
+        except Exception as e:
+            logger.error("获取微博热搜错误", e=e)
+    return "获取失败,请十分钟后再试..."
+
+
+async def get_hot_image() -> tuple[BuildImage | str, list]:
+    """构造图片
+
+    返回:
+        BuildImage | str: 热搜图片
+    """
+    data = await get_data()
+    if isinstance(data, str):
+        return data, []
+    bk = BuildImage(700, 32 * 50 + 280, color="#797979")
+    wbtop_bk = BuildImage(700, 280, background=f"{IMAGE_PATH}/other/webtop.png")
+    await bk.paste(wbtop_bk)
+    text_bk = BuildImage(700, 32 * 50, color="#797979")
+    image_list = []
+    for i, data in enumerate(data):
+        title = f"{i + 1}. {data['hot_word']}"
+        hot = str(data["hot_word_num"])
+        img = BuildImage(700, 30, font_size=20)
+        w, h = img.getsize(title)
+        await img.text((10, int((30 - h) / 2)), title)
+        await img.text((580, int((30 - h) / 2)), hot)
+        image_list.append(img)
+    text_bk = await text_bk.auto_paste(image_list, 1, 2, 0)
+    await bk.paste(text_bk, (0, 280))
+    return bk, data