From 97e87c7ec3cee3fca38497422e0cbc0a669b5e7e Mon Sep 17 00:00:00 2001 From: passthem Date: Wed, 18 Mar 2026 17:26:36 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=20Typst=20=E7=9A=84=E4=BA=8C?= =?UTF-8?q?=E8=BF=9B=E5=88=B6=E6=96=87=E4=BB=B6=E4=B8=8B=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- konabot/common/artifact.py | 51 ++++++++++++---- konabot/plugins/typst/__init__.py | 97 +++++++++++++++++++++++++++++-- 2 files changed, 133 insertions(+), 15 deletions(-) diff --git a/konabot/common/artifact.py b/konabot/common/artifact.py index 2c4453b..3f70f61 100644 --- a/konabot/common/artifact.py +++ b/konabot/common/artifact.py @@ -1,9 +1,10 @@ import asyncio +from typing import Any, Awaitable, Callable import aiohttp import hashlib import platform -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path import nonebot @@ -14,6 +15,8 @@ from pydantic import BaseModel @dataclass class ArtifactDepends: + _Callback = Callable[[bool], Awaitable[Any]] + url: str sha256: str target: Path @@ -27,6 +30,9 @@ class ArtifactDepends: use_proxy: bool = True "网络问题，赫赫；使用的是 Discord 模块配置的 proxy" + callbacks: list[_Callback] = field(default_factory=list) + "在任务完成以后，应该做的事情" + def is_corresponding_platform(self) -> bool: if self.required_os is not None: if self.required_os.lower() != platform.system().lower(): @@ -36,26 +42,43 @@ class ArtifactDepends: return False return True + def on_finished(self, task: _Callback) -> _Callback: + self.callbacks.append(task) + return task + + async def _finished(self, downloaded: bool) -> list[Any | BaseException]: + tasks = set() + for f in self.callbacks: + tasks.add(f(downloaded)) + return await asyncio.gather(*tasks, return_exceptions=True) + class Config(BaseModel): prefetch_artifact: bool = False "是否提前下载好二进制依赖" -artifact_list = [] +artifact_list: list[ArtifactDepends] = [] driver = nonebot.get_driver() config = nonebot.get_plugin_config(Config) + @driver.on_startup async def _(): if config.prefetch_artifact: logger.info("启动检测中：正在检测需求的二进制是否下载") semaphore = asyncio.Semaphore(10) + async def _task(artifact: ArtifactDepends): async with semaphore: - await ensure_artifact(artifact) + downloaded = await ensure_artifact(artifact) + result = await artifact._finished(downloaded) + for r in result: + if isinstance(r, BaseException): + logger.warning("完成了二进制文件的下载，但是有未捕捉的错误") + logger.exception(r) tasks: set[asyncio.Task] = set() for a in artifact_list: @@ -78,35 +101,43 @@ async def download_artifact(artifact: ArtifactDepends): async with aiohttp.ClientSession(proxy=proxy) as client: result = await client.get(artifact.url) if result.status != 200: - logger.warning(f"已经下载了二进制，但是注意服务器没有返回 200！ URL={artifact.url} TARGET={artifact.target} CODE={result.status}") + logger.warning( + f"已经下载了二进制，但是注意服务器没有返回 200！ URL={artifact.url} TARGET={artifact.target} CODE={result.status}" + ) data = await result.read() artifact.target.write_bytes(data) - if not platform.system().lower() == 'windows': + if not platform.system().lower() == "windows": artifact.target.chmod(0o755) logger.info(f"下载好了 TARGET={artifact.target} URL={artifact.url}") m = hashlib.sha256(artifact.target.read_bytes()) if m.hexdigest().lower() != artifact.sha256.lower(): - logger.warning(f"下载到的二进制的 sha256 与需求不同 TARGET={artifact.target} REQUESTED={artifact.sha256} ACTUAL={m.hexdigest()}") + logger.warning( + f"下载到的二进制的 sha256 与需求不同 TARGET={artifact.target} REQUESTED={artifact.sha256} ACTUAL={m.hexdigest()}" + ) -async def ensure_artifact(artifact: ArtifactDepends): +async def ensure_artifact(artifact: ArtifactDepends) -> bool: if not artifact.is_corresponding_platform(): - return + return False if not artifact.target.exists(): logger.info(f"二进制依赖 {artifact.target} 不存在") if not artifact.target.parent.exists(): artifact.target.parent.mkdir(parents=True, exist_ok=True) await download_artifact(artifact) + return True else: m = hashlib.sha256(artifact.target.read_bytes()) if m.hexdigest().lower() != artifact.sha256.lower(): - logger.info(f"二进制依赖 {artifact.target} 的哈希无法对应需求的哈希，准备重新下载") + logger.info( + f"二进制依赖 {artifact.target} 的哈希无法对应需求的哈希，准备重新下载" + ) artifact.target.unlink() await download_artifact(artifact) + return True + return False def register_artifacts(*artifacts: ArtifactDepends): artifact_list.extend(artifacts) - diff --git a/konabot/plugins/typst/__init__.py b/konabot/plugins/typst/__init__.py index 86d156a..5f47aeb 100644 --- a/konabot/plugins/typst/__init__.py +++ b/konabot/plugins/typst/__init__.py @@ -1,9 +1,11 @@ import asyncio +import os import subprocess from pathlib import Path from tempfile import TemporaryDirectory from typing import cast +import zipfile from loguru import logger from nonebot import on_command @@ -13,22 +15,97 @@ from nonebot.adapters.onebot.v11.event import MessageEvent as OB11MessageEvent from nonebot.adapters.onebot.v11.bot import Bot as OB11Bot from nonebot.adapters.onebot.v11.message import Message as OB11Message +from konabot.common.artifact import ArtifactDepends, register_artifacts from konabot.common.longtask import DepLongTaskTarget -from konabot.common.path import TMP_PATH +from konabot.common.path import BINARY_PATH, TMP_PATH + + +arti_typst_linux = ArtifactDepends( + url="https://github.com/typst/typst/releases/download/v0.14.2/typst-x86_64-unknown-linux-musl.tar.xz", + sha256="a6044cbad2a954deb921167e257e120ac0a16b20339ec01121194ff9d394996d", + target=BINARY_PATH / "typst.tar.xz", + required_os="Linux", + required_arch="x86_64", +) +arti_typst_windows = ArtifactDepends( + url="https://github.com/typst/typst/releases/download/v0.14.2/typst-x86_64-pc-windows-msvc.zip", + sha256="51353994ac83218c3497052e89b2c432c53b9d4439cdc1b361e2ea4798ebfc13", + target=BINARY_PATH / "typst.zip", + required_os="Windows", + required_arch="AMD64", +) + + +bin_path: Path | None = None + + +@arti_typst_linux.on_finished +async def _(downloaded: bool): + global bin_path + + tar_path = arti_typst_linux.target + bin_path = tar_path.with_name("typst") + + if downloaded or not bin_path.exists(): + bin_path.unlink(missing_ok=True) + process = await asyncio.create_subprocess_exec( + "tar", + "-xvf", + tar_path, + "--strip-components=1", + "-C", + BINARY_PATH, + "typst-x86_64-unknown-linux-musl/typst", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + if process.returncode != 0 or not bin_path.exists(): + logger.warning( + "似乎没有成功解压 Typst 二进制文件，检查一下吧！ " + f"stdout={stdout} stderr={stderr}" + ) + else: + os.chmod(bin_path, 0o755) + + +@arti_typst_windows.on_finished +async def _(downloaded: bool): + global bin_path + zip_path = arti_typst_windows.target + bin_path = BINARY_PATH / "typst.exe" + + if downloaded or not bin_path.exists(): + bin_path.unlink(missing_ok=True) + with zipfile.ZipFile(zip_path, "r") as zf: + target_name = "typst-x86_64-pc-windows-msvc/typst.exe" + if target_name not in zf.namelist(): + logger.warning("在 Zip 压缩包里面没有找到目标文件") + return + zf.extract(target_name, BINARY_PATH) + (BINARY_PATH / target_name).rename(bin_path) + (BINARY_PATH / "typst-x86_64-pc-windows-msvc").rmdir() + + +register_artifacts(arti_typst_linux) +register_artifacts(arti_typst_windows) TEMPLATE_PATH = Path(__file__).parent / "template.typ" TEMPLATE = TEMPLATE_PATH.read_text() -def render_sync(code: str) -> bytes: +def render_sync(code: str) -> bytes | None: + if bin_path is None: + return + with TemporaryDirectory(dir=TMP_PATH) as tmpdirname: temp_dir = Path(tmpdirname).resolve() temp_typ = temp_dir / "page.typ" temp_typ.write_text(TEMPLATE + "\n\n" + code) cmd = [ - "typst", + bin_path, "compile", temp_typ.name, "--format", @@ -61,7 +138,7 @@ def render_sync(code: str) -> bytes: return result_png.read_bytes() -async def render(code: str) -> bytes: +async def render(code: str) -> bytes | None: task = asyncio.to_thread(lambda: render_sync(code)) return await task @@ -70,7 +147,15 @@ cmd = on_command("typst") @cmd.handle() -async def _(evt: Event, bot: Bot, msg: UniMsg, target: DepLongTaskTarget): +async def _( + evt: Event, + bot: Bot, + msg: UniMsg, + target: DepLongTaskTarget, +): + if bin_path is None: + return + typst_code = "" if isinstance(evt, OB11MessageEvent): if evt.reply is not None: @@ -92,6 +177,8 @@ async def _(evt: Event, bot: Bot, msg: UniMsg, target: DepLongTaskTarget): try: res = await render(typst_code) + if res is None: + raise FileNotFoundError("没有渲染出来内容") except FileNotFoundError as e: await target.send_message("渲染出错：内部错误") raise e from e