feat: experimental htmlkit

Author: ProgramRipper

.json

@@ -0,0 +1,18 @@
+{
+    "AppClientVersion": 39038,
+    "AppId": 1600001615,
+    "AppIdQrCode": 537313942,
+    "CurrentVersion": "3.2.19-39038",
+    "Kernel": "Linux",
+    "MainSigMap": 169742560,
+    "MiscBitmap": 32764,
+    "NTLoginType": 1,
+    "Os": "Linux",
+    "PackageName": "com.tencent.qq",
+    "PtVersion": "2.0.0",
+    "SsoVersion": 19,
+    "SubAppId": 537313942,
+    "SubSigMap": 0,
+    "VendorOs": "linux",
+    "WtLoginSdk": "nt.wtlogin.0.0.1"
+}

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:alpine AS base
+FROM python:3.13-alpine AS base
 WORKDIR /app
 
 

pdm.lock

@@ -0,0 +1,67 @@
+# /// script
+# requires-python = ">=3.10"
+# dependencies = [
+#     "httpx",
+#     "starlette",
+#     "uvicorn",
+#     "uvloop",
+# ]
+# ///
+import logging
+
+from httpx import AsyncClient
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+
+logging.basicConfig(level=logging.DEBUG)
+logging.getLogger("httpx").setLevel(logging.DEBUG)
+
+
+client = AsyncClient()
+app = Starlette()
+
+
+@app.route("/39038/appinfo_v2")
+async def appinfo_v2(_: Request) -> JSONResponse:
+    return JSONResponse(
+        {
+            "Os": "Linux",
+            "Kernel": "Linux",
+            "VendorOs": "linux",
+            "VendorOs": "linux",
+            "CurrentVersion": "3.2.19-39038",
+            "PtVersion": "2.0.0",
+            "SsoVersion": 19,
+            "PackageName": "com.tencent.qq",
+            "ApkSignatureMd5": "Y29tLnRlbmNlbnQucXE=",
+            "SdkInfo": {
+                "SdkBuildTime": 0,
+                "SdkVersion": "nt.wtlogin.0.0.1",
+                "MiscBitMap": 32764,
+                "SubSigMap": 0,
+                "MainSigMap": 169742560,
+            },
+            "AppId": 1600001615,
+            "SubAppId": 537313942,
+            "AppClientVersion": 39038,
+        }
+    )
+
+
+@app.route("/39038", ["GET", "POST"])
+async def _(request: Request) -> Response:
+    response = await client.request(
+        method=request.method,
+        url=f"http://103.217.204.57:39038/sign",
+        headers={key: value for key, value in request.headers.items() if key != "host"},
+        params=request.query_params,
+        content=await request.body(),
+    )
+    return Response(response.read(), response.status_code, response.headers)
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="0.0.0.0")

proxy.py

@@ -3,13 +3,14 @@ name = ""
 version = "0.1.0"
 description = "Twelve"
 readme = "README.md"
