添加若干有用的小模块

konabot/plugins/longtask_core.py

@@ -0,0 +1,50 @@
+import asyncio
+
+# import datetime
+from loguru import logger
+import nonebot
+
+# from nonebot.adapters import Bot, Event
+# from nonebot_plugin_alconna import UniMessage
+from konabot.common.longtask import (
+    # DepLongTaskTarget,
+    # LongTask,
+    # create_longtask,
+    # get_long_task_target,
+    # handle_long_task,
+    init_longtask,
+)
+
+
+driver = nonebot.get_driver()
+INIT_FLAG = {"flag": False}
+
+
+@driver.on_bot_connect
+async def _():
+    if INIT_FLAG["flag"]:
+        return
+    INIT_FLAG["flag"] = True
+    logger.info("有 Bot 连接，等待 5 秒后初始化 LongTask 模块")
+    await asyncio.sleep(5)
+    await init_longtask()
+    logger.info("LongTask 初始化完成")
+
+
+# cmd1 = nonebot.on_command("test114")
+#
+#
+# @handle_long_task("test_callback_001")
+# async def _(lt: LongTask):
+#     await lt.target.send_message(UniMessage().text("Hello, world!"), at=True)
+#
+#
+# @cmd1.handle()
+# async def _(target: DepLongTaskTarget):
+#     await create_longtask(
+#         handler="test_callback_001",
+#         data={},
+#         target=target,
+#         deadline=datetime.datetime.now() + datetime.timedelta(seconds=2),
+#     )
+# await target.send_message(UniMessage().text("Hello, world!"), at=True)