add: nonebot-plugin-gotify

This commit is contained in:
远野千束 2024-10-20 03:05:15 +08:00
parent 20ad8dc53f
commit d5ccd105a2
7 changed files with 168 additions and 1 deletions

View File

@ -5,4 +5,5 @@ nonebot:
nickname: [ "liteyuki" ]
default_language: zh
driver: ~fastapi+~httpx+~websockets
alconna_use_command_start: true
alconna_use_command_start: true
gotify_token: "empty token"

View File

@ -0,0 +1,109 @@
import asyncio
import aiohttp
from nonebot import logger, on_message, Bot
from nonebot.internal.adapter import Event
from nonebot.plugin import PluginMetadata
from nonebot import get_driver
from .adapter_ctx import Context
from .config import plugin_config, Config, NOTICE, MESSAGE
from .gotify import Message, msg_chan, fetch_msg
__plugin_meta__ = PluginMetadata(
name="Gotify",
description="使用Gotify API发送消息",
usage="后台服务插件,无需用户交互",
)
# on plugin load
driver = get_driver()
async def push_thread():
async with aiohttp.ClientSession() as session:
while True:
msg = await fetch_msg()
try:
logger.debug(f"Pushing message: {msg}")
async with session.post(
url=plugin_config.gotify_url + "/message",
params={"token": plugin_config.gotify_token},
data={
"title": msg.title,
"message": msg.message,
"priority": msg.priority,
},
) as resp:
if resp.status != 200:
logger.error(
f"Push message to server failed: {await resp.text()}"
)
else:
logger.info(f"Push message to server success: {msg}")
except Exception as e:
logger.error(f"Push message to server failed: {e}")
@driver.on_startup
async def start_push_thread():
asyncio.create_task(push_thread())
logger.info(
f"Gotify plugin loaded, server: {plugin_config.gotify_url}, token: {plugin_config.gotify_token}"
)
if MESSAGE in plugin_config.gotify_includes:
@on_message().handle()
async def _(event: Event):
ctx = Context(
user_id=event.get_user_id(),
nickname="",
message=event.get_plaintext(),
message_type=event.get_type(),
)
ctx.handle(event)
msg_chan << Message(
title=plugin_config.gotify_title.format(**ctx.model_dump()),
message=plugin_config.gotify_message.format(**ctx.model_dump()),
)
if NOTICE in plugin_config.gotify_includes:
@driver.on_startup
async def startup():
if NOTICE in plugin_config.gotify_includes:
msg_chan << Message(
title=plugin_config.gotify_nickname,
message="Bot started",
priority=plugin_config.gotify_priority,
)
@driver.on_shutdown
async def shutdown():
if NOTICE in plugin_config.gotify_includes:
msg_chan << Message(
title=plugin_config.gotify_nickname,
message="Bot stopped",
priority=plugin_config.gotify_priority,
)
@driver.on_bot_connect
async def bot_connect(bot: Bot):
if NOTICE in plugin_config.gotify_includes:
msg_chan << Message(
title=plugin_config.gotify_nickname,
message=f"Bot connected: {bot.self_id}",
priority=plugin_config.gotify_priority,
)
@driver.on_bot_disconnect
async def bot_disconnect(bot: Bot):
if NOTICE in plugin_config.gotify_includes:
msg_chan << Message(
title=plugin_config.gotify_nickname,
message=f"Bot disconnected: {bot.self_id}",
priority=plugin_config.gotify_priority,
)

View File

@ -0,0 +1,12 @@
from pydantic import BaseModel, field_validator
class Context(BaseModel):
user_id: str
nickname: str
message: str
message_type: str | None = None
def handle(self, event):
pass

View File

@ -0,0 +1,22 @@
from nonebot import get_plugin_config
from pydantic import BaseModel, field_validator
NOTICE = "notice"
MESSAGE = "message"
class Config(BaseModel):
# required fields
gotify_token: str
# optional fields
gotify_url: str = "http://127.0.0.1:40266"
gotify_priority: int = 5
gotify_nickname: str = "NoneBot"
gotify_title: str = "{message_type}: {nickname}({user_id})"
gotify_message: str = "{message}"
gotify_includes: list[str, ...] = [NOTICE, MESSAGE]
plugin_config = get_plugin_config(Config)
if plugin_config.gotify_url.endswith("/"):
plugin_config.gotify_url = plugin_config.gotify_url[:-1]

View File

@ -0,0 +1,23 @@
import asyncio
from asyncio import Future
from collections.abc import Coroutine
from typing import Any
import aiohttp
from magicoca import Chan
from nonebot import logger
from pydantic import BaseModel
from .config import plugin_config
msg_chan = Chan["Message"]()
class Message(BaseModel):
title: str
message: str
priority: int = plugin_config.gotify_priority
def fetch_msg() -> Future[Any]:
return asyncio.get_event_loop().run_in_executor(func=msg_chan.recv, executor=None)