up

Author: lgc2333

README.md

@@ -156,7 +156,7 @@ Telegram：[@lgc2333](https://t.me/lgc2333)
 
 - HTML 渲染方案
 
-### [故梦 API](https://api.gumengya.com) & [LoliApi](https://docs.loliapi.com/) & [Lolicon API](https://api.lolicon.app/)
+### [LoliApi](https://docs.loliapi.com/) & [Lolicon API](https://api.lolicon.app/)
 
 - 背景图来源
 

examples/external_example/README.md

@@ -0,0 +1,10 @@
+# External Example
+
+这是一个如何为本插件增加自定义内容的插件示例
+
+你可以直接把此文件夹当做插件包扔给 nonebot2 加载，并按照下方说明更改配置项查看效果
+
+```properties
+PS_TEMPLATE=example_template
+PS_BG_PROVIDER=lgc_icon
+```

examples/external_example/__init__.py

@@ -0,0 +1,11 @@
+# ruff: noqa: E402
+
+from nonebot import require
+
+require("nonebot_plugin_picstatus")  # 别忘 require
+
+from . import (
+    bg_provider as bg_provider,
+    collectors as collectors,
+    templates as templates,
+)

examples/external_example/bg_provider.py

@@ -0,0 +1,23 @@
+from httpx import AsyncClient
+from nonebot_plugin_picstatus.bg_provider import BgData, bg_provider, resp_to_bg_data
+from nonebot_plugin_picstatus.config import config
+
+# 添加自定义背景源演示
+# 实际应用例请见 https://github.com/lgc-NB2Dev/nonebot-plugin-picstatus/blob/master/nonebot_plugin_picstatus/bg_provider.py
+
+
+# 需要用 bg_provider 装饰器注册函数为背景源
+# 背景源名称默认为函数名，当然你也可以手动指定名称，比如
+# @bg_provider("lgc_icon")
+@bg_provider()
+async def lgc_icon() -> BgData:
+    async with AsyncClient(
+        follow_redirects=True,
+        proxies=config.proxy,
+        timeout=config.ps_req_timeout,
+    ) as cli:
+        return resp_to_bg_data(
+            (
+                await cli.get("https://blog.lgc2333.top/assets/favicon.png")
+            ).raise_for_status(),
+        )

examples/external_example/collectors.py

@@ -0,0 +1,108 @@
+import random
+from typing_extensions import override
+
+from nonebot_plugin_picstatus.collectors import (
+    TimeBasedCounterCollector,
+    collector,
+    first_time_collector,
+    normal_collector,
+    periodic_collector,
+)
+
+# 添加自定义数据源展示
+# 实际应用例请见 https://github.com/lgc-NB2Dev/nonebot-plugin-picstatus/tree/master/nonebot_plugin_picstatus/collectors
+
+# PicStatus 内置了三种可以直接用装饰器注册的 collector
+
+# region 1. normal_collector
+
+counter_count = 0
+
+
+# 该 collector 在每次生成状态图片时都会被调用
+# 使用该装饰器时，默认使用函数名作为数据源名称，你也可以手动指定，例如
+# @normal_collector("counter")
+@normal_collector()
+async def counter():
+    global counter_count
+    counter_count += 1
+    return counter_count
+
+
+# endregion
+
+# region 2. first_time_collector
+
+first_time_counter_count = 0
+
+
+# 该 collector 只会在 nonebot 启动时被调用一次并缓存结果
+# 下面的例子中，该 collector 的结果将会始终为 1
+# 装饰器使用教程同上
+@first_time_collector()
+async def first_time_counter():
+    global first_time_counter_count
+    first_time_counter_count += 1
+    return first_time_counter_count
+
+
+# endregion
+
+# region 3. periodic_collector
+
+periodic_counter_count = 0
+
+
+# 该 collector 每隔一定时间被调用一次，并保存一定数量的结果在 deque 中
+# 调用的间隔与保留的结果数量可以在 PicStatus 的配置中找到
+# 装饰器使用教程同上
+@periodic_collector()
+async def periodic_counter():
+    global periodic_counter_count
+    periodic_counter_count += 1
+    return periodic_counter_count
+
+
+# endregion
+
+
+# 此外 PicStatus 内置了另外一种较特殊的 collector
+# 它不可以直接用装饰器装饰一个函数使用
+
+
+# region 4. TimeBasedCounterCollector
+
+
+# 该 collector 基于 PeriodicCollector
+# 它会记录该次调用与上次调用的时间间隔
+# 你可以根据这段时间间隔来计算你希望展示并缓存的值
+# 这对展示 IO 速度等数据可能会很有帮助
+
+
+# 使用此 collector 你需要继承两个 abstract method
+# 并使用 collector 装饰器注册，注意在这里 collector 名称必须手动指定
+# 你也可以使用 collector 装饰器来注册上述 collector，不过具体操作请自行看插件源码理解
+@collector("time_counter")
+class TestTimeBasedCounter(TimeBasedCounterCollector[int, str]):
+    @override
+    async def _calc(self, past: int, now: int, time_passed: float) -> str:
+        """
+        此方法计算你最终想要返回的结果
+
+        Args:
+            past: 上次调用时返回的原始结果
+            now: 本次调用时返回的原始结果
+            time_passed: 从上次调用到本次调用的时间间隔（秒）
+
+        Returns:
+            处理后的结果，和 PeriodicCollector 一样保存在 deque 中
+        """
+        return f"{(now - past) / time_passed:.2f}/s"
+
+    @override
+    async def _get_obj(self) -> int:
+        """此方法返回传入 _calc 方法中的原始结果"""
+        return random.randint(0, 100)
+
+
+# endregion

examples/external_example/templates/__init__.py

@@ -0,0 +1,10 @@
+def _import():
+    from pathlib import Path
+
+    from cookit import auto_import
+
+    auto_import(Path(__file__).parent, __package__)
+
+
+_import()
+del _import

examples/external_example/templates/example_template/__init__.py

@@ -0,0 +1,93 @@
+# ruff: noqa: E402
+
+import json
+from collections import deque
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+import jinja2 as jj
+from cookit.jinja import make_register_jinja_filter_deco
+from nonebot import get_plugin_config, require
+from nonebot_plugin_picstatus.bg_provider import BgData
+from nonebot_plugin_picstatus.templates import pic_template
+from nonebot_plugin_picstatus.templates.pw_render import (
+    ROUTE_URL,
+    add_background_router,
+    add_root_router,
+    base_router_group,
+    register_global_filter_to,
+)
+from pydantic import BaseModel
+
+# 添加自定义图片模板示例
+# 示例应用例请见 https://github.com/lgc-NB2Dev/nonebot-plugin-picstatus/blob/master/nonebot_plugin_picstatus/templates/default/__init__.py
+
+# 可以使用你喜欢的任意库来绘图
+# 示例使用 playwright，你也可以使用 Pillow 等
+# 本插件中自带一些使用 playwright 时绘图有帮助的工具函数
+# 详见 https://github.com/lgc-NB2Dev/nonebot-plugin-picstatus/blob/master/nonebot_plugin_picstatus/templates/pw_render.py
+
+require("nonebot_plugin_htmlrender")
+
+from nonebot_plugin_htmlrender import get_new_page
+
+RES_DIR = Path(__file__).parent / "res"
+
+template_env = jj.Environment(
+    loader=jj.FileSystemLoader(RES_DIR),
+    autoescape=True,
+    enable_async=True,
+)
+# 将插件提供的实用模板 filter 注册到模板环境中
+register_global_filter_to(template_env)
+
+
+jinja_filter = make_register_jinja_filter_deco(template_env)
+
+
+@jinja_filter
+def json_dumps(obj: Any, **kwargs) -> str:
+    if isinstance(obj, dict):
+        obj = {k: list(v) if isinstance(v, deque) else v for k, v in obj.items()}
+    return json.dumps(obj, **kwargs)
+
+
+# 复制一份插件自带的路由组以便在该份代码范围中增删
+template_router_group = base_router_group.copy()
+
+
+# 可以把模板特定配置项声明在这里
+# 你也可以不在这里写，换成写到插件的 config.py 文件中，看个人喜好了
+class TemplateConfig(BaseModel):
+    ps_example_template_example_config: Optional[str] = "example"
+
+
+template_config = get_plugin_config(TemplateConfig)
+
+
+# 注册模板渲染函数
+# name 参数为模板名称，不提供时使用函数名
+# collecting 传入需要启用的 collector 名称集合，如不提供则启用所有 collector
+@pic_template(
+    collecting={
+        "counter",
+        "first_time_counter",
+        "periodic_counter",
+        "time_counter",
+    },
+)
+async def example_template(collected: Dict[str, Any], bg: BgData, **_):
+    template = template_env.get_template("index.html.jinja")
+    html = await template.render_async(d=collected, config=template_config)
+
+    # copy 一份以添加这次图片渲染中特定的 router
+    router_group = template_router_group.copy()
+    add_root_router(router_group, html)  # 注册根路由，访问返回渲好的 html
+    add_background_router(router_group, bg)  # 注册背景图片路由
+
+    async with get_new_page() as page:
+        await router_group.apply(page)
+        await page.goto(f"{ROUTE_URL}/", wait_until="load")
+        elem = await page.wait_for_selector("main")
+        assert elem
+        return await elem.screenshot(type="jpeg")

examples/external_example/templates/example_template/res/index.html.jinja

@@ -0,0 +1,44 @@
+<!DOCTYPE html>
+<html lang="en">
+  <head>
+    <meta charset="UTF-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <style>
+      main {
+        background: linear-gradient(to bottom right, #159957, #155799);
+        width: 600px;
+        font-family: sans-serif;
+      }
+
+      pre,
+      code {
+        font-family: monospace;
+      }
+
+      .box {
+        padding: 20px;
+        background-color: #fff8;
+      }
+    </style>
+  </head>
+  <body>
+    <main>
+      <div style="padding: 20px">
+        <div class="box">
+          <div style="text-align: center">
+            <h1>Example Template</h1>
+            <p>This is an example status picture template.</p>
+          </div>
+          <div style="margin: auto; width: fit-content">
+            <pre><code>d = {{ d | json_dumps(indent=2) }}</code></pre>
+            <pre><code>config = {{ config.model_dump_json(indent=2) }}</code></pre>
+          </div>
+          <div style="text-align: center">
+            <!-- 使用 router 获取背景图 -->
+            <img src="/api/background" style="max-width: 300px; height: auto" />
+          </div>
+        </div>
+      </div>
+    </main>
+  </body>
+</html>

nonebot_plugin_picstatus/collectors/__init__.py

@@ -24,6 +24,8 @@
 from ..config import config
 
 T = TypeVar("T")
+TI = TypeVar("TI")
+TR = TypeVar("TR")
 TC = TypeVar("TC", bound="Collector")
 TCF = TypeVar("TCF", bound=Callable[[], Awaitable[Any]])
 R = TypeVar("R")
@@ -35,24 +37,29 @@ class SkipCollectError(Exception):
     pass
 
 
-class Collector(Generic[T]):
+class Collector(Generic[TI, TR]):
     @abstractmethod
-    async def _get(self) -> T: ...
+    async def _get(self) -> TI: ...
 
-    async def get(self) -> T:
-        return await self._get()
+    @abstractmethod
+    async def get(self) -> TR: ...
 
 
-class BaseNormalCollector(Collector[T], Generic[T]):
+class BaseNormalCollector(Collector[T, T], Generic[T]):
     def __init__(self) -> None:
         super().__init__()
 
+    @override
+    async def get(self) -> T:
+        return await self._get()
+
 
-class BaseFirstTimeCollector(Collector[T], Generic[T]):
+class BaseFirstTimeCollector(Collector[T, T], Generic[T]):
     def __init__(self) -> None:
         super().__init__()
         self._cached: Union[T, Undefined] = Undefined()
 
+    @override
     async def get(self) -> T:
         if not isinstance(self._cached, Undefined):
             return self._cached
@@ -61,21 +68,18 @@ async def get(self) -> T:
         return data
 
 
-class BasePeriodicCollector(Collector[Deque[T]], Generic[T]):
+class BasePeriodicCollector(Collector[T, Deque[T]], Generic[T]):
     def __init__(self, size: int = config.ps_default_collect_cache_size) -> None:
         super().__init__()
         self.data = deque(maxlen=size)
 
     @override
-    async def _get(self) -> Deque[T]:
+    async def get(self) -> Deque[T]:
         return self.data
 
-    @abstractmethod
-    async def _do_collect(self) -> T: ...
-
     async def collect(self):
         try:
-            data = await self._do_collect()
+            data = await self._get()
         except SkipCollectError:
             return
         except Exception:
@@ -157,7 +161,7 @@ async def _calc(self, past: T, now: T, time_passed: float) -> R: ...
     async def _get_obj(self) -> T: ...
 
     @override
-    async def _do_collect(self) -> R:
+    async def _get(self) -> R:
         past = self.last_obj
         past_time = self.last_time
         time_now = time.time()