更新 Celeste Classic 功能

konabot/plugins/celeste_classic/__init__.py

@@ -0,0 +1,132 @@
+from pathlib import Path
+import subprocess
+import tempfile
+from loguru import logger
+from nonebot import on_command
+from pydantic import BaseModel
+
+from nonebot.adapters import Event, Bot
+from nonebot_plugin_alconna import UniMessage, UniMsg
+from nonebot.adapters.onebot.v11.event import MessageEvent as OB11MessageEvent
+
+from konabot.common.artifact import ArtifactDepends, ensure_artifact, register_artifacts
+from konabot.common.data_man import DataManager
+from konabot.common.path import BINARY_PATH, DATA_PATH
+
+
+arti_ccleste_wrap_linux = ArtifactDepends(
+    url="https://github.com/Passthem-desu/pt-ccleste-wrap/releases/download/v0.1.5/ccleste-wrap",
+    sha256="ba4118c6465d1ca1547cdd1bd11c6b9e6a6a98ea8967b55485aeb6b77bb7e921",
+    target=BINARY_PATH / "ccleste-wrap",
+    required_os="Linux",
+    required_arch="x86_64",
+)
+arti_ccleste_wrap_windows = ArtifactDepends(
+    url="https://github.com/Passthem-desu/pt-ccleste-wrap/releases/download/v0.1.5/ccleste-wrap.exe",
+    sha256="7df382486a452485cdcf2115eabd7f772339ece470ab344074dc163fc7981feb",
+    target=BINARY_PATH / "ccleste-wrap.exe",
+    required_os="Windows",
+    required_arch="AMD64",
+)
+
+
+register_artifacts(arti_ccleste_wrap_linux)
+register_artifacts(arti_ccleste_wrap_windows)
+
+
+class CelesteStatus(BaseModel):
+    records: dict[str, str] = {}
+
+
+celeste_status = DataManager(CelesteStatus, DATA_PATH / "celeste-status.json")
+
+
+cmd = on_command(cmd="celeste", aliases={"蔚蓝", "爬山", "鳌太线"})
+
+
+@cmd.handle()
+async def _(msg: UniMsg, evt: Event, bot: Bot):
+    prev = None
+    if isinstance(evt, OB11MessageEvent):
+        if evt.reply is not None:
+            prev = f"QQ:{bot.self_id}:" + str(evt.reply.message_id)
+        else:
+            for seg in evt.get_message():
+                if seg.type == 'reply':
+                    msgid = seg.get('id')
+                    prev = f"QQ:{bot.self_id}:" + str(msgid)
+
+    actions = msg.extract_plain_text().strip().removeprefix("celeste")
+    for alias in {"蔚蓝", "爬山", "鳌太线"}:
+        actions = actions.removeprefix(alias)
+    actions = actions.strip()
+    if len(actions) == 0:
+        return
+
+    if prev is not None:
+        async with celeste_status.get_data() as data:
+            prev = data.records.get(prev)
+
+    await ensure_artifact(arti_ccleste_wrap_linux)
+    await ensure_artifact(arti_ccleste_wrap_windows)
+
+    bin: Path | None = None
+    for arti in (
+        arti_ccleste_wrap_linux,
+        arti_ccleste_wrap_windows,
+    ):
+        if not arti.is_corresponding_platform():
+            continue
+        bin = arti.target
+        if not bin.exists():
+            continue
+        break
+
+    if bin is None:
+        logger.warning("Celeste 模块没有找到该系统需要的二进制文件")
+        return
+
+    if prev is not None:
+        prev_append = ["-p", prev]
+    else:
+        prev_append = []
+
+    try:
+        with tempfile.TemporaryDirectory() as _tempdir:
+            tempdir = Path(_tempdir)
+            gif_path = tempdir / "render.gif"
+            cmd_celeste = [
+                bin,
+                "-a",
+                actions,
+                "-o",
+                gif_path,
+            ] + prev_append
+            logger.info(f"执行指令调用 celeste: CMD={cmd_celeste}")
+            res = subprocess.run(cmd_celeste, timeout=5, capture_output=True)
+            if res.returncode != 0:
+                logger.warning(f"渲染 Celeste 时的输出不是 0 CODE={res.returncode} STDOUT={res.stdout} STDERR={res.stderr}")
+                await UniMessage.text(f"渲染 Celeste 时出错啦！下面是输出：\n\n{res.stdout.decode()}{res.stderr.decode()}").send(evt, bot, at_sender=True)
+                return
+            if not gif_path.exists():
+                logger.warning("没有找到 Celeste 渲染的文件")
+                await UniMessage.text("渲染 Celeste 时出错啦！").send(evt, bot, at_sender=True)
+                return
+            gif_data = gif_path.read_bytes()
+    except TimeoutError:
+        logger.warning("在渲染 Celeste 时超时了")
+        await UniMessage("渲染 Celeste 时超时了！请检查你的操作清单，不能太长").send(evt, bot, at_sender=True)
+        return
+
+    receipt = await UniMessage.image(raw=gif_data).send(evt, bot)
+    async with celeste_status.get_data() as data:
+        if prev:
+            actions = prev + "\n" + actions
+        if isinstance(evt, OB11MessageEvent):
+            for _msgid in receipt.msg_ids:
+                msgid = _msgid["message_id"]
+                data.records[f"QQ:{bot.self_id}:{msgid}"] = actions
+        else:
+            for msgid in receipt.msg_ids:
+                data.records[f"DISCORD:{bot.self_id}:{msgid}"] = actions
+