mttu-developers/konabot
8
2
Fork 3
Code Issues 13 Pull Requests 1 Actions Packages Releases 79 Activity

添加 Typst 的二进制文件下载

Browse Source
This commit is contained in:
钟晓帕
2026-03-18 17:26:36 +08:00
parent 8c1651ad3d
commit 97e87c7ec3
2 changed files with 133 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
konabot/common/artifact.py
View File

@ -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)

97
konabot/plugins/typst/__init__.py
View File

@ -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