Merge branch 'master' of https://gitea.service.jazzwhom.top/mttu-developers/konabot
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
2025-12-09 20:33:20 +08:00
4 changed files with 34 additions and 37 deletions

View File

@ -1,10 +1,10 @@
## 指令介绍
**`ntfy`** - 配置使用 [ntfy](https://ntfy.sh/) 来更好地为你通知此方 BOT 的办事项。
**`ntfy`** - 配置使用 [ntfy](https://ntfy.sh/) 来更好地为你通知此方 BOT 的办事项。
## 指令示例
- **`ntfy 创建`**
创建一个随机的 ntfy 订阅主题来提醒办。此方 Bot 将会给你使用指引。你可以前往 [https://ntfy.sh/](https://ntfy.sh/) 官网下载 ntfy APP或者使用网页版 ntfy。
创建一个随机的 ntfy 订阅主题来提醒办。此方 Bot 将会给你使用指引。你可以前往 [https://ntfy.sh/](https://ntfy.sh/) 官网下载 ntfy APP或者使用网页版 ntfy。
- **`ntfy 创建 kagami-notice`**
创建一个名称包含 `kagami-notice` 的 ntfy 订阅主题。

View File

@ -1,8 +1,11 @@
from io import BytesIO
import random
from PIL import Image
from konabot.common.web_render import konaweb
from konabot.common.web_render.core import WebRenderer
import numpy as np
from playwright.async_api import Page
class NoticeUI:
@ -18,13 +21,19 @@ class NoticeUI:
# 直到 setContent 函数加载完成
await page.wait_for_function("typeof setContent === 'function'", timeout=1000)
# 设置标题和消息内容
await page.evaluate(f'setContent("{title}", "{message}")')
await page.evaluate("""([title, message]) => {
return setContent(title, message);
}""",
[title, message])
async def mask_function(page: Page):
# 直到 setContent 函数加载完成
await page.wait_for_function("typeof setContent === 'function'", timeout=1000)
# 设置标题和消息内容
await page.evaluate(f'setContent("{title}", "{message}")')
await page.evaluate("""([title, message]) => {
return setContent(title, message);
}""",
[title, message])
# 直到 setMaskMode 函数加载完成
await page.wait_for_function("typeof setMaskMode === 'function'", timeout=1000)
await page.evaluate('setMaskMode(true)')
@ -44,33 +53,16 @@ class NoticeUI:
image = Image.open(BytesIO(image_bytes)).convert("RGBA")
mask = Image.open(BytesIO(mask_bytes)).convert("L")
# 使用mask作为alpha通道
r, g, b, _ = image.split()
transparent_image = Image.merge("RGBA", (r, g, b, mask))
# 先创建一个纯白色背景,然后粘贴透明图像
background = Image.new("RGBA", transparent_image.size, (255, 255, 255, 255))
composite = Image.alpha_composite(background, transparent_image)
palette_img = composite.convert("RGB").convert(
"P",
palette=Image.Palette.WEB,
colors=256,
dither=Image.Dither.NONE
)
# 将alpha值小于128的设为透明
alpha_mask = mask.point(lambda x: 0 if x < 128 else 255)
# 遮罩抖动二值化
mask = mask.convert('1') # 先转换为1位图像
image.putalpha(mask)
# 保存为GIF
output_buffer = BytesIO()
palette_img.save(
image.save(
output_buffer,
format="GIF",
transparency=0, # 将索引0设为透明
disposal=2,
loop=0
disposal=2
)
output_buffer.seek(0)

View File

@ -1,4 +1,3 @@
import re
import aiohttp
import asyncio as asynkio
from math import ceil
@ -6,6 +5,7 @@ from pathlib import Path
from typing import Any
import nanoid
from nonebot.rule import KeywordsRule, Rule
from konabot.plugins.notice_ui.notice import NoticeUI
import nonebot
from loguru import logger
@ -15,10 +15,9 @@ from nonebot_plugin_alconna import Alconna, Args, Subcommand, UniMessage, UniMsg
from pydantic import BaseModel
from konabot.common.longtask import DepLongTaskTarget, LongTask, create_longtask, handle_long_task, longtask_data
from konabot.common.nb.match_keyword import match_keyword
from konabot.plugins.simple_notify.ask_llm import ask_ai
evt = on_message(rule=match_keyword(re.compile("^.+提醒我.+$")))
evt = on_message(rule=Rule(KeywordsRule("提醒我")))
(Path(__file__).parent.parent.parent.parent / "data").mkdir(exist_ok=True)
DATA_FILE_PATH = Path(__file__).parent.parent.parent.parent / "data" / "notify.json"
@ -96,7 +95,7 @@ async def _(msg: UniMsg, mEvt: Event, target: DepLongTaskTarget):
await target.send_message(
UniMessage().text(f"了解啦!将会在 {target_time.strftime(FMT_STRING)} 提醒你哦~")
)
logger.info(f"创建了一条于 {target_time}办提醒")
logger.info(f"创建了一条于 {target_time}办提醒")
driver = nonebot.get_driver()
@ -106,9 +105,9 @@ driver = nonebot.get_driver()
async def _(task: LongTask):
message = task.data["message"]
await task.target.send_message(
UniMessage().text(f"办提醒:{message}")
UniMessage().text(f"办提醒:{message}")
)
notice_bytes = await NoticeUI.render_notice("办提醒", message)
notice_bytes = await NoticeUI.render_notice("办提醒", message)
await task.target.send_message(
UniMessage().image(raw=notice_bytes),
at=False
@ -124,7 +123,7 @@ USER_CHECKOUT_TASK_CACHE: dict[str, dict[str, str]] = {}
cmd_check_notify_list = on_alconna(Alconna(
"re:(?:我有哪些|查询)(?:提醒|办)",
"re:(?:我有哪些|查询)(?:提醒|办)",
Args["page", int, 1]
))
@ -142,7 +141,7 @@ async def _(page: int, target: DepLongTaskTarget):
await target.send_message(UniMessage().text(f"最多也就 {pages} 页啦!"))
tasks = tasks[(page - 1) * PAGE_SIZE: page * PAGE_SIZE]
message = "你可以输入「删除提醒 序号」来删除一个提醒\n====== 办清单 ======\n\n"
message = "你可以输入「删除提醒 序号」来删除一个提醒\n====== 办清单 ======\n\n"
to_cache = {}
if len(tasks) == 0:
@ -159,7 +158,7 @@ async def _(page: int, target: DepLongTaskTarget):
cmd_remove_task = on_alconna(Alconna(
"re:删除(?:提醒|办)",
"re:删除(?:提醒|办)",
Args["checker", str],
))
@ -236,7 +235,7 @@ async def _(target: DepLongTaskTarget, notify_id: str = ""):
))
await send_notify_to_ntfy_instance(
"如果你看到这条消息,说明你已经成功订阅主题!此方 BOT 将会在这里提醒你你的办!",
"如果你看到这条消息,说明你已经成功订阅主题!此方 BOT 将会在这里提醒你你的办!",
channel_name,
)

View File

@ -5,7 +5,13 @@ def main():
with playwright.sync_api.sync_playwright() as p:
browser = p.chromium.launch()
page = browser.new_page()
page.goto("https://www.baidu.com")
content = page.content()
if "<html" in content or len(content) < 100:
print("✅ Playwright + Chromium 环境正常 (在 about:blank 页面上测试成功)")
else:
print("⚠️ Playwright + Chromium 环境启动,但页面内容检查结果异常。")
print(content)
raise Exception()
print("Playwright + Chromium 环境正常")
browser.close()