-requires-python = "~=3.12"
+requires-python = "~=3.13"
 dependencies = [
     "backoff~=2.2",
     "nonebot2[fastapi,httpx,websockets]~=2.4",
-    "nonebot-adapter-milky~=0.5.0rc3",
-    "nonebot-plugin-alconna~=0.59",
+    "nonebot-adapter-milky~=1.0",
+    "nonebot-plugin-alconna~=0.60",
     "nonebot-plugin-apscheduler~=0.5",
+    "nonebot-plugin-htmlkit~=0.1.0rc4",
     "nonebot-plugin-htmlrender~=0.6",
     "nonebot-plugin-orm[sqlite]~=0.8",
     "nonebot-plugin-uninfo @ https://github.com/RF-Tar-Railt/nonebot-plugin-uninfo/archive/5b87d916af09d6a878039d3445193efda2ed1af3.zip",
@@ -30,6 +31,7 @@ adapters = [
 plugins = [
     "nonebot_plugin_alconna",
     "nonebot_plugin_apscheduler",
+    "nonebot_plugin_htmlkit",
     "nonebot_plugin_htmlrender",
     "nonebot_plugin_orm",
     "nonebot_plugin_uninfo",

pyproject.toml

@@ -1,5 +1,6 @@
 from asyncio import gather
 from contextlib import asynccontextmanager
+from importlib.resources import as_file, files
 from queue import PriorityQueue
 from typing import Annotated, Any, AsyncGenerator
 
@@ -11,6 +12,7 @@
 from nonebot.plugin import PluginMetadata
 from nonebot_plugin_alconna import Alconna, Image, Subcommand, UniMessage, on_alconna
 from nonebot_plugin_apscheduler import scheduler
+from nonebot_plugin_htmlkit import template_to_pic
 from nonebot_plugin_htmlrender.browser import get_browser
 from nonebot_plugin_orm import async_scoped_session, get_session
 from nonebot_plugin_uninfo import MEMBER
@@ -21,6 +23,7 @@
 from .....utils import run_task, send_message
 from ... import plugin_config as bilibili_config
 from ...utils import UID_ARG, get_share_click, handle_error, raise_for_status
+from . import templates
 from .config import Config
 from .models import Dynamic, Dynamics, Subscription
 
@@ -82,7 +85,19 @@ async def get_dynamics(page: int = 1) -> Dynamics:
                 "type": "all",
                 "page": page,
                 "features": ",".join(
-                    ("itemOpusStyle", "listOnlyfans", "opusBigCover", "onlyfansVote")
+                    (
+                        "itemOpusStyle",
+                        "listOnlyfans",
+                        "opusBigCover",
+                        "onlyfansVote",
+                        "decorationCard",
+                        "onlyfansAssetsV2",
+                        "forwardListHidden",
+                        "ugcDelete",
+                        "onlyfansQaCard",
+                        "commentsNewVersion",
+                        "avatarAutoTheme",
+                    )
                 ),
             },
         )
@@ -163,7 +178,7 @@ async def render_screenshot(id_str: str) -> bytes:
         )
         await page.add_style_tag(
             content="""
-                * {
+                body {
                   font-family: "LXGW ZhenKai GB", "LXGW WenKai GB", sans-serif !important;
                 }
 
@@ -198,15 +213,33 @@ async def render_screenshot(id_str: str) -> bytes:
 async def broadcast(dynamics: list[Dynamic]):
     async with get_session() as session:
         for dynamic in dynamics:
-            screenshot, url, subs = await gather(
-                render_screenshot(dynamic["id_str"]),
-                get_share_click(dynamic["id_str"], "dynamic", "dt.dt-detail.0.0.pv"),
-                session.scalars(
-                    select(Subscription).where(
-                        Subscription.uid == dynamic["modules"]["module_author"]["mid"]
-                    )
-                ),
-            )
+            with as_file(files(templates)) as templates_path:
+                screenshot, url, subs = await gather(
+                    (
+                        template_to_pic(
+                            str(templates_path),
+                            "draw.html.j2",
+                            dynamic,
+                            max_width=360 * 3,
+                            device_height=640 * 3,
+                            allow_refit=False,
+                            image_format="jpeg",
+                            jpeg_quality=80,
+                        )
+                        if dynamic["type"] == "DYNAMIC_TYPE_DRAW"
+                        and not dynamic["modules"]["module_dynamic"]["additional"]
+                        else render_screenshot(dynamic["id_str"])
+                    ),
+                    get_share_click(
+                        dynamic["id_str"], "dynamic", "dt.dt-detail.0.0.pv"
+                    ),
+                    session.scalars(
+                        select(Subscription).where(
+                            Subscription.uid
+                            == dynamic["modules"]["module_author"]["mid"]
+                        )
+                    ),
+                )
 
             msg = plugin_config.template.format(
                 name=dynamic["modules"]["module_author"]["name"],
@@ -310,16 +343,48 @@ async def _(id_str: str):
         dynamic = raise_for_status(
             await client.get(
                 "/polymer/web-dynamic/v1/detail",
-                params={"id": id_str, "features": "itemOpusStyle"},
+                params={
+                    "id": id_str,
+                    "features": ",".join(
+                        (
+                            "itemOpusStyle",
+                            "listOnlyfans",
+                            "opusBigCover",
+                            "onlyfansVote",
+                            "decorationCard",
+                            "onlyfansAssetsV2",
+                            "forwardListHidden",
+                            "ugcDelete",
+                            "onlyfansQaCard",
+                            "commentsNewVersion",
+                            "avatarAutoTheme",
+                        )
+                    ),
+                },
             )
         )["item"]
     except Exception:
         await handle_error("获取动态信息失败")
 
-    screenshot, url = await gather(
-        render_screenshot(id_str),
-        get_share_click(id_str, "dynamic", "dt.dt-detail.0.0.pv"),
-    )
+    with as_file(files(templates)) as templates_path:
+        screenshot, url = await gather(
+            (
+                template_to_pic(
+                    str(templates_path),
+                    "draw.html.j2",
+                    dynamic,
+                    max_width=360 * 3,
+                    device_height=640 * 3,
+                    allow_refit=False,
+                    image_format="jpeg",
+                    jpeg_quality=80,
+                )
+                if dynamic["type"] == "DYNAMIC_TYPE_DRAW"
+                and not dynamic["modules"]["module_dynamic"]["additional"]
+                else render_screenshot(dynamic["id_str"])
+            ),
+            get_share_click(id_str, "dynamic", "dt.dt-detail.0.0.pv"),
+        )
     await plugin_config.template.format(
         name=dynamic["modules"]["module_author"]["name"],
         action=dynamic["modules"]["module_author"]["pub_action"]

src/plugins/bilibili/plugins/dynamic/__init__.py

@@ -1,4 +1,4 @@
-from typing import TypedDict
+from typing import Any, TypedDict
 
 from nonebot_plugin_orm import Model
 from nonebot_plugin_uninfo.orm import SceneModel
@@ -14,6 +14,7 @@ class ModuleAuthor(TypedDict):
 
 class Modules(TypedDict):
     module_author: ModuleAuthor
+    module_dynamic: dict[str, Any]
 
 
 class Dynamic(TypedDict):