♻️ refactor(UI): 重构UI渲染服务为组件化分层架构 (#2025)
Some checks failed
检查bot是否运行正常 / bot check (push) Waiting to run
Sequential Lint and Type Check / ruff-call (push) Waiting to run
Sequential Lint and Type Check / pyright-call (push) Blocked by required conditions
Release Drafter / Update Release Draft (push) Waiting to run
Force Sync to Aliyun / sync (push) Waiting to run
Update Version / update-version (push) Waiting to run
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Has been cancelled
CodeQL Code Security Analysis / Analyze (${{ matrix.language }}) (none, python) (push) Has been cancelled

* ♻️ refactor(UI): 重构UI渲染服务为组件化分层架构

♻️ **架构重构**
- UI渲染服务重构为组件化分层架构
- 解耦主题管理、HTML生成、截图功能

 **新增功能**
- `zhenxun.ui` 统一入口,提供 `render`、`markdown`、`vstack` 等API
- `RenderableComponent` 基类和渲染协议抽象
- 新增主题管理器和截图引擎模块

⚙️ **配置优化**
- UI配置迁移至 `superuser/ui_manager.py`
- 新增"重载UI主题"管理指令

🔧 **性能改进**
- 优化渲染缓存,支持组件级透明缓存
- 所有UI组件适配新渲染流程

* 🚨 auto fix by pre-commit hooks

---------

Co-authored-by: webjoin111 <455457521@qq.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Rumio 2025-08-18 23:08:22 +08:00 committed by GitHub
parent 11524bcb04
commit 6124e217d0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
43 changed files with 1334 additions and 928 deletions

View File

@ -14,12 +14,13 @@ from nonebot_plugin_alconna import (
from nonebot_plugin_session import EventSession
import pytz
from zhenxun import ui
from zhenxun.configs.config import Config
from zhenxun.configs.utils import Command, PluginExtraData, RegisterConfig
from zhenxun.models.chat_history import ChatHistory
from zhenxun.models.group_member_info import GroupInfoUser
from zhenxun.services.log import logger
from zhenxun.ui import TableBuilder
from zhenxun.ui.builders import TableBuilder
from zhenxun.ui.models import ImageCell, TextCell
from zhenxun.utils.enum import PluginType
from zhenxun.utils.message import MessageUtils
@ -174,7 +175,7 @@ async def _(
builder = TableBuilder(f"消息排行({count.result})", date_str)
builder.set_headers(column_name).add_rows(rows_data)
image_bytes = await builder.build()
image_bytes = await ui.render(builder.build())
logger.info(
f"查看消息排行 数量={count.result}", arparma.header_result, session=session

View File

@ -5,10 +5,10 @@ from nonebot.plugin import PluginMetadata
from nonebot.rule import Rule, to_me
from nonebot_plugin_alconna import Alconna, on_alconna
from zhenxun import ui
from zhenxun.configs.config import Config
from zhenxun.configs.utils import PluginExtraData, RegisterConfig
from zhenxun.services.log import logger
from zhenxun.services.renderer import renderer_service
from zhenxun.utils.enum import PluginType
from zhenxun.utils.message import MessageUtils
from zhenxun.utils.rules import notice_rule
@ -68,8 +68,9 @@ async def handle_self_check():
try:
data_dict = await get_status_info()
image_bytes = await renderer_service.render(
"pages/builtin/check", data=data_dict
image_bytes = await ui.render_template(
"pages/builtin/check",
data=data_dict,
)
await MessageUtils.build_message(image_bytes).send()

View File

@ -1,6 +1,7 @@
import nonebot
from nonebot_plugin_uninfo import Uninfo
from zhenxun import ui
from zhenxun.configs.config import BotConfig, Config
from zhenxun.configs.path_config import IMAGE_PATH
from zhenxun.configs.utils import PluginExtraData
@ -13,15 +14,14 @@ from zhenxun.services import (
LLMException,
LLMMessage,
generate,
renderer_service,
)
from zhenxun.services.log import logger
from zhenxun.ui import (
from zhenxun.ui.builders import (
InfoCardBuilder,
NotebookBuilder,
PluginMenuBuilder,
PluginMenuCategory,
)
from zhenxun.ui.models import PluginMenuCategory
from zhenxun.utils.common_utils import format_usage_for_markdown
from zhenxun.utils.enum import BlockType, PluginType
from zhenxun.utils.platform import PlatformUtils
@ -120,7 +120,7 @@ async def create_help_img(
PluginMenuCategory(name=category["name"], items=category["items"])
)
return await builder.build()
return await ui.render(builder.build())
async def get_user_allow_help(user_id: str) -> list[PluginType]:
@ -214,7 +214,7 @@ async def get_plugin_help(user_id: str, name: str, is_superuser: bool) -> str |
render_dict = model_dump(builder._data)
render_dict["style_name"] = style_name
return await renderer_service.render("pages/builtin/help", data=render_dict)
return await ui.render_template("pages/builtin/help", data=render_dict)
return "糟糕! 该功能没有帮助喔..."
return "没有查找到这个功能噢..."
@ -295,7 +295,7 @@ async def get_llm_help(question: str, user_id: str) -> str | bytes:
if len(reply_text) > threshold:
builder = NotebookBuilder()
builder.text(reply_text)
return await builder.build()
return await ui.render(builder.build())
return reply_text

View File

@ -5,12 +5,12 @@ from nonebot_plugin_uninfo import Uninfo
from tortoise.expressions import RawSQL
from tortoise.functions import Count
from zhenxun import ui
from zhenxun.models.chat_history import ChatHistory
from zhenxun.models.level_user import LevelUser
from zhenxun.models.sign_user import SignUser
from zhenxun.models.statistics import Statistics
from zhenxun.models.user_console import UserConsole
from zhenxun.services import renderer_service
from zhenxun.utils.platform import PlatformUtils
RACE = [
@ -198,4 +198,4 @@ async def get_user_info(
},
}
return await renderer_service.render("pages/builtin/my_info", data=profile_data)
return await ui.render_template("pages/builtin/my_info", data=profile_data)

View File

@ -1,9 +1,10 @@
from typing import Any
from zhenxun.services import renderer_service
from zhenxun.services.llm.core import KeyStatus
from zhenxun.services.llm.types import ModelModality
from zhenxun.ui import MarkdownBuilder, TableBuilder
from zhenxun.ui.models import StatusBadgeCell, TextCell
from zhenxun.ui.builders import MarkdownBuilder, TableBuilder
from zhenxun.ui.models.core.table import StatusBadgeCell, TextCell
def _format_seconds(seconds: int) -> str:
@ -35,7 +36,7 @@ class Presenters:
builder = TableBuilder(
title=title, tip="当前没有配置任何LLM模型。"
).set_headers(["提供商", "模型名称", "API类型", "状态"])
return await builder.build()
return await renderer_service.render(builder.build())
column_name = ["提供商", "模型名称", "API类型", "状态"]
data_list = []
@ -60,7 +61,7 @@ class Presenters:
)
builder.set_headers(column_name)
builder.add_rows(data_list)
return await builder.build(use_cache=True)
return await renderer_service.render(builder.build(), use_cache=True)
@staticmethod
async def format_model_details_as_markdown_image(details: dict[str, Any]) -> bytes:
@ -99,7 +100,7 @@ class Presenters:
builder.text(f"- **最大Token**: {token_value}")
builder.text(f"- **核心能力**: {', '.join(cap_list) or '纯文本'}")
return await builder.with_style("light").build()
return await renderer_service.render(builder.with_style("light").build())
@staticmethod
async def format_key_status_as_image(
@ -181,4 +182,4 @@ class Presenters:
]
)
builder.add_rows(data_list)
return await builder.build(use_cache=False)
return await renderer_service.render(builder.build(), use_cache=False)

View File

@ -6,9 +6,9 @@ from nonebot_plugin_apscheduler import scheduler
from nonebot_plugin_uninfo import Uninfo
from nonebot_plugin_waiter import prompt_until
from zhenxun import ui
from zhenxun.configs.utils import PluginExtraData, RegisterConfig
from zhenxun.services.log import logger
from zhenxun.services.renderer import renderer_service
from zhenxun.utils.depends import UserName
from zhenxun.utils.message import MessageUtils
from zhenxun.utils.utils import is_number
@ -193,7 +193,7 @@ async def _(session: Uninfo, arparma: Arparma, uname: str = UserName()):
render_data = {"page_type": "user", "payload": user_payload}
image_bytes = await renderer_service.render(
image_bytes = await ui.render_template(
"pages/builtin/mahiro_bank",
data=render_data,
viewport={"width": 386, "height": 10},
@ -209,7 +209,7 @@ async def _(session: Uninfo, arparma: Arparma):
render_data = {"page_type": "overview", "payload": overview_payload}
image_bytes = await renderer_service.render(
image_bytes = await ui.render_template(
"pages/builtin/mahiro_bank",
data=render_data,
viewport={"width": 450, "height": 10},

View File

@ -84,6 +84,7 @@ async def _(session: EventSession):
try:
result = await StoreManager.get_plugins_info()
logger.info("查看插件列表", "插件商店", session=session)
await MessageUtils.build_message([*result]).send()
except Exception as e:
logger.error(f"查看插件列表失败 e: {e}", "插件商店", session=session, e=e)

View File

@ -5,12 +5,14 @@ import shutil
from aiocache import cached
import ujson as json
from zhenxun import ui
from zhenxun.builtin_plugins.plugin_store.models import StorePluginInfo
from zhenxun.configs.path_config import TEMP_PATH
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.services.log import logger
from zhenxun.services.plugin_init import PluginInitManager
from zhenxun.utils.image_utils import BuildImage, ImageTemplate, RowStyle
from zhenxun.ui.builders import TableBuilder
from zhenxun.ui.models import StatusBadgeCell, TextCell
from zhenxun.utils.manager.virtual_env_package_manager import VirtualEnvPackageManager
from zhenxun.utils.repo_utils import RepoFileManager
from zhenxun.utils.repo_utils.models import RepoFileInfo, RepoType
@ -25,22 +27,6 @@ from .config import (
from .exceptions import PluginStoreException
def row_style(column: str, text: str) -> RowStyle:
"""被动技能文本风格
参数:
column: 表头
text: 文本内容
返回:
RowStyle: RowStyle
"""
style = RowStyle()
if column == "-" and text == "已安装":
style.font_color = "#67C23A"
return style
class StoreManager:
@classmethod
@cached(60)
@ -105,61 +91,123 @@ class StoreManager:
return await PluginInfo.filter(load_status=True).values_list(*args)
@classmethod
async def get_plugins_info(cls) -> list[BuildImage] | str:
async def get_plugins_info(cls) -> list[bytes] | str:
"""插件列表
返回:
BuildImage | str: 返回消息
bytes | str: 返回消息
"""
plugin_list, extra_plugin_list = await cls.get_data()
column_name = ["-", "ID", "名称", "简介", "作者", "版本", "类型"]
db_plugin_list = await cls.get_loaded_plugins("module", "version")
suc_plugin = {p[0]: (p[1] or "0.1") for p in db_plugin_list}
HIGHLIGHT_COLOR = "#E6A23C"
structured_native_list = []
structured_extra_list = []
index = 0
data_list = []
extra_data_list = []
for plugin_info in plugin_list:
data_list.append(
[
"已安装" if plugin_info.module in suc_plugin else "",
index,
plugin_info.name,
plugin_info.description,
plugin_info.author,
cls.version_check(plugin_info, suc_plugin),
plugin_info.plugin_type_name,
]
is_new = cls.check_version_is_new(plugin_info, suc_plugin)
structured_native_list.append(
{
"is_installed": plugin_info.module in suc_plugin,
"id": index,
"name": plugin_info.name,
"description": plugin_info.description,
"author": plugin_info.author,
"version_str": cls.version_check(plugin_info, suc_plugin),
"type_name": plugin_info.plugin_type_name,
"has_update": not is_new and plugin_info.module in suc_plugin,
}
)
index += 1
for plugin_info in extra_plugin_list:
extra_data_list.append(
[
"已安装" if plugin_info.module in suc_plugin else "",
index,
plugin_info.name,
plugin_info.description,
plugin_info.author,
cls.version_check(plugin_info, suc_plugin),
plugin_info.plugin_type_name,
]
is_new = cls.check_version_is_new(plugin_info, suc_plugin)
structured_extra_list.append(
{
"is_installed": plugin_info.module in suc_plugin,
"id": index,
"name": plugin_info.name,
"description": plugin_info.description,
"author": plugin_info.author,
"version_str": cls.version_check(plugin_info, suc_plugin),
"type_name": plugin_info.plugin_type_name,
"has_update": not is_new and plugin_info.module in suc_plugin,
}
)
index += 1
return [
await ImageTemplate.table_page(
"原生插件列表",
"通过添加/移除插件 ID 来管理插件",
column_name,
data_list,
text_style=row_style,
),
await ImageTemplate.table_page(
"第三方插件列表",
"通过添加/移除插件 ID 来管理插件",
column_name,
extra_data_list,
text_style=row_style,
),
]
native_table_builder = TableBuilder(
title="原生插件列表", tip="通过添加/移除插件 ID 来管理插件"
).set_headers(column_name)
native_rows_data = []
for row_data in structured_native_list:
row_color = HIGHLIGHT_COLOR if row_data["has_update"] else None
status_cell = (
StatusBadgeCell(text="已安装", status_type="ok")
if row_data["is_installed"]
else TextCell(content="")
)
native_rows_data.append(
[
status_cell,
TextCell(content=str(row_data["id"]), color=row_color),
TextCell(content=row_data["name"], color=row_color),
TextCell(content=row_data["description"], color=row_color),
TextCell(content=row_data["author"], color=row_color),
TextCell(
content=row_data["version_str"],
color=row_color,
bold=bool(row_color),
),
TextCell(content=row_data["type_name"], color=row_color),
]
)
native_table_builder.add_rows(native_rows_data)
native_table_bytes = await ui.render(
native_table_builder.build(),
viewport={"width": 1400, "height": 10},
device_scale_factor=2,
)
extra_table_builder = TableBuilder(
title="第三方插件列表", tip="通过添加/移除插件 ID 来管理插件"
).set_headers(column_name)
extra_rows_data = []
for row_data in structured_extra_list:
row_color = HIGHLIGHT_COLOR if row_data["has_update"] else None
status_cell = (
StatusBadgeCell(text="已安装", status_type="ok")
if row_data["is_installed"]
else TextCell(content="")
)
extra_rows_data.append(
[
status_cell,
TextCell(content=str(row_data["id"]), color=row_color),
TextCell(content=row_data["name"], color=row_color),
TextCell(content=row_data["description"], color=row_color),
TextCell(content=row_data["author"], color=row_color),
TextCell(
content=row_data["version_str"],
color=row_color,
bold=bool(row_color),
),
TextCell(content=row_data["type_name"], color=row_color),
]
)
extra_table_builder.add_rows(extra_rows_data)
extra_table_bytes = await ui.render(
extra_table_builder.build(),
viewport={"width": 1400, "height": 10},
device_scale_factor=2,
)
return [native_table_bytes, extra_table_bytes]
@classmethod
async def get_plugin_by_value(
@ -290,7 +338,6 @@ class StoreManager:
await VirtualEnvPackageManager.install_requirement(requirement_file)
if not is_install_req:
# 从仓库根目录查找文件
rand = random.randint(1, 10000)
requirement_path = TEMP_PATH / f"plugin_store_{rand}_req.txt"
requirements_path = TEMP_PATH / f"plugin_store_{rand}_reqs.txt"
@ -345,19 +392,20 @@ class StoreManager:
return f"插件 {plugin_info.name} 移除成功! 重启后生效"
@classmethod
async def search_plugin(cls, plugin_name_or_author: str) -> BuildImage | str:
async def search_plugin(cls, plugin_name_or_author: str) -> bytes | str:
"""搜索插件
参数:
plugin_name_or_author: 插件名称或作者
返回:
BuildImage | str: 返回消息
bytes | str: 返回消息
"""
plugin_list, extra_plugin_list = await cls.get_data()
all_plugin_list = plugin_list + extra_plugin_list
db_plugin_list = await cls.get_loaded_plugins("module", "version")
suc_plugin = {p[0]: (p[1] or "Unknown") for p in db_plugin_list}
filtered_data = [
(id, plugin_info)
for id, plugin_info in enumerate(all_plugin_list)
@ -365,28 +413,50 @@ class StoreManager:
or plugin_name_or_author.lower() in plugin_info.author.lower()
]
data_list = [
[
"已安装" if plugin_info.module in suc_plugin else "",
id,
plugin_info.name,
plugin_info.description,
plugin_info.author,
cls.version_check(plugin_info, suc_plugin),
plugin_info.plugin_type_name,
]
for id, plugin_info in filtered_data
]
if not data_list:
if not filtered_data:
return "未找到相关插件..."
HIGHLIGHT_COLOR = "#E6A23C"
column_name = ["-", "ID", "名称", "简介", "作者", "版本", "类型"]
return await ImageTemplate.table_page(
"商店插件列表",
"通过添加/移除插件 ID 来管理插件",
column_name,
data_list,
text_style=row_style,
builder = TableBuilder(
title=f"插件搜索结果: '{plugin_name_or_author}'",
tip="通过添加/移除插件 ID 来管理插件",
)
builder.set_headers(column_name)
rows_to_add = []
for id, plugin_info in filtered_data:
is_new = cls.check_version_is_new(plugin_info, suc_plugin)
has_update = not is_new and plugin_info.module in suc_plugin
row_color = HIGHLIGHT_COLOR if has_update else None
status_cell = (
StatusBadgeCell(text="已安装", status_type="ok")
if plugin_info.module in suc_plugin
else TextCell(content="")
)
rows_to_add.append(
[
status_cell,
TextCell(content=str(id), color=row_color),
TextCell(content=plugin_info.name, color=row_color),
TextCell(content=plugin_info.description, color=row_color),
TextCell(content=plugin_info.author, color=row_color),
TextCell(
content=cls.version_check(plugin_info, suc_plugin),
color=row_color,
bold=has_update,
),
TextCell(content=plugin_info.plugin_type_name, color=row_color),
]
)
builder.add_rows(rows_to_add)
render_viewport = {"width": 1400, "height": 10}
return await ui.render(builder.build(), viewport=render_viewport)
@classmethod
async def update_plugin(cls, index_or_module: str) -> str:

View File

@ -13,6 +13,7 @@ from nonebot_plugin_uninfo import Uninfo
from pydantic import BaseModel, Field, create_model
from tortoise.expressions import Q
from zhenxun import ui
from zhenxun.configs.config import BotConfig
from zhenxun.models.friend_user import FriendUser
from zhenxun.models.goods_info import GoodsInfo
@ -20,7 +21,6 @@ from zhenxun.models.group_member_info import GroupInfoUser
from zhenxun.models.user_console import UserConsole
from zhenxun.models.user_gold_log import UserGoldLog
from zhenxun.models.user_props_log import UserPropsLog
from zhenxun.services import renderer_service
from zhenxun.services.log import logger
from zhenxun.utils.enum import GoldHandle, PropHandle
from zhenxun.utils.image_utils import BuildImage, ImageTemplate
@ -622,4 +622,4 @@ async def prepare_shop_data() -> bytes:
"categories": categories,
}
return await renderer_service.render("pages/builtin/shop", data=shop_data)
return await ui.render_template("pages/builtin/shop", data=shop_data)

View File

@ -9,10 +9,10 @@ from nonebot.drivers import Driver
from nonebot_plugin_uninfo import Uninfo
import pytz
from zhenxun import ui
from zhenxun.configs.config import BotConfig, Config
from zhenxun.models.sign_log import SignLog
from zhenxun.models.sign_user import SignUser
from zhenxun.services import renderer_service
from zhenxun.utils.manager.priority_manager import PriorityLifecycle
from zhenxun.utils.platform import PlatformUtils
@ -281,7 +281,7 @@ async def _generate_html_card(
"last_sign_date_str": last_sign_date_str,
}
image_bytes = await renderer_service.render("pages/builtin/sign", data=card_data)
image_bytes = await ui.render_template("pages/builtin/sign", data=card_data)
async with aiofiles.open(card_file, "wb") as f:
await f.write(image_bytes)

View File

@ -7,7 +7,6 @@ from zhenxun.models.statistics import Statistics
from zhenxun.utils.echart_utils import ChartUtils
from zhenxun.utils.echart_utils.models import Barh
from zhenxun.utils.enum import PluginType
from zhenxun.utils.image_utils import BuildImage
from zhenxun.utils.time_utils import TimeUtils
@ -60,7 +59,7 @@ class StatisticsManage:
@classmethod
async def get_global_statistics(
cls, plugin_name: str | None, day: int | None, title: str
) -> BuildImage | str:
) -> bytes | str:
query = Statistics
if plugin_name:
query = query.filter(plugin_name=plugin_name)
@ -114,7 +113,7 @@ class StatisticsManage:
)
@classmethod
async def __build_image(cls, data_list: list[tuple[str, int]], title: str):
async def __build_image(cls, data_list: list[tuple[str, int]], title: str) -> bytes:
module2count = {x[0]: x[1] for x in data_list}
plugin_info = await PluginInfo.filter(
module__in=module2count.keys(),

View File

@ -8,7 +8,6 @@ from nonebot_plugin_session import EventSession
from zhenxun.configs.config import Config
from zhenxun.configs.utils import PluginExtraData, RegisterConfig
from zhenxun.services.log import logger
from zhenxun.services.renderer import renderer_service
from zhenxun.utils.enum import PluginType
from zhenxun.utils.message import MessageUtils
@ -56,9 +55,6 @@ _matcher = on_alconna(
async def _(session: EventSession, arparma: Arparma):
Config.reload()
logger.debug("自动重载配置文件", arparma.header_result, session=session)
await renderer_service.reload_theme()
await MessageUtils.build_message("重载完成!").send(reply_to=True)

View File

@ -0,0 +1,62 @@
from nonebot.permission import SUPERUSER
from nonebot.plugin import PluginMetadata
from nonebot.rule import to_me
from nonebot_plugin_alconna import Alconna, Arparma, on_alconna
from zhenxun.configs.utils import PluginExtraData, RegisterConfig
from zhenxun.services import renderer_service
from zhenxun.services.log import logger
from zhenxun.utils.enum import PluginType
from zhenxun.utils.message import MessageUtils
__plugin_meta__ = PluginMetadata(
name="UI管理",
description="管理UI、主题和渲染服务的相关配置",
usage="""
指令
重载UI主题
""".strip(),
extra=PluginExtraData(
author="HibiKier",
version="0.1",
plugin_type=PluginType.SUPERUSER,
configs=[
RegisterConfig(
module="UI",
key="THEME",
value="default",
help="设置渲染服务使用的全局主题名称(对应 resources/themes/下的目录名)",
default_value="default",
type=str,
),
RegisterConfig(
module="UI",
key="CACHE",
value=True,
help="是否为渲染服务生成的图片启用文件缓存",
default_value=True,
type=bool,
),
],
).to_dict(),
)
_matcher = on_alconna(
Alconna("重载主题"),
rule=to_me(),
permission=SUPERUSER,
priority=1,
block=True,
)
@_matcher.handle()
async def _(arparma: Arparma):
theme_name = await renderer_service.reload_theme()
logger.info(
f"UI主题已重载为: {theme_name}", "UI管理器", session=arparma.header_result
)
await MessageUtils.build_message(f"UI主题已成功重载为 '{theme_name}'").send(
reply_to=True
)

View File

@ -4,10 +4,12 @@ import nonebot
from nonebot.plugin import PluginMetadata
from pydantic import BaseModel
from zhenxun import ui
from zhenxun.configs.config import BotConfig
from zhenxun.models.plugin_info import PluginInfo
from zhenxun.models.task_info import TaskInfo
from zhenxun.ui import HelpCategory, HelpItem, PluginHelpPageBuilder
from zhenxun.ui.builders import PluginHelpPageBuilder
from zhenxun.ui.models import HelpCategory, HelpItem
from zhenxun.utils.common_utils import format_usage_for_markdown
from zhenxun.utils.enum import PluginType
@ -104,6 +106,6 @@ async def create_plugin_help_image(
)
)
image_bytes = await builder.build()
image_bytes = await ui.render(builder.build(), use_cache=True)
return image_bytes

View File

@ -1,37 +1,13 @@
"""
图片渲染服务
提供一个统一的可扩展的接口来将结构化数据渲染成图片
"""
from zhenxun.configs.config import Config
from zhenxun.utils.manager.priority_manager import PriorityLifecycle
from .service import RendererService
Config.add_plugin_config(
"UI",
"THEME",
"default",
help="设置渲染服务使用的全局主题名称 (对应 resources/themes/下的目录名)",
default_value="default",
type=str,
)
Config.add_plugin_config(
"UI",
"CACHE",
True,
help="是否为渲染服务生成的图片启用文件缓存",
default_value=True,
type=bool,
)
renderer_service = RendererService()
@PriorityLifecycle.on_startup(priority=10)
async def _init_renderer_service():
"""在Bot启动时预热渲染服务,扫描并加载所有模板"""
"""在Bot启动时初始化渲染服务及其依赖。"""
await renderer_service.initialize()

View File

@ -0,0 +1,34 @@
from pathlib import Path
from nonebot_plugin_htmlrender import html_to_pic
from .protocols import ScreenshotEngine
class PlaywrightEngine(ScreenshotEngine):
"""使用 nonebot-plugin-htmlrender 实现的截图引擎。"""
async def render(self, html: str, base_url_path: Path, **render_options) -> bytes:
base_url_for_browser = base_url_path.absolute().as_uri()
if not base_url_for_browser.endswith("/"):
base_url_for_browser += "/"
final_render_options = {
"viewport": {"width": 800, "height": 10},
**render_options,
"base_url": base_url_for_browser,
}
return await html_to_pic(
html=html,
template_path=base_url_for_browser,
**final_render_options,
)
def get_screenshot_engine() -> ScreenshotEngine:
"""
截图引擎工厂函数
目前只返回 PlaywrightEngine, 未来可以根据配置返回不同的引擎
"""
return PlaywrightEngine()

View File

@ -1,221 +0,0 @@
from abc import ABC, abstractmethod
from pathlib import Path
import aiofiles
from jinja2 import Environment
import markdown
from nonebot_plugin_htmlrender import html_to_pic
from pydantic import BaseModel
from zhenxun.configs.path_config import THEMES_PATH
from zhenxun.services.log import logger
from .models import Theme
THEME_PATH = THEMES_PATH
RESOURCE_ROOT = THEMES_PATH.parent
class BaseEngine(ABC):
"""渲染引擎的抽象基类。"""
@abstractmethod
async def render(
self,
template_name: str,
data: BaseModel | dict | None,
theme: Theme,
jinja_env: "Environment | None" = None,
extra_css_paths: list[Path] | None = None,
custom_css_path: Path | None = None,
**kwargs,
) -> bytes:
"""所有引擎都必须实现的渲染方法。"""
pass
class BaseHtmlRenderingEngine(BaseEngine):
"""
一个专门用于处理HTML到图片转换的引擎基类
它使用模板方法模式定义了渲染的固定流程
并将具体的HTML内容生成委托给子类的抽象方法 `get_html_content`
"""
@abstractmethod
async def get_html_content(
self,
template_name: str,
data: BaseModel | dict | None,
theme: Theme,
jinja_env: "Environment",
extra_css_paths: list[Path] | None,
custom_css_path: Path | None,
frameless: bool,
**kwargs,
) -> str:
"""
[抽象方法] 子类必须实现此方法以生成最终的HTML字符串
"""
pass
async def render(
self,
template_name: str,
data: BaseModel | dict | None,
theme: Theme,
jinja_env: "Environment | None" = None,
extra_css_paths: list[Path] | None = None,
custom_css_path: Path | None = None,
**kwargs,
) -> bytes:
"""
[通用渲染流程] 调用 `get_html_content` 获取HTML然后调用 `html_to_pic` 生成图片
"""
if not jinja_env:
raise ValueError("HTML渲染器需要一个有效的Jinja2环境实例。")
frameless = kwargs.pop("frameless", False)
html_content = await self.get_html_content(
template_name,
data,
theme,
jinja_env,
extra_css_paths,
custom_css_path,
frameless=frameless,
**kwargs,
)
base_url_for_browser = RESOURCE_ROOT.absolute().as_uri()
if not base_url_for_browser.endswith("/"):
base_url_for_browser += "/"
pages_config = {
"viewport": kwargs.pop("viewport", {"width": 800, "height": 10}),
"base_url": base_url_for_browser,
}
final_screenshot_kwargs = kwargs.copy()
final_screenshot_kwargs.update(pages_config)
return await html_to_pic(
html=html_content,
template_path=base_url_for_browser,
**final_screenshot_kwargs,
)
class HtmlRenderer(BaseHtmlRenderingEngine):
"""使用 nonebot-plugin-htmlrender 渲染HTML模板的引擎。"""
async def get_html_content(
self,
template_name: str,
data: BaseModel | dict | None,
theme: Theme,
jinja_env: "Environment",
extra_css_paths: list[Path] | None,
custom_css_path: Path | None,
frameless: bool,
**kwargs,
) -> str:
def asset_loader(asset_path: str) -> str:
current_theme_asset = theme.assets_dir / asset_path
if current_theme_asset.exists():
return current_theme_asset.relative_to(RESOURCE_ROOT).as_posix()
default_theme_asset = theme.default_assets_dir / asset_path
if default_theme_asset.exists():
return default_theme_asset.relative_to(RESOURCE_ROOT).as_posix()
logger.warning(
f"资源文件在主题 '{theme.name}''default' 中均未找到: {asset_path}"
)
return ""
extra_css_content = ""
if extra_css_paths:
css_contents = []
for path in extra_css_paths:
if path.exists():
async with aiofiles.open(path, encoding="utf-8") as f:
css_contents.append(await f.read())
extra_css_content = "\n".join(css_contents)
template_context = {
"data": data,
"extra_css": extra_css_content,
"frameless": frameless,
"theme": {
"name": theme.name,
"palette": theme.palette,
"asset": asset_loader,
},
}
template = jinja_env.get_template(template_name)
return await template.render_async(**template_context)
class MarkdownEngine(BaseHtmlRenderingEngine):
"""在服务端渲染 Markdown 为 HTML然后截图的引擎。"""
async def get_html_content(
self,
template_name: str,
data: BaseModel | dict | None,
theme: Theme,
jinja_env: "Environment",
extra_css_paths: list[Path] | None,
custom_css_path: Path | None,
frameless: bool,
**kwargs,
) -> str:
if isinstance(data, BaseModel):
raw_md = getattr(data, "markdown", "") if hasattr(data, "markdown") else ""
else:
raw_md = (data or {}).get("markdown", "")
md_html = markdown.markdown(
raw_md,
extensions=[
"pymdownx.tasklist",
"tables",
"fenced_code",
"codehilite",
"mdx_math",
"pymdownx.tilde",
],
extension_configs={"mdx_math": {"enable_dollar_delimiter": True}},
)
final_css_content = ""
if custom_css_path and custom_css_path.exists():
logger.debug(f"正在为 Markdown 渲染加载自定义样式: {custom_css_path}")
async with aiofiles.open(custom_css_path, encoding="utf-8") as f:
final_css_content = await f.read()
else:
css_paths = [
theme.default_assets_dir / "css/markdown/github-light.css",
theme.default_assets_dir / "css/markdown/pygments-default.css",
]
css_contents = []
for path in css_paths:
if path.exists():
async with aiofiles.open(path, encoding="utf-8") as f:
css_contents.append(await f.read())
final_css_content = "\n".join(css_contents)
template_context = {
"data": data,
"theme_css": theme.style_css,
"custom_style_css": final_css_content,
"md_html": md_html,
"extra_css": "",
"frameless": frameless,
"theme": {"name": theme.name},
}
template = jinja_env.get_template(template_name)
return await template.render_async(**template_context)

View File

@ -11,7 +11,8 @@ class Theme(BaseModel):
name: str = Field(..., description="主题名称")
palette: dict[str, Any] = Field(
default_factory=dict, description="用于PIL渲染的调色板"
default_factory=dict,
description="主题的调色板用于定义CSS变量和Jinja2模板中的颜色常量",
)
style_css: str = Field("", description="用于HTML渲染的全局CSS内容")
assets_dir: Path = Field(..., description="主题的资产目录路径")
@ -32,9 +33,6 @@ class TemplateManifest(BaseModel):
entrypoint: str = Field(
..., description="模板的入口文件 (例如 'template.html''renderer.py')"
)
schema_path: str | None = Field(
None, description="用于数据验证的Pydantic模型的Python导入路径"
)
render_options: dict[str, Any] = Field(
default_factory=dict, description="传递给渲染引擎的额外选项 (如viewport)"
)

View File

@ -0,0 +1,73 @@
from abc import ABC, abstractmethod
from collections.abc import Awaitable
from pathlib import Path
from typing import Any, Protocol
from pydantic import BaseModel
class Renderable(ABC):
"""
一个协议定义了任何可被渲染的UI组件必须具备的形态
"""
@property
@abstractmethod
def template_name(self) -> str:
"""组件声明它需要哪个模板文件。"""
...
async def prepare(self) -> None:
"""
[可选] 一个生命周期钩子用于在渲染前执行异步数据获取和预处理
"""
pass
def get_required_scripts(self) -> list[str]:
"""[可选] 返回此组件所需的JS脚本路径列表 (相对于assets目录)。"""
return []
def get_required_styles(self) -> list[str]:
"""[可选] 返回此组件所需的CSS样式表路径列表 (相对于assets目录)。"""
return []
@abstractmethod
def get_render_data(self) -> dict[str, Any | Awaitable[Any]]:
"""
返回一个将传递给模板的数据字典
重要字典的值可以是协程(Awaitable)渲染服务会自动解析它们
"""
...
def get_extra_css(self, theme_manager: Any) -> str | Awaitable[str]:
"""
[可选] 一个生命周期钩子让组件可以提供额外的CSS
可以返回 str awaitable[str]
"""
return ""
class ScreenshotEngine(Protocol):
"""
一个协议定义了截图引擎的核心能力
"""
async def render(self, html: str, base_url_path: Path, **render_options) -> bytes:
"""
将HTML字符串截图为图片
参数:
html: 要渲染的HTML内容
base_url_path: 用于解析相对路径如CSS, JS, 图片的基础URL路径
**render_options: 传递给底层截图库的额外选项 ( viewport)
"""
...
class RenderResult(BaseModel):
"""
渲染服务的统一返回类型
"""
image_bytes: bytes | None = None
html_content: str | None = None

View File

@ -1,69 +1,58 @@
import asyncio
from collections.abc import Callable, Generator
from collections.abc import Callable
import hashlib
import json
from pathlib import Path
from typing import ClassVar, Literal
import aiofiles
from jinja2 import ChoiceLoader, Environment, FileSystemLoader, PrefixLoader
import markdown
from pydantic import BaseModel, ValidationError
from jinja2 import (
Environment,
FileSystemLoader,
select_autoescape,
)
from nonebot.utils import is_coroutine_callable
import ujson as json
from zhenxun.configs.config import Config
from zhenxun.configs.path_config import THEMES_PATH, UI_CACHE_PATH
from zhenxun.services.log import logger
from zhenxun.utils.exception import RenderingError
from .engines import BaseEngine, HtmlRenderer, MarkdownEngine
from .models import TemplateManifest, Theme
THEME_PATH = THEMES_PATH
from .engine import get_screenshot_engine
from .protocols import Renderable, RenderResult, ScreenshotEngine
from .theme import ThemeManager
class RendererService:
"""图片渲染服务管理器。"""
"""
图片渲染服务的统一门面
负责编排和调用底层渲染服务提供统一的渲染接口
支持多种渲染方式组件渲染模板渲染等
"""
_plugin_template_paths: ClassVar[dict[str, Path]] = {}
def __init__(self):
self._engines: dict[str, BaseEngine] = {
"html": HtmlRenderer(),
"markdown": MarkdownEngine(),
}
self._templates: dict[str, TemplateManifest] = {}
self._template_paths: dict[str, Path] = {}
self._plugin_template_paths: dict[str, Path] = {}
self._plugin_manifests: dict[str, TemplateManifest] = {}
self._init_lock = asyncio.Lock()
self._theme_manager: ThemeManager | None = None
self._screenshot_engine: ScreenshotEngine | None = None
self._initialized = False
self._current_theme_data: Theme | None = None
self._jinja_environments: dict[str, Environment] = {}
self._init_lock = asyncio.Lock()
self._custom_filters: dict[str, Callable] = {}
self._custom_globals: dict[str, Callable] = {}
self._markdown_styles: dict[str, Path] = {}
def register_template_namespace(self, namespace: str, path: Path):
"""
为插件注册一个模板命名空间
参数:
namespace: 插件的唯一命名空间 (建议使用插件模块名)
path: 包含模板文件的目录路径
"""
"""[新增] 插件注册模板路径的入口点"""
if namespace in self._plugin_template_paths:
logger.warning(f"模板命名空间 '{namespace}' 已被注册,将被覆盖。")
if not path.is_dir():
raise ValueError(f"提供的路径 '{path}' 不是一个有效的目录。")
self._plugin_template_paths[namespace] = path
logger.debug(f"已注册模板命名空间 '{namespace}' -> '{path}'")
def register_markdown_style(self, name: str, path: Path):
"""
[新增] Markdown 渲染器注册一个具名样式
参数:
name: 样式的唯一名称 (建议使用 '插件名:样式名' 格式以避免冲突)
path: 指向 CSS 文件的 Path 对象
Markdown 渲染器注册一个具名样式
"""
if name in self._markdown_styles:
logger.warning(f"Markdown 样式 '{name}' 已被注册,将被覆盖。")
@ -75,10 +64,6 @@ class RendererService:
def filter(self, name: str) -> Callable:
"""
装饰器注册一个自定义 Jinja2 过滤器
参数:
name: 过滤器在模板中的调用名称为避免冲突强烈建议使用
'插件名_过滤器名' 的格式
"""
def decorator(func: Callable) -> Callable:
@ -93,10 +78,6 @@ class RendererService:
def global_function(self, name: str) -> Callable:
"""
装饰器注册一个自定义 Jinja2 全局函数
参数:
name: 函数在模板中的调用名称为避免冲突强烈建议使用
'插件名_函数名' 的格式
"""
def decorator(func: Callable) -> Callable:
@ -108,382 +89,235 @@ class RendererService:
return decorator
async def _load_theme(self, theme_name: str):
"""加载指定主题的配置和样式。"""
theme_dir = THEME_PATH / theme_name
if not theme_dir.is_dir():
logger.error(f"主题 '{theme_name}' 不存在,将回退到默认主题。")
if theme_name == "default":
return
theme_name = "default"
theme_dir = THEME_PATH / "default"
palette_path = theme_dir / "palette.json"
default_palette_path = THEMES_PATH / "default" / "palette.json"
palette = {}
if palette_path.exists():
try:
palette = json.loads(palette_path.read_text(encoding="utf-8"))
except json.JSONDecodeError:
logger.warning(f"主题 '{theme_name}' 的 palette.json 文件解析失败。")
if not palette and default_palette_path.exists():
logger.debug(
f"主题 '{theme_name}' 未提供有效的 palette.json"
"回退到默认主题的调色板。"
)
try:
palette = json.loads(default_palette_path.read_text(encoding="utf-8"))
except json.JSONDecodeError:
logger.error("默认主题的 palette.json 文件解析失败,调色板将为空。")
palette = {}
elif not palette:
logger.error("当前主题和默认主题均未找到有效的 palette.json。")
self._current_theme_data = Theme(
name=theme_name,
palette=palette,
style_css="",
assets_dir=theme_dir / "assets",
default_assets_dir=THEMES_PATH / "default" / "assets",
)
self._jinja_environments.clear()
logger.info(f"渲染服务已加载主题: {theme_name}")
async def reload_theme(self) -> str:
"""
重新加载当前主题的配置和样式并清除缓存的Jinja环境
"""
async with self._init_lock:
current_theme_name = Config.get_config("UI", "THEME", "default")
await self._load_theme(current_theme_name)
logger.info(f"主题 '{current_theme_name}' 已成功重载。")
return current_theme_name
def _get_or_create_jinja_env(self, theme: Theme) -> Environment:
"""为指定主题获取或创建一个缓存的 Jinja2 环境。"""
if theme.name in self._jinja_environments:
return self._jinja_environments[theme.name]
logger.debug(f"为主题 '{theme.name}' 创建新的 Jinja2 环境...")
prefix_loader = PrefixLoader(
{
namespace: FileSystemLoader(str(path.absolute()))
for namespace, path in self._plugin_template_paths.items()
}
)
current_theme_templates_dir = THEMES_PATH / theme.name / "templates"
default_theme_templates_dir = THEMES_PATH / "default" / "templates"
theme_loader = FileSystemLoader(
[
str(current_theme_templates_dir.absolute()),
str(default_theme_templates_dir.absolute()),
]
)
final_loader = ChoiceLoader([prefix_loader, theme_loader])
env = Environment(
loader=final_loader,
enable_async=True,
autoescape=True,
)
def markdown_filter(text: str) -> str:
"""一个将 Markdown 文本转换为 HTML 的 Jinja2 过滤器。"""
if not isinstance(text, str):
return ""
return markdown.markdown(
text,
extensions=[
"pymdownx.tasklist",
"tables",
"fenced_code",
"codehilite",
"mdx_math",
"pymdownx.tilde",
],
extension_configs={"mdx_math": {"enable_dollar_delimiter": True}},
)
env.filters["md"] = markdown_filter
if self._custom_filters:
env.filters.update(self._custom_filters)
logger.debug(
f"向 Jinja2 环境注入了 {len(self._custom_filters)} 个自定义过滤器。"
)
if self._custom_globals:
env.globals.update(self._custom_globals)
logger.debug(
f"向 Jinja2 环境注入了 {len(self._custom_globals)} 个自定义全局函数。"
)
self._jinja_environments[theme.name] = env
return env
async def initialize(self):
"""扫描并加载所有模板清单。"""
"""[新增] 延迟初始化方法,在 on_startup 钩子中调用"""
if self._initialized:
return
async with self._init_lock:
if self._initialized:
return
logger.info("开始扫描渲染模板...")
base_template_path = THEMES_PATH / "default" / "templates"
base_template_path.mkdir(exist_ok=True, parents=True)
for manifest_path in base_template_path.glob("**/manifest.json"):
template_dir = manifest_path.parent
try:
manifest = TemplateManifest.parse_file(manifest_path)
template_name = template_dir.relative_to(
base_template_path
).as_posix()
self._templates[template_name] = manifest
self._template_paths[template_name] = template_dir
logger.debug(
f"发现并加载基础模板 '{template_name}' "
f"(引擎: {manifest.engine})"
)
except ValidationError as e:
logger.error(f"解析模板清单 '{manifest_path}' 失败: {e}")
for namespace, plugin_template_path in self._plugin_template_paths.items():
for manifest_path in plugin_template_path.glob("**/manifest.json"):
template_dir = manifest_path.parent
try:
manifest = TemplateManifest.parse_file(manifest_path)
relative_path = template_dir.relative_to(
plugin_template_path
).as_posix()
template_name_with_ns = f"{namespace}:{relative_path}"
self._plugin_manifests[template_name_with_ns] = manifest
logger.debug(
f"发现并加载插件模板 '{template_name_with_ns}' "
f"(引擎: {manifest.engine})"
)
except ValidationError as e:
logger.error(f"解析插件模板清单 '{manifest_path}' 失败: {e}")
self._screenshot_engine = get_screenshot_engine()
self._theme_manager = ThemeManager(
self._plugin_template_paths,
self._custom_filters,
self._custom_globals,
self._markdown_styles,
)
current_theme_name = Config.get_config("UI", "THEME", "default")
await self._load_theme(current_theme_name)
await self._theme_manager.load_theme(current_theme_name)
self._initialized = True
logger.info(
f"渲染模板扫描完成,共加载 {len(self._templates)} 个基础模板和 "
f"{len(self._plugin_manifests)} 个插件模板。"
)
def _yield_theme_paths(self, relative_path: Path) -> Generator[Path, None, None]:
async def _render_component(
self, component: Renderable, use_cache: bool = False, **render_options
) -> RenderResult:
"""
按优先级生成一个资源的完整路径当前主题 -> 默认主题
核心的私有渲染方法执行完整的渲染流程
"""
if not self._current_theme_data:
return
cache_path = None
if Config.get_config("UI", "CACHE") and use_cache:
try:
template_name = component.template_name
data_dict = component.get_render_data()
current_theme_path = THEMES_PATH / self._current_theme_data.name / relative_path
yield current_theme_path
resolved_data_dict = {}
for key, value in data_dict.items():
if is_coroutine_callable(value): # type: ignore
resolved_data_dict[key] = await value
else:
resolved_data_dict[key] = value
if self._current_theme_data.name != "default":
default_theme_path = THEMES_PATH / "default" / relative_path
yield default_theme_path
data_str = json.dumps(resolved_data_dict, sort_keys=True)
def _resolve_markdown_style_path(self, style_name: str) -> Path | None:
"""
按照 注册 -> 主题约定 -> 默认约定 的顺序解析 Markdown 样式路径
"""
if style_name in self._markdown_styles:
logger.debug(f"找到已注册的 Markdown 样式: '{style_name}'")
return self._markdown_styles[style_name]
cache_key_str = f"{template_name}:{data_str}"
cache_filename = (
f"{hashlib.sha256(cache_key_str.encode()).hexdigest()}.png"
)
cache_path = UI_CACHE_PATH / cache_filename
conventional_relative_paths = [
Path("templates")
/ "components"
/ "cards"
/ "markdown_image"
/ "styles"
/ f"{style_name}.css",
Path("assets") / "css" / "markdown" / f"{style_name}.css",
]
for relative_path in conventional_relative_paths:
for potential_path in self._yield_theme_paths(relative_path):
if potential_path.exists():
logger.debug(f"在约定路径找到 Markdown 样式: {potential_path}")
return potential_path
logger.warning(f"样式 '{style_name}' 在注册表和约定路径中均未找到。")
return None
def _resolve_style_path(self, template_name: str, style_name: str) -> Path | None:
"""
[重构后] 实现 当前主题 -> 默认主题 的回退查找逻辑
"""
relative_style_path = (
Path("templates") / template_name / "styles" / f"{style_name}.css"
)
for potential_path in self._yield_theme_paths(relative_style_path):
if potential_path.exists():
logger.debug(f"找到样式 '{style_name}': {potential_path}")
return potential_path
logger.warning(f"样式 '{style_name}' 在当前主题和默认主题中均未找到。")
return None
async def render(
self,
template_name: str,
data: dict | BaseModel | None = None,
use_cache: bool = False,
style_name: str | None = None,
**render_options_override,
) -> bytes:
"""
渲染指定的模板并支持透明缓存
"""
await self.initialize()
if cache_path.exists():
logger.debug(f"UI缓存命中: {cache_path}")
async with aiofiles.open(cache_path, "rb") as f:
image_bytes = await f.read()
return RenderResult(
image_bytes=image_bytes, html_content="<!-- from cache -->"
)
logger.debug(f"UI缓存未命中: {cache_key_str[:100]}...")
except Exception as e:
logger.warning(f"UI缓存读取失败: {e}", e=e)
cache_path = None
try:
extra_css_paths = []
custom_markdown_css_path = None
manifest: TemplateManifest | None = self._templates.get(
template_name
) or self._plugin_manifests.get(template_name)
if not self._initialized:
await self.initialize()
assert self._theme_manager is not None, "ThemeManager 未初始化"
assert self._screenshot_engine is not None, "ScreenshotEngine 未初始化"
if style_name:
if manifest and manifest.engine == "markdown":
custom_markdown_css_path = self._resolve_markdown_style_path(
style_name
)
else:
resolved_path = self._resolve_style_path(template_name, style_name)
if resolved_path:
extra_css_paths.append(resolved_path)
if hasattr(component, "prepare"):
await component.prepare()
cache_path = None
if Config.get_config("UI", "CACHE") and use_cache:
try:
if isinstance(data, BaseModel):
data_str = f"{data.__class__.__name__}:{data!s}"
else:
data_str = json.dumps(data or {}, sort_keys=True)
cache_key_str = f"{template_name}:{data_str}"
cache_filename = (
f"{hashlib.sha256(cache_key_str.encode()).hexdigest()}.png"
)
cache_path = UI_CACHE_PATH / cache_filename
required_scripts = set(component.get_required_scripts())
required_styles = set(component.get_required_styles())
if cache_path.exists():
logger.debug(f"UI缓存命中: {cache_path}")
async with aiofiles.open(cache_path, "rb") as f:
return await f.read()
logger.debug(f"UI缓存未命中: {cache_key_str[:100]}...")
except Exception as e:
logger.warning(f"UI缓存读取失败: {e}", e=e)
cache_path = None
if hasattr(component, "required_scripts"):
required_scripts.update(getattr(component, "required_scripts"))
if hasattr(component, "required_styles"):
required_styles.update(getattr(component, "required_styles"))
if not self._current_theme_data:
raise RuntimeError("主题未被正确加载,无法进行渲染。")
data_dict = component.get_render_data()
manifest: TemplateManifest | None = None
final_template_dir: Path | None = None
relative_template_name: str = ""
is_plugin_template = ":" in template_name
component_render_options = data_dict.get("render_options", {})
if not isinstance(component_render_options, dict):
component_render_options = {}
if is_plugin_template:
namespace, path_part = template_name.split(":", 1)
manifest = self._plugin_manifests.get(template_name)
if namespace in self._plugin_template_paths:
plugin_base_path = self._plugin_template_paths[namespace]
final_template_dir = plugin_base_path / Path(path_part).parent
manifest_options = {}
if manifest := await self._theme_manager.get_template_manifest(
component.template_name
):
manifest_options = manifest.render_options or {}
relative_template_name = template_name
if manifest:
logger.debug(f"使用插件模板: '{template_name}'")
if (
getattr(component, "_is_standalone_template", False)
and hasattr(component, "template_path")
and isinstance(
template_path := getattr(component, "template_path"), Path
)
and template_path.is_absolute()
):
logger.debug(f"正在渲染独立模板: '{template_path}'", "RendererService")
template_dir = template_path.parent
temp_loader = FileSystemLoader(str(template_dir))
temp_env = Environment(
loader=temp_loader,
enable_async=True,
autoescape=select_autoescape(["html", "xml"]),
)
temp_env.globals["theme"] = self._theme_manager.jinja_env.globals.get(
"theme", {}
)
temp_env.filters["md"] = self._theme_manager._markdown_filter
template = temp_env.get_template(template_path.name)
html_content = await template.render_async(data=data_dict)
final_render_options = component_render_options.copy()
final_render_options.update(render_options)
image_bytes = await self._screenshot_engine.render(
html=html_content,
base_url_path=template_dir,
**final_render_options,
)
if Config.get_config("UI", "CACHE") and use_cache and cache_path:
try:
async with aiofiles.open(cache_path, "wb") as f:
await f.write(image_bytes)
logger.debug(f"UI缓存写入成功: {cache_path}")
except Exception as e:
logger.warning(f"UI缓存写入失败: {e}", e=e)
return RenderResult(image_bytes=image_bytes, html_content=html_content)
else:
theme_template_dir = (
THEMES_PATH
/ self._current_theme_data.name
/ "templates"
/ template_name
)
default_template_dir = (
THEMES_PATH / "default" / "templates" / template_name
final_render_options = component_render_options.copy()
final_render_options.update(manifest_options)
final_render_options.update(render_options)
if not self._theme_manager.current_theme:
raise RenderingError("渲染失败:主题未被正确加载。")
html_content = await self._theme_manager._render_component_to_html(
component,
required_scripts=list(required_scripts),
required_styles=list(required_styles),
**final_render_options,
)
if (
theme_template_dir.is_dir()
and (theme_template_dir / "manifest.json").is_file()
):
final_template_dir = theme_template_dir
logger.debug(
f"使用主题 '{self._current_theme_data.name}' "
f"覆盖的模板: '{template_name}'"
)
elif (
default_template_dir.is_dir()
and (default_template_dir / "manifest.json").is_file()
):
final_template_dir = default_template_dir
logger.debug(f"使用基础(default)模板: '{template_name}'")
screenshot_options = final_render_options.copy()
screenshot_options.pop("extra_css", None)
screenshot_options.pop("frameless", None)
if final_template_dir:
image_bytes = await self._screenshot_engine.render(
html=html_content,
base_url_path=THEMES_PATH.parent,
**screenshot_options,
)
if Config.get_config("UI", "CACHE") and use_cache and cache_path:
try:
manifest = TemplateManifest.parse_file(
final_template_dir / "manifest.json"
)
relative_template_name = (
Path(template_name) / manifest.entrypoint
).as_posix()
except (ValidationError, FileNotFoundError) as e:
logger.error(f"无法加载模板 '{template_name}' 的清单文件: {e}")
manifest = None
async with aiofiles.open(cache_path, "wb") as f:
await f.write(image_bytes)
logger.debug(f"UI缓存写入成功: {cache_path}")
except Exception as e:
logger.warning(f"UI缓存写入失败: {e}", e=e)
if not manifest or not final_template_dir:
raise ValueError(f"模板 '{template_name}' 未找到或清单文件加载失败。")
engine_name = manifest.engine
engine = self._engines.get(engine_name)
if not engine:
raise ValueError(f"未找到名为 '{engine_name}' 的渲染引擎。")
jinja_environment = self._get_or_create_jinja_env(self._current_theme_data)
final_render_options = manifest.render_options.copy()
final_render_options.update(render_options_override)
image_bytes = await engine.render(
template_name=relative_template_name,
data=data,
theme=self._current_theme_data,
jinja_env=jinja_environment,
extra_css_paths=extra_css_paths,
custom_css_path=custom_markdown_css_path,
**final_render_options,
)
if Config.get_config("UI", "CACHE") and use_cache and cache_path:
try:
async with aiofiles.open(cache_path, "wb") as f:
await f.write(image_bytes)
logger.debug(f"UI缓存写入成功: {cache_path}")
except Exception as e:
logger.warning(f"UI缓存写入失败: {e}", e=e)
return image_bytes
return RenderResult(image_bytes=image_bytes, html_content=html_content)
except Exception as e:
logger.error(
f"渲染模板 '{template_name}' 时发生错误", "RendererService", e=e
f"渲染组件 '{component.__class__.__name__}' 时发生错误",
"RendererService",
e=e,
)
raise RenderingError(f"渲染模板 '{template_name}' 失败") from e
raise RenderingError(
f"渲染组件 '{component.__class__.__name__}' 失败"
) from e
async def render(
self,
component: Renderable,
use_cache: bool = False,
debug_mode: Literal["none", "log"] = "none",
**render_options,
) -> bytes:
"""
统一的多态的渲染入口直接返回图片字节
参数:
component: 一个 Renderable 实例 ( RenderableComponent) 或一个
模板路径字符串
use_cache: (可选) 是否启用渲染缓存默认为 False
**render_options: 传递给底层渲染引擎的额外参数
返回:
bytes: 渲染后的图片数据
"""
result = await self._render_component(
component,
use_cache=use_cache,
**render_options,
)
if debug_mode == "log" and result.html_content:
logger.info(
f"--- [UI DEBUG] HTML for {component.__class__.__name__} ---\n"
f"{result.html_content}\n"
f"--- [UI DEBUG] End of HTML ---"
)
if result.image_bytes is None:
raise RenderingError("渲染成功但未能生成图片字节数据。")
return result.image_bytes
async def render_to_html(self, component: Renderable) -> str:
"""调试方法只执行到HTML生成步骤。"""
if not self._initialized:
await self.initialize()
assert self._theme_manager is not None, "ThemeManager 未初始化"
return await self._theme_manager._render_component_to_html(component)
async def reload_theme(self) -> str:
"""
重新加载当前主题的配置和样式并清除缓存的Jinja环境
"""
if not self._initialized:
await self.initialize()
assert self._theme_manager is not None, "ThemeManager 未初始化"
current_theme_name = Config.get_config("UI", "THEME", "default")
await self._theme_manager.load_theme(current_theme_name)
logger.info(f"主题 '{current_theme_name}' 已成功重载。")
return current_theme_name

View File

@ -0,0 +1,267 @@
from collections.abc import Callable
import inspect
from pathlib import Path
from typing import Any
import aiofiles
from jinja2 import (
ChoiceLoader,
Environment,
FileSystemLoader,
PrefixLoader,
TemplateNotFound,
select_autoescape,
)
import markdown
from pydantic import BaseModel
import ujson as json
from zhenxun.configs.path_config import THEMES_PATH
from zhenxun.services.log import logger
from zhenxun.services.renderer.models import TemplateManifest
from zhenxun.services.renderer.protocols import Renderable
from zhenxun.utils.exception import RenderingError
from zhenxun.utils.pydantic_compat import model_dump
class Theme(BaseModel):
name: str
palette: dict[str, Any]
style_css: str = ""
assets_dir: Path
default_assets_dir: Path
class ThemeManager:
def __init__(
self,
plugin_template_paths: dict[str, Path],
custom_filters: dict[str, Callable],
custom_globals: dict[str, Callable],
markdown_styles: dict[str, Path],
):
prefix_loader = PrefixLoader(
{
namespace: FileSystemLoader(str(path.absolute()))
for namespace, path in plugin_template_paths.items()
}
)
theme_loader = FileSystemLoader(
[
str(THEMES_PATH / "current_theme_placeholder" / "templates"),
str(THEMES_PATH / "default" / "templates"),
]
)
final_loader = ChoiceLoader([prefix_loader, theme_loader])
self.jinja_env = Environment(
loader=final_loader,
enable_async=True,
autoescape=select_autoescape(["html", "xml"]),
)
self.current_theme: Theme | None = None
self._custom_filters = custom_filters
self._custom_globals = custom_globals
self._markdown_styles = markdown_styles
self.jinja_env.globals["resolve_template"] = self._resolve_component_template
self.jinja_env.filters["md"] = self._markdown_filter
@staticmethod
def _markdown_filter(text: str) -> str:
"""一个将 Markdown 文本转换为 HTML 的 Jinja2 过滤器。"""
if not isinstance(text, str):
return ""
return markdown.markdown(
text,
extensions=[
"pymdownx.tasklist",
"tables",
"fenced_code",
"codehilite",
"mdx_math",
"pymdownx.tilde",
],
extension_configs={"mdx_math": {"enable_dollar_delimiter": True}},
)
async def load_theme(self, theme_name: str = "default"):
theme_dir = THEMES_PATH / theme_name
if not theme_dir.is_dir():
logger.error(f"主题 '{theme_name}' 不存在,将回退到默认主题。")
if theme_name == "default":
raise FileNotFoundError("默认主题 'default' 未找到!")
theme_name = "default"
theme_dir = THEMES_PATH / "default"
if self.jinja_env.loader and isinstance(self.jinja_env.loader, ChoiceLoader):
current_loaders = list(self.jinja_env.loader.loaders)
if len(current_loaders) > 1:
current_loaders[1] = FileSystemLoader(
[
str(theme_dir / "templates"),
str(THEMES_PATH / "default" / "templates"),
]
)
self.jinja_env.loader = ChoiceLoader(current_loaders)
else:
logger.error("Jinja2 loader 不是 ChoiceLoader 或未设置,无法更新主题路径。")
palette_path = theme_dir / "palette.json"
palette = (
json.loads(palette_path.read_text("utf-8")) if palette_path.exists() else {}
)
self.current_theme = Theme(
name=theme_name,
palette=palette,
assets_dir=theme_dir / "assets",
default_assets_dir=THEMES_PATH / "default" / "assets",
)
theme_context_dict = {
"name": theme_name,
"palette": palette,
"assets_dir": theme_dir / "assets",
"default_assets_dir": THEMES_PATH / "default" / "assets",
}
self.jinja_env.globals["theme"] = theme_context_dict
logger.info(f"主题管理器已加载主题: {theme_name}")
async def _resolve_component_template(self, component_path: str) -> str:
"""
智能解析组件路径
如果路径是目录则查找 manifest.json 以获取入口点
"""
if Path(component_path).suffix:
return component_path
manifest_path_str = f"{component_path}/manifest.json"
if not self.jinja_env.loader:
raise TemplateNotFound(
f"Jinja2 loader 未配置。无法查找 '{manifest_path_str}'"
)
try:
_, full_path, _ = self.jinja_env.loader.get_source(
self.jinja_env, manifest_path_str
)
if full_path and Path(full_path).exists():
async with aiofiles.open(full_path, encoding="utf-8") as f:
manifest_data = json.loads(await f.read())
entrypoint = manifest_data.get("entrypoint")
if not entrypoint:
raise RenderingError(
f"组件 '{component_path}' 的 manifest.json 中缺少 "
f"'entrypoint' 键。"
)
return f"{component_path}/{entrypoint}"
except TemplateNotFound:
logger.debug(
f"未找到 '{manifest_path_str}',将回退到默认的 'main.html' 入口点。"
)
return f"{component_path}/main.html"
raise TemplateNotFound(f"无法为组件 '{component_path}' 找到模板入口点。")
async def get_template_manifest(
self, component_path: str
) -> TemplateManifest | None:
"""
查找并解析组件的 manifest.json 文件
"""
manifest_path_str = f"{component_path}/manifest.json"
if not self.jinja_env.loader:
return None
try:
_, full_path, _ = self.jinja_env.loader.get_source(
self.jinja_env, manifest_path_str
)
if full_path and Path(full_path).exists():
async with aiofiles.open(full_path, encoding="utf-8") as f:
manifest_data = json.loads(await f.read())
return TemplateManifest(**manifest_data)
except TemplateNotFound:
return None
return None
def _resolve_markdown_style_path(self, style_name: str) -> Path | None:
"""
按照 注册 -> 主题约定 -> 默认约定 的顺序解析 Markdown 样式路径
"""
if style_name in self._markdown_styles:
logger.debug(f"找到已注册的 Markdown 样式: '{style_name}'")
return self._markdown_styles[style_name]
logger.warning(f"样式 '{style_name}' 在注册表中未找到。")
return None
async def _render_component_to_html(
self,
component: Renderable,
required_scripts: list[str] | None = None,
required_styles: list[str] | None = None,
**kwargs,
) -> str:
"""将 Renderable 组件渲染成 HTML 字符串,并处理异步数据。"""
if not self.current_theme:
await self.load_theme()
assert self.current_theme is not None, "主题加载失败"
data_dict = component.get_render_data()
custom_style_css = ""
if hasattr(component, "get_extra_css"):
css_result = component.get_extra_css(self)
if inspect.isawaitable(css_result):
custom_style_css = await css_result
else:
custom_style_css = css_result
def asset_loader(asset_path: str) -> str:
"""[新增] 用于在Jinja2模板中解析静态资源的辅助函数。"""
assert self.current_theme is not None
current_theme_asset = self.current_theme.assets_dir / asset_path
if current_theme_asset.exists():
return current_theme_asset.relative_to(THEMES_PATH.parent).as_posix()
default_theme_asset = self.current_theme.default_assets_dir / asset_path
if default_theme_asset.exists():
return default_theme_asset.relative_to(THEMES_PATH.parent).as_posix()
logger.warning(
f"资源文件在主题 '{self.current_theme.name}''default' 中均未找到: "
f"{asset_path}"
)
return ""
theme_context_dict = model_dump(self.current_theme)
theme_context_dict["asset"] = asset_loader
resolved_template_name = await self._resolve_component_template(
str(component.template_name)
)
logger.debug(
f"正在渲染组件 '{component.template_name}' "
f"(主题: {self.current_theme.name}),解析模板: '{resolved_template_name}'",
"RendererService",
)
if self._custom_filters:
self.jinja_env.filters.update(self._custom_filters)
if self._custom_globals:
self.jinja_env.globals.update(self._custom_globals)
template = self.jinja_env.get_template(resolved_template_name)
template_context = {
"data": data_dict,
"theme": theme_context_dict,
"theme_css": "",
"custom_style_css": custom_style_css,
"required_scripts": required_scripts or [],
"required_styles": required_styles or [],
}
template_context.update(kwargs)
return await template.render_async(**template_context)

View File

@ -1,40 +1,140 @@
from . import builders, models
from .builders import (
InfoCardBuilder,
LayoutBuilder,
MarkdownBuilder,
NotebookBuilder,
PluginHelpPageBuilder,
PluginMenuBuilder,
TableBuilder,
)
from .models import (
HelpCategory,
HelpItem,
InfoCardData,
PluginHelpPageData,
PluginMenuCategory,
PluginMenuData,
PluginMenuItem,
RenderableComponent,
)
from pathlib import Path
from typing import Any, Literal
__all__ = [
"HelpCategory",
"HelpItem",
"InfoCardBuilder",
"InfoCardData",
"LayoutBuilder",
"MarkdownBuilder",
"NotebookBuilder",
"PluginHelpPageBuilder",
"PluginHelpPageData",
"PluginMenuBuilder",
"PluginMenuCategory",
"PluginMenuData",
"PluginMenuItem",
"RenderableComponent",
"TableBuilder",
"builders",
"models",
]
from zhenxun.services.renderer.protocols import Renderable
from .builders.core.layout import LayoutBuilder
from .models.core.base import RenderableComponent
from .models.core.markdown import MarkdownData
from .models.core.template import TemplateComponent
def template(path: str | Path, data: dict[str, Any]) -> TemplateComponent:
"""
创建一个基于独立模板文件的UI组件
"""
if isinstance(path, str):
path = Path(path)
return TemplateComponent(template_path=path, data=data)
def markdown(content: str, style: str | Path | None = "default") -> MarkdownData:
"""
创建一个基于Markdown内容的UI组件
"""
if isinstance(style, Path):
return MarkdownData(markdown=content, css_path=str(style.absolute()))
return MarkdownData(markdown=content, style_name=style)
def vstack(children: list[RenderableComponent], **layout_options) -> "LayoutBuilder":
"""
创建一个垂直布局组件
"""
builder = LayoutBuilder.column(**layout_options)
for child in children:
builder.add_item(child)
return builder
def hstack(children: list[RenderableComponent], **layout_options) -> "LayoutBuilder":
"""
创建一个水平布局组件
"""
builder = LayoutBuilder.row(**layout_options)
for child in children:
builder.add_item(child)
return builder
async def render(
component_or_path: Renderable | str | Path,
data: dict | None = None,
*,
use_cache: bool = False,
debug_mode: Literal["none", "log"] = "none",
**kwargs,
) -> bytes:
"""
统一的UI渲染入口
用法:
1. 渲染一个已构建的UI组件: `render(my_builder.build())`
2. 直接渲染一个模板文件: `render("path/to/template", data={...})`
"""
from zhenxun.services import renderer_service
component: Renderable
if isinstance(component_or_path, str | Path):
if data is None:
raise ValueError("使用模板路径渲染时必须提供 'data' 参数。")
component = TemplateComponent(template_path=component_or_path, data=data)
else:
component = component_or_path
return await renderer_service.render(
component, use_cache=use_cache, debug_mode=debug_mode, **kwargs
)
async def render_template(
path: str | Path, data: dict, use_cache: bool = False, **kwargs
) -> bytes:
"""
渲染一个独立的Jinja2模板文件
这是一个便捷函数封装了 render() 函数的调用提供更简洁的模板渲染接口
参数:
path: 模板文件路径相对于主题模板目录
data: 传递给模板的数据字典
use_cache: (可选) 是否启用渲染缓存默认为 False
**kwargs: 传递给渲染服务的额外参数
返回:
bytes: 渲染后的图片数据
"""
return await render(path, data, use_cache=use_cache, **kwargs)
async def render_markdown(
md: str, style: str | Path | None = "default", use_cache: bool = False, **kwargs
) -> bytes:
"""
将Markdown字符串渲染为图片
这是一个便捷函数封装了 render() 函数的调用专门用于渲染Markdown内容
参数:
md: 要渲染的Markdown内容字符串
style: (可选) 样式名称或自定义CSS文件路径默认为 "default"
use_cache: (可选) 是否启用渲染缓存默认为 False
**kwargs: 传递给渲染服务的额外参数
返回:
bytes: 渲染后的图片数据
"""
component: MarkdownData
if isinstance(style, Path):
component = MarkdownData(markdown=md, css_path=str(style.absolute()))
else:
component = MarkdownData(markdown=md, style_name=style)
return await render(component, use_cache=use_cache, **kwargs)
from zhenxun.services.renderer.protocols import RenderResult
async def render_full_result(
component: Renderable, use_cache: bool = False, **kwargs
) -> RenderResult:
"""
渲染组件并返回包含图片和HTML的完整结果对象用于调试和高级用途
"""
from zhenxun.services import renderer_service
return await renderer_service._render_component(
component, use_cache=use_cache, **kwargs
)

View File

@ -3,8 +3,6 @@ from typing_extensions import Self
from pydantic import BaseModel
from zhenxun.services import renderer_service
T_DataModel = TypeVar("T_DataModel", bound=BaseModel)
@ -15,6 +13,12 @@ class BaseBuilder(Generic[T_DataModel]):
self._data: T_DataModel = data_model
self._style_name: str | None = None
self._template_name = template_name
self._inline_style: dict | None = None
self._extra_css: str | None = None
@property
def data(self) -> T_DataModel:
return self._data
def with_style(self, style_name: str) -> Self:
"""
@ -23,19 +27,31 @@ class BaseBuilder(Generic[T_DataModel]):
self._style_name = style_name
return self
async def build(self, use_cache: bool = False, **render_options) -> bytes:
def with_inline_style(self, style: dict[str, str]) -> Self:
"""
通用的构建方法将数据渲染为图片
为组件的根元素应用动态的内联样式
参数:
style: 一个CSS样式字典例如 {"background-color":"#fff","font-size":"16px"}
"""
self._inline_style = style
return self
def with_extra_css(self, css: str) -> Self:
"""
向页面注入一段自定义的CSS样式字符串
参数:
css: 包含CSS规则的字符串
"""
self._extra_css = css
return self
def build(self) -> T_DataModel:
"""
构建并返回配置好的数据模型
"""
if self._style_name and hasattr(self._data, "style_name"):
setattr(self._data, "style_name", self._style_name)
data_to_render = self._data
return await renderer_service.render(
template_name=self._template_name,
data=data_to_render,
use_cache=use_cache,
style_name=self._style_name,
**render_options,
)
return self._data

View File

@ -1,7 +1,7 @@
import base64
from typing import Any
from typing_extensions import Self
from ...models.core.base import RenderableComponent
from ...models.core.layout import LayoutData, LayoutItem
from ..base import BaseBuilder
@ -10,108 +10,100 @@ __all__ = ["LayoutBuilder"]
class LayoutBuilder(BaseBuilder[LayoutData]):
"""
一个用于将多个图片bytes组合成单张图片的链式构建器
采用混合模式提供便捷的工厂方法和灵活的自定义模板能力
一个用于将多个UI组件组合成单张图片的链式构建器
它通过在单个渲染流程中动态包含子模板来实现高质量的输出
"""
def __init__(self):
super().__init__(LayoutData(), template_name="")
self._items: list[LayoutItem] = []
self._options: dict[str, Any] = {}
self._preset_template_name: str | None = None
@classmethod
def column(cls, **options: Any) -> Self:
"""
工厂方法创建一个垂直列布局的构建器
:param options: 传递给模板的选项 gap, padding, align_items
"""
builder = cls()
builder._preset_template_name = "layouts/column"
builder._template_name = "layouts/column"
builder._options.update(options)
return builder
@classmethod
def grid(cls, **options: Any) -> Self:
"""
工厂方法创建一个网格布局的构建器
:param options: 传递给模板的选项 columns, gap, padding
"""
def row(cls, **options: Any) -> Self:
builder = cls()
builder._preset_template_name = "layouts/grid"
builder._template_name = "layouts/row"
builder._options.update(options)
return builder
@classmethod
def vstack(cls, images: list[bytes], **options: Any) -> Self:
"""
工厂方法创建一个垂直堆叠布局的构建器并直接添加图片
def hstack(
cls, components: list["BaseBuilder | RenderableComponent"], **options: Any
) -> Self:
builder = cls.row(**options)
for component in components:
builder.add_item(component)
return builder
参数:
images: 要垂直堆叠的图片字节流列表
options: 传递给模板的选项 gap, padding, align_items
"""
@classmethod
def vstack(
cls, components: list["BaseBuilder | RenderableComponent"], **options: Any
) -> Self:
builder = cls.column(**options)
for image_bytes in images:
builder.add_item(image_bytes)
return builder
@classmethod
def hstack(cls, images: list[bytes], **options: Any) -> Self:
"""
工厂方法创建一个水平堆叠布局的构建器并直接添加图片
参数:
images: 要水平堆叠的图片字节流列表
options: 传递给模板的选项 gap, padding, align_items
"""
builder = cls()
builder._preset_template_name = "layouts/row"
builder._options.update(options)
for image_bytes in images:
builder.add_item(image_bytes)
for component in components:
builder.add_item(component)
return builder
def add_item(
self, image_bytes: bytes, metadata: dict[str, Any] | None = None
self,
component: "BaseBuilder | RenderableComponent",
metadata: dict[str, Any] | None = None,
) -> Self:
"""
向布局中添加一个图片项目
:param image_bytes: 图片的原始字节数据
:param metadata: (可选) 与此项目关联的元数据可用于模板
向布局中添加一个组件支持多种组件类型的添加
参数:
component: 一个 Builder 实例 ( TableBuilder) 或一个 RenderableComponent
数据模型
metadata: (可选) 与此项目关联的元数据可用于模板
返回:
Self: 返回当前布局构建器实例支持链式调用
"""
b64_string = base64.b64encode(image_bytes).decode("utf-8")
src = f"data:image/png;base64,{b64_string}"
self._items.append(LayoutItem(src=src, metadata=metadata))
component_data = (
component.data if isinstance(component, BaseBuilder) else component
)
self._data.children.append(
LayoutItem(component=component_data, metadata=metadata)
)
return self
def add_option(self, key: str, value: Any) -> Self:
"""
为布局添加一个自定义选项该选项会传递给模板
参数:
key: 选项的键名用于在模板中引用
value: 选项的值可以是任意类型的数据
返回:
Self: 返回当前布局构建器实例支持链式调用
"""
self._options[key] = value
return self
async def build(
self, use_cache: bool = False, template: str | None = None, **render_options
) -> bytes:
def build(self) -> LayoutData:
"""
构建最终的布局图片
:param use_cache: 是否使用缓存
:param template: (可选) 强制使用指定的模板覆盖工厂方法的预设
这是实现自定义布局的关键
:param render_options: 传递给渲染引擎的额外选项
"""
final_template_name = template or self._preset_template_name
[修改] 构建并返回 LayoutData 模型实例
此方法现在是同步的并且不执行渲染
if not final_template_name:
参数:
返回:
LayoutData: 配置好的布局数据模型
"""
if not self._template_name:
raise ValueError(
"必须通过工厂方法 (如 LayoutBuilder.column()) 或在 build() "
"方法中提供一个模板名称。"
"必须通过工厂方法 (如 LayoutBuilder.column()) 初始化布局类型。"
)
self._data.items = self._items
self._data.options = self._options
self._template_name = final_template_name
return await super().build(use_cache=use_cache, **render_options)
self._data.layout_type = self._template_name.split("/")[-1]
return self._data

View File

@ -140,10 +140,12 @@ class MarkdownBuilder(BaseBuilder[MarkdownData]):
self._append_element(RawHtmlElement(html="---"))
return self
async def build(self, use_cache: bool = False, **render_options) -> bytes:
"""构建Markdown图片"""
def build(self) -> MarkdownData:
"""
构建并返回 MarkdownData 模型实例
"""
final_markdown = "\n\n".join(part.to_markdown() for part in self._parts).strip()
self._data.markdown = final_markdown
self._data.width = self._width
self._data.css_path = self._css_path
return await super().build(use_cache=use_cache, **render_options)
return super().build()

View File

@ -78,10 +78,25 @@ class NotebookBuilder(BaseBuilder[NotebookData]):
self.add_component(Divider(**kwargs))
return self
def add_component(self, component: RenderableComponent) -> "NotebookBuilder":
"""向 Notebook 中添加一个可渲染的自定义组件。"""
def add_component(
self, component: "RenderableComponent | BaseBuilder"
) -> "NotebookBuilder":
"""
Notebook 中添加一个可渲染的自定义组件
"""
component_data = (
component.data if isinstance(component, BaseBuilder) else component
)
if not isinstance(component_data, RenderableComponent):
raise TypeError(
f"add_component 只能接受 RenderableComponent 或其 Builder"
f"但收到了 {type(component)}"
)
self._elements.append(
NotebookElement(type="component", component_data=component)
NotebookElement(type="component", component=component_data)
)
return self
@ -97,11 +112,9 @@ class NotebookBuilder(BaseBuilder[NotebookData]):
self.quote(quote)
return self
async def build(
self, use_cache: bool = False, frameless: bool = False, **render_options
) -> bytes:
"""构建Notebook图片"""
def build(self) -> NotebookData:
"""
构建并返回 NotebookData 模型实例
"""
self._data.elements = self._elements
return await super().build(
use_cache=use_cache, frameless=frameless, **render_options
)
return super().build()

View File

@ -1,13 +1,21 @@
from typing import Literal
import uuid
from pydantic import BaseModel
from pydantic import BaseModel, Field
from .core.base import RenderableComponent
class BaseChartData(BaseModel):
class BaseChartData(RenderableComponent):
"""所有图表数据模型的基类"""
style_name: str | None = None
title: str
chart_id: str = Field(default_factory=lambda: f"chart-{uuid.uuid4().hex}")
def get_required_scripts(self) -> list[str]:
"""声明此组件需要 ECharts 库。"""
return ["js/echarts.min.js"]
class BarChartData(BaseChartData):
@ -18,6 +26,10 @@ class BarChartData(BaseChartData):
direction: Literal["horizontal", "vertical"] = "horizontal"
background_image: str | None = None
@property
def template_name(self) -> str:
return "components/charts/bar_chart"
class PieChartDataItem(BaseModel):
name: str
@ -29,6 +41,10 @@ class PieChartData(BaseChartData):
data: list[PieChartDataItem]
@property
def template_name(self) -> str:
return "components/charts/pie_chart"
class LineChartSeries(BaseModel):
name: str
@ -41,3 +57,7 @@ class LineChartData(BaseChartData):
category_data: list[str]
series: list[LineChartSeries]
@property
def template_name(self) -> str:
return "components/charts/line_chart"

View File

@ -19,4 +19,4 @@ class Badge(RenderableComponent):
@property
def template_name(self) -> str:
return "components/widgets/badge/main.html"
return "components/widgets/badge"

View File

@ -18,7 +18,7 @@ class Divider(RenderableComponent):
@property
def template_name(self) -> str:
return "components/widgets/divider/main.html"
return "components/widgets/divider"
class Rectangle(RenderableComponent):
@ -32,4 +32,4 @@ class Rectangle(RenderableComponent):
@property
def template_name(self) -> str:
return "components/widgets/rectangle/main.html"
return "components/widgets/rectangle"

View File

@ -21,4 +21,4 @@ class ProgressBar(RenderableComponent):
@property
def template_name(self) -> str:
return "components/widgets/progress_bar/main.html"
return "components/widgets/progress_bar"

View File

@ -20,4 +20,4 @@ class UserInfoBlock(RenderableComponent):
@property
def template_name(self) -> str:
return "components/widgets/user_info_block/main.html"
return "components/widgets/user_info_block"

View File

@ -20,6 +20,7 @@ from .markdown import (
)
from .notebook import NotebookData, NotebookElement
from .table import BaseCell, ImageCell, StatusBadgeCell, TableCell, TableData, TextCell
from .template import TemplateComponent
__all__ = [
"BaseCell",
@ -42,6 +43,7 @@ __all__ = [
"TableCell",
"TableData",
"TableElement",
"TemplateComponent",
"TextCell",
"TextElement",
]

View File

@ -1,20 +1,88 @@
"""
核心基础模型定义
用于存放 RenderableComponent 基类
"""
from abc import ABC, abstractmethod
import asyncio
from collections.abc import Awaitable, Iterator
from typing import Any
from nonebot.compat import model_dump
from pydantic import BaseModel
__all__ = ["RenderableComponent"]
from zhenxun.services.renderer.protocols import Renderable
__all__ = ["ContainerComponent", "RenderableComponent"]
class RenderableComponent(BaseModel, ABC):
class RenderableComponent(BaseModel, Renderable):
"""所有可渲染UI组件的抽象基类。"""
_is_standalone_template: bool = False
@property
@abstractmethod
def template_name(self) -> str:
"""返回用于渲染此组件的Jinja2模板的路径。"""
"""
返回用于渲染此组件的Jinja2模板的路径
这是一个抽象属性所有子类都必须覆盖它
"""
raise NotImplementedError(
"Subclasses must implement the 'template_name' property."
)
async def prepare(self) -> None:
"""[可选] 生命周期钩子,默认无操作。"""
pass
def get_required_scripts(self) -> list[str]:
"""[可选] 返回此组件所需的JS脚本路径列表 (相对于assets目录)。"""
return []
def get_required_styles(self) -> list[str]:
"""[可选] 返回此组件所需的CSS样式表路径列表 (相对于assets目录)。"""
return []
def get_render_data(self) -> dict[str, Any | Awaitable[Any]]:
"""默认实现,返回模型自身的数据字典。"""
return model_dump(self)
def get_extra_css(self, theme_manager: Any) -> str | Awaitable[str]:
return ""
class ContainerComponent(RenderableComponent, ABC):
"""
一个为容器类组件设计的抽象基类封装了预渲染子组件的通用逻辑
"""
@abstractmethod
def _get_renderable_child_items(self) -> Iterator[Any]:
"""
一个抽象方法子类必须实现它来返回一个可迭代的对象
迭代器中的每个项目都必须具有 'component' 'html_content' 属性
"""
raise NotImplementedError
async def prepare(self) -> None:
"""
通用的 prepare 方法负责预渲染所有子组件
"""
from zhenxun.services import renderer_service
child_items = list(self._get_renderable_child_items())
if not child_items:
return
components_to_render = [
item.component for item in child_items if item.component
]
prepare_tasks = [
comp.prepare() for comp in components_to_render if hasattr(comp, "prepare")
]
if prepare_tasks:
await asyncio.gather(*prepare_tasks)
render_tasks = [
renderer_service.render_to_html(comp) for comp in components_to_render
]
rendered_htmls = await asyncio.gather(*render_tasks)
for item, html in zip(child_items, rendered_htmls):
item.html_content = html

View File

@ -2,23 +2,48 @@ from typing import Any
from pydantic import BaseModel, Field
from .base import ContainerComponent, RenderableComponent
__all__ = ["LayoutData", "LayoutItem"]
class LayoutItem(BaseModel):
"""布局中的单个项目,通常是一张图片"""
"""布局中的单个项目,现在持有可渲染组件的数据模型"""
src: str = Field(..., description="图片的Base64数据URI")
component: RenderableComponent = Field(..., description="要渲染的组件的数据模型")
metadata: dict[str, Any] | None = Field(None, description="传递给模板的额外元数据")
html_content: str | None = None
class LayoutData(BaseModel):
class LayoutData(ContainerComponent):
"""布局构建器的数据模型"""
style_name: str | None = None
items: list[LayoutItem] = Field(
layout_type: str = "column"
children: list[LayoutItem] = Field(
default_factory=list, description="要布局的项目列表"
)
options: dict[str, Any] = Field(
default_factory=dict, description="传递给模板的布局选项"
default_factory=dict, description="传递给模板的选项"
)
def get_required_scripts(self) -> list[str]:
"""[新增] 聚合所有子组件的脚本依赖。"""
scripts = set()
for item in self.children:
scripts.update(item.component.get_required_scripts())
return list(scripts)
def get_required_styles(self) -> list[str]:
"""[新增] 聚合所有子组件的样式依赖。"""
styles = set()
for item in self.children:
styles.update(item.component.get_required_styles())
return list(styles)
@property
def template_name(self) -> str:
return f"layouts/{self.layout_type}"
def _get_renderable_child_items(self):
yield from self.children

View File

@ -1,8 +1,14 @@
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Literal
import aiofiles
from pydantic import BaseModel, Field
from zhenxun.services.log import logger
from .base import RenderableComponent
__all__ = [
"CodeElement",
"HeadingElement",
@ -115,10 +121,35 @@ class ListElement(ContainerElement):
return "\n".join(lines)
class MarkdownData(BaseModel):
class MarkdownData(RenderableComponent):
"""Markdown转图片的数据模型"""
style_name: str | None = None
markdown: str
width: int = 800
css_path: str | None = None
@property
def template_name(self) -> str:
return "components/core/markdown"
async def get_extra_css(self, theme_manager) -> str:
if self.css_path:
css_file = Path(self.css_path)
if css_file.is_file():
async with aiofiles.open(css_file, encoding="utf-8") as f:
return await f.read()
else:
logger.warning(f"Markdown自定义CSS文件不存在: {self.css_path}")
else:
style_name = self.style_name or "github-light"
css_path = (
theme_manager.current_theme.default_assets_dir
/ "css"
/ "markdown"
/ f"{style_name}.css"
)
if css_path.exists():
async with aiofiles.open(css_path, encoding="utf-8") as f:
return await f.read()
return ""

View File

@ -2,7 +2,7 @@ from typing import Literal
from pydantic import BaseModel
from .base import RenderableComponent
from .base import ContainerComponent, RenderableComponent
__all__ = ["NotebookData", "NotebookElement"]
@ -28,11 +28,21 @@ class NotebookElement(BaseModel):
language: str | None = None
data: list[str] | None = None
ordered: bool | None = None
component_data: RenderableComponent | None = None
component: RenderableComponent | None = None
html_content: str | None = None
class NotebookData(BaseModel):
class NotebookData(ContainerComponent):
"""Notebook转图片的数据模型"""
style_name: str | None = None
elements: list[NotebookElement]
@property
def template_name(self) -> str:
return "components/core/notebook"
def _get_renderable_child_items(self):
for element in self.elements:
if element.type == "component" and element.component:
yield element

View File

@ -2,6 +2,8 @@ from typing import Literal
from pydantic import BaseModel, Field
from .base import RenderableComponent
__all__ = [
"BaseCell",
"ImageCell",
@ -49,7 +51,7 @@ class StatusBadgeCell(BaseCell):
TableCell = TextCell | ImageCell | StatusBadgeCell | str | int | float | None
class TableData(BaseModel):
class TableData(RenderableComponent):
"""通用表格的数据模型"""
style_name: str | None = None
@ -57,3 +59,7 @@ class TableData(BaseModel):
tip: str | None = Field(None, description="表格下方的提示信息")
headers: list[str] = Field(default_factory=list, description="表头列表")
rows: list[list[TableCell]] = Field(default_factory=list, description="数据行列表")
@property
def template_name(self) -> str:
return "components/core/table"

View File

@ -0,0 +1,25 @@
from pathlib import Path
from typing import Any
from .base import RenderableComponent
__all__ = ["TemplateComponent"]
class TemplateComponent(RenderableComponent):
"""基于独立模板文件的UI组件"""
_is_standalone_template: bool = True
template_path: str | Path
data: dict[str, Any]
@property
def template_name(self) -> str:
"""返回模板路径"""
if isinstance(self.template_path, Path):
return self.template_path.as_posix()
return str(self.template_path)
def get_render_data(self) -> dict[str, Any]:
"""返回传递给模板的数据"""
return self.data

View File

@ -33,5 +33,4 @@ class InfoCardData(RenderableComponent):
@property
def template_name(self) -> str:
"""返回用于渲染此组件的Jinja2模板的路径。"""
return "components/presets/info_card/main.html"
return "components/presets/info_card"

View File

@ -35,4 +35,4 @@ class PluginHelpPageData(RenderableComponent):
@property
def template_name(self) -> str:
return "pages/core/help_page/main.html"
return "pages/core/help_page"

View File

@ -39,4 +39,4 @@ class PluginMenuData(RenderableComponent):
@property
def template_name(self) -> str:
return "pages/core/plugin_menu/main.html"
return "pages/core/plugin_menu"

View File

@ -2,8 +2,8 @@ import os
from pathlib import Path
import random
from zhenxun.services import renderer_service
from zhenxun.utils._build_image import BuildImage
from zhenxun import ui
from zhenxun.ui.models import BarChartData
from .models import Barh
@ -14,18 +14,19 @@ BACKGROUND_PATH = (
class ChartUtils:
@classmethod
async def barh(cls, data: Barh) -> BuildImage:
async def barh(cls, data: Barh) -> bytes:
"""横向统计图"""
background_image_name = random.choice(os.listdir(BACKGROUND_PATH))
render_data = {
"title": data.title,
"category_data": data.category_data,
"data": data.data,
"background_image": background_image_name,
"direction": "horizontal",
}
image_bytes = await renderer_service.render(
"components/charts/bar_chart", data=render_data
background_image_name = (
random.choice(os.listdir(BACKGROUND_PATH))
if BACKGROUND_PATH.exists()
else None
)
return BuildImage.open(image_bytes)
chart_component = BarChartData(
title=data.title,
category_data=data.category_data,
data=data.data,
background_image=background_image_name,
direction="horizontal",
)
return await ui.render(chart_component)

View File

@ -6,12 +6,12 @@ import aiofiles
import nonebot
from pydantic import BaseModel
from zhenxun import ui
from zhenxun.configs.config import BotConfig, Config
from zhenxun.configs.path_config import DATA_PATH
from zhenxun.configs.utils.models import PluginExtraData
from zhenxun.models.statistics import Statistics
from zhenxun.models.user_console import UserConsole
from zhenxun.services import renderer_service
from zhenxun.services.log import logger
from zhenxun.utils.platform import PlatformUtils
from zhenxun.utils.pydantic_compat import model_dump
@ -161,8 +161,10 @@ class BotProfileManager:
"tags": tags,
"title": f"{BotConfig.self_nickname}简介",
}
return await renderer_service.render(
"pages/builtin/bot_profile", data=profile_data
return await ui.render_template(
"pages/builtin/bot_profile",
data=profile_data,
use_cache=True,
)