[^;；\s]+.*)",
- flags=re.S,
- block=True,
- priority=13,
-)
+compiler = on_command("lang", force_whitespace=True, block=True, priority=13)
@compiler.handle()
-async def _(msg: dict = RegexDict()):
- language = str(msg["language"]).strip()
- code = unescape(unescape(str(msg["code"]))).strip()
+async def _(matcher: Matcher, msg: Message = CommandArg()):
+ text = msg.extract_plain_text().lstrip()
+ if not text:
+ await matcher.finish()
+ matched = re.match(r"^(?P[^;；\s]+)[;；\s]+(?P[^;；\s]+.*)", text)
+ if not matched:
+ await matcher.finish()
+
+ args = matched.groupdict()
+ language = str(args["language"]).strip()
+ code = str(args["code"]).strip()
if language not in legal_language:
- await compiler.finish(f"支持的语言：{', '.join(list(legal_language.keys()))}")
+ await matcher.finish(f"支持的语言：{language}")
result = await network_compile(language, code)
if not result:
- await compiler.finish("出错了，请稍后再试")
+ await matcher.finish("出错了，请稍后再试")
else:
- await compiler.finish(
- result["output"] if result["output"] else result["errors"]
- )
+ await matcher.finish(result["output"] if result["output"] else result["errors"])
diff --git a/src/plugins/compiler/data_source.py b/src/plugins/compiler/data_source.py
index b1b249b..88cf7ae 100644
--- a/src/plugins/compiler/data_source.py
+++ b/src/plugins/compiler/data_source.py
@@ -1,7 +1,6 @@
import httpx
from nonebot.log import logger
-
legal_language = {
"R": 80,
"vb": 84,
@@ -38,7 +37,11 @@ async def network_compile(language: str, code: str):
"fileext": language,
}
headers = {
- "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.141 Safari/537.36"
+ "User-Agent": (
+ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+ "AppleWebKit/537.36 (KHTML, like Gecko) "
+ "Chrome/87.0.4280.141 Safari/537.36"
+ )
}
try:
async with httpx.AsyncClient() as client:
diff --git a/src/plugins/essay/__init__.py b/src/plugins/essay/__init__.py
index 8a7aff6..7e191e1 100644
--- a/src/plugins/essay/__init__.py
+++ b/src/plugins/essay/__init__.py
@@ -1,11 +1,12 @@
import shlex
from typing import Type
+
from nonebot import on_command
+from nonebot.adapters import Message
from nonebot.matcher import Matcher
-from nonebot.typing import T_Handler
from nonebot.params import CommandArg
from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import Message
+from nonebot.typing import T_Handler
from .data_source import commands, get_essay
diff --git a/src/plugins/essay/data_source.py b/src/plugins/essay/data_source.py
index c957a52..f2cb70f 100644
--- a/src/plugins/essay/data_source.py
+++ b/src/plugins/essay/data_source.py
@@ -1,10 +1,9 @@
-import re
import json
-import httpx
import random
import traceback
-from typing import List
from pathlib import Path
+from typing import List
+
from nonebot.log import logger
dir_path = Path(__file__).parent
@@ -47,7 +46,9 @@ async def get_chicken_soup() -> str:
"ussrjoke": {
"aliases": {"苏联笑话"},
"func": get_ussrjoke,
- "help": "苏联笑话 {要讽刺的事} {谁提出来的} {有助于什么} {针对的是谁} {起作用范围}",
+ "help": (
+ "苏联笑话 {要讽刺的事} {谁提出来的} {有助于什么} {针对的是谁} {起作用范围}"
+ ),
"arg_num": 5,
},
"cpstory": {
@@ -75,6 +76,6 @@ async def get_essay(type: str, texts: List[str]) -> str:
try:
func = commands[type]["func"]
return await func(*texts)
- except:
+ except Exception:
logger.warning(traceback.format_exc())
return "出错了，请稍后再试"
diff --git a/src/plugins/fortune/__init__.py b/src/plugins/fortune/__init__.py
index 453bb19..caaac60 100644
--- a/src/plugins/fortune/__init__.py
+++ b/src/plugins/fortune/__init__.py
@@ -1,13 +1,21 @@
import traceback
-from nonebot import on_command
-from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import MessageEvent, MessageSegment
+
+from nonebot import on_command, require
+from nonebot.adapters import Event
from nonebot.log import logger
+from nonebot.matcher import Matcher
+from nonebot.plugin import PluginMetadata
+
+require("nonebot_plugin_alconna")
+require("nonebot_plugin_userinfo")
+require("nonebot_plugin_htmlrender")
+
+from nonebot_plugin_alconna import UniMessage
+from nonebot_plugin_userinfo import EventUserInfo, UserInfo
from .config import Config
from .data_source import get_fortune
-
__plugin_meta__ = PluginMetadata(
name="运势",
description="测试你的今日运势",
@@ -20,17 +28,17 @@
@jrrp.handle()
-async def _(event: MessageEvent):
- user_id = event.user_id
- username = event.sender.card or event.sender.nickname or ""
+async def _(matcher: Matcher, event: Event, userinfo: UserInfo = EventUserInfo()):
+ user_id = event.get_user_id()
+ username = userinfo.user_displayname or userinfo.user_name
res = None
try:
res = await get_fortune(user_id, username)
- except:
+ except Exception:
logger.warning(traceback.format_exc())
if not res:
- await jrrp.finish("出错了，请稍后再试")
+ await matcher.finish("出错了，请稍后再试")
- await jrrp.finish(MessageSegment.image(res))
+ await UniMessage.image(raw=res).send()
diff --git a/src/plugins/fortune/config.py b/src/plugins/fortune/config.py
index fab8818..bbb2b48 100644
--- a/src/plugins/fortune/config.py
+++ b/src/plugins/fortune/config.py
@@ -1,5 +1,9 @@
-from pydantic import BaseModel, Extra
+from nonebot import get_plugin_config
+from pydantic import BaseModel
-class Config(BaseModel, extra=Extra.ignore):
+class Config(BaseModel):
fortune_style: str = "summer"
+
+
+fortune_config = get_plugin_config(Config)
diff --git a/src/plugins/fortune/data_source.py b/src/plugins/fortune/data_source.py
index e5b791e..bc0889f 100644
--- a/src/plugins/fortune/data_source.py
+++ b/src/plugins/fortune/data_source.py
@@ -1,16 +1,13 @@
import json
-import jinja2
import random
+from datetime import datetime
from pathlib import Path
from typing import Optional
-from datetime import datetime
-from nonebot import get_driver
+import jinja2
from nonebot_plugin_htmlrender import html_to_pic
-from .config import Config
-
-fortune_config = Config.parse_obj(get_driver().config.dict())
+from .config import fortune_config
dir_path = Path(__file__).parent
resource_path = dir_path / "resources"
@@ -20,7 +17,7 @@
)
-async def get_fortune(user_id: int, username: str) -> Optional[bytes]:
+async def get_fortune(user_id: str, username: str) -> Optional[bytes]:
with (resource_path / "copywriting.json").open("r", encoding="utf-8") as f:
data = json.load(f)
date = datetime.now().strftime("%Y%m%d")
@@ -30,7 +27,9 @@ async def get_fortune(user_id: int, username: str) -> Optional[bytes]:
# fmt: off
rank = random.choices(
[27, 26, 25, 24, 23, 21, 20, 10, 9, 8, 7, 6, 5, 4, -6, -7, -8, -9, -10],
- weights=[18, 18, 18, 18, 18, 18, 18, 38, 46, 55, 46, 38, 31, 25, 20, 16, 13, 11, 10],
+ weights=[
+ 18, 18, 18, 18, 18, 18, 18, 38, 46, 55, 46, 38, 31, 25, 20, 16, 13, 11, 10
+ ],
)[0]
# fmt: on
diff --git a/src/plugins/friend_req.py b/src/plugins/friend_req.py
deleted file mode 100644
index ff094e7..0000000
--- a/src/plugins/friend_req.py
+++ /dev/null
@@ -1,23 +0,0 @@
-import asyncio
-from nonebot import on_request
-from nonebot.adapters.onebot.v11 import Bot
-from nonebot.adapters.onebot.v11.event import FriendRequestEvent
-from nonebot.log import logger
-
-
-async def _friend_req(event: FriendRequestEvent) -> bool:
- logger.info(f"FriendRequestEvent: {event.user_id}")
- return True
-
-
-friend_req = on_request(_friend_req, priority=5)
-
-
-@friend_req.handle()
-async def _(bot: Bot, event: FriendRequestEvent):
- # await asyncio.sleep(5)
- # await event.approve(bot)
- await bot.send_private_msg(
- user_id=int(list(bot.config.superusers)[0]),
- message=f"{event.user_id} 请求添加好友",
- )
diff --git a/src/plugins/help/__init__.py b/src/plugins/help/__init__.py
index 8f68966..256bcfd 100644
--- a/src/plugins/help/__init__.py
+++ b/src/plugins/help/__init__.py
@@ -1,62 +1,78 @@
-from typing import Optional, Union
-from nonebot import on_command
+from typing import Annotated, Optional, Union
+
+from nonebot import on_command, require
+from nonebot.adapters import Event, Message
from nonebot.params import CommandArg
-from nonebot.adapters.onebot.v11 import (
- MessageEvent,
- GroupMessageEvent,
- Message,
- MessageSegment,
-)
-from nonebot.plugin import PluginMetadata
+from nonebot.plugin import PluginMetadata, get_loaded_plugins
+
+require("nonebot_plugin_alconna")
+require("nonebot_plugin_session")
+require("nonebot_plugin_htmlrender")
-from .plugin import get_plugins
-from .data_source import get_help_img, get_plugin_img
+from nonebot_plugin_alconna import UniMessage
+from nonebot_plugin_session import SessionId, SessionIdType
+from src.utils.plugin_manager import plugin_manager
+
+from .data_source import get_help_img, get_plugin_img, get_plugin_info
__plugin_meta__ = PluginMetadata(
name="帮助",
description="查看插件帮助",
- usage='@我 发送 "help/帮助" 查看已加载插件\n发送 "help {plugin}" 查看插件详情',
+ usage='@我 发送 "help/帮助" 查看已加载插件\n发送 "help 插件名" 查看插件详情',
extra={
"example": "help help",
},
)
+UserId = Annotated[str, SessionId(SessionIdType.GROUP, include_bot_type=False)]
help = on_command("help", aliases={"帮助", "功能"}, block=True)
@help.handle()
-async def _(event: MessageEvent, msg: Message = CommandArg()):
- plugin_name = msg.extract_plain_text().strip()
+async def _(event: Event, user_id: UserId, msg: Message = CommandArg()):
+ keyword = msg.extract_plain_text().strip()
help_msg = None
- if plugin_name:
- help_msg = await get_help_msg(event, plugin_name)
+ if keyword:
+ help_msg = await get_help_msg(user_id, keyword)
elif event.is_tome():
- help_msg = await get_help_msg(event)
+ help_msg = await get_help_msg(user_id)
if help_msg:
- await help.finish(help_msg)
+ if isinstance(help_msg, str):
+ await help.finish(help_msg)
+ else:
+ await UniMessage.image(raw=help_msg).send()
+
+async def get_help_msg(user_id: str, keyword: str = "") -> Optional[Union[str, bytes]]:
+ def visible_plugin(plugin_name: str) -> bool:
+ if plugin_config := plugin_manager.get_config(plugin_name):
+ return bool(plugin_config.mode & 4)
+ return False
-async def get_help_msg(
- event: MessageEvent, plugin_name: str = ""
-) -> Optional[Union[str, MessageSegment]]:
- plugins = get_plugins(event)
+ plugins = get_loaded_plugins()
+ plugins = list(filter(lambda p: visible_plugin(p.name), plugins))
- if not plugin_name:
+ if not keyword:
if not plugins:
return "暂时没有可用的功能"
- event_type = "group" if isinstance(event, GroupMessageEvent) else "private"
- img = await get_help_img(event_type, plugins)
- return MessageSegment.image(img) if img else "出错了，请稍后再试"
+ plugin_infos = []
+ for plugin in plugins:
+ if info := get_plugin_info(user_id, plugin.name):
+ plugin_infos.append(info)
+ if not plugin_infos:
+ return "暂时没有可用的功能"
+ plugin_infos = sorted(plugin_infos, key=lambda i: i.package_name)
+ img = await get_help_img(plugin_infos)
+ return img if img else "出错了，请稍后再试"
else:
- for p in plugins:
- if plugin_name.lower() in (
- p.package_name.lower(),
- p.name.lower(),
- p.extra.get("unique_name", "_").lower(),
- ):
- img = await get_plugin_img(p)
- return MessageSegment.image(img) if img else "出错了，请稍后再试"
- return None
+ plugin_name = plugin_manager.find(keyword)
+ if (
+ plugin_name
+ and visible_plugin(plugin_name)
+ and (info := get_plugin_info(user_id, plugin_name))
+ ):
+ img = await get_plugin_img(info)
+ return img if img else "出错了，请稍后再试"
diff --git a/src/plugins/help/data_source.py b/src/plugins/help/data_source.py
index 3de1acf..d6cac09 100644
--- a/src/plugins/help/data_source.py
+++ b/src/plugins/help/data_source.py
@@ -1,10 +1,13 @@
-import jinja2
-from typing import List, Optional
+from dataclasses import dataclass, field
from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import jinja2
+from nonebot import get_plugin
from nonebot.log import logger
from nonebot_plugin_htmlrender import html_to_pic
-from .plugin import PluginInfo
+from src.utils.plugin_manager import plugin_manager
dir_path = Path(__file__).parent
template_path = dir_path / "template"
@@ -13,10 +16,37 @@
)
-async def get_help_img(event_type: str, plugins: List[PluginInfo]) -> Optional[bytes]:
+@dataclass
+class PluginInfo:
+ package_name: str
+ name: str = ""
+ description: str = ""
+ usage: str = ""
+ extra: Dict[Any, Any] = field(default_factory=dict)
+ enabled: bool = True
+ locked: bool = False
+
+
+def get_plugin_info(user_id: str, plugin_name: str) -> Optional[PluginInfo]:
+ plugin = get_plugin(plugin_name)
+ if not plugin:
+ return
+ info = PluginInfo(package_name=plugin.name)
+ if metadata := plugin.metadata:
+ info.name = metadata.name
+ info.description = metadata.description
+ info.usage = metadata.usage
+ info.extra.update(metadata.extra)
+ if plugin_config := plugin_manager.get_config(plugin.name):
+ info.enabled = plugin_manager.check(plugin.name, user_id)
+ info.locked = not bool(plugin_config.mode & 2)
+ return info
+
+
+async def get_help_img(plugins: List[PluginInfo]) -> Optional[bytes]:
try:
template = env.get_template("help.html")
- content = await template.render_async(type=event_type, plugins=plugins)
+ content = await template.render_async(plugins=plugins)
return await html_to_pic(
content,
wait=0,
diff --git a/src/plugins/help/plugin.py b/src/plugins/help/plugin.py
deleted file mode 100644
index ed9bd2d..0000000
--- a/src/plugins/help/plugin.py
+++ /dev/null
@@ -1,119 +0,0 @@
-import json
-from pathlib import Path
-from typing import List, Dict, Any
-from pydantic import BaseModel, ValidationError
-
-from nonebot import get_driver, require
-from nonebot.plugin import Plugin, get_loaded_plugins
-from nonebot.adapters.onebot.v11 import MessageEvent, GroupMessageEvent
-
-require("nonebot_plugin_manager")
-from nonebot_plugin_manager.manager import plugin_manager
-
-
-info_path = Path("data/plugin_info.json")
-
-
-class PluginInfo(BaseModel):
- package_name: str
- name: str = ""
- description: str = ""
- usage: str = ""
- extra: Dict[Any, Any] = {}
- enabled: bool = True
- locked: bool = False
-
- def __eq__(self, other: "PluginInfo"):
- return self.package_name == other.package_name
-
- def __hash__(self):
- return hash(self.package_name)
-
-
-def load_plugin_info() -> List[PluginInfo]:
- if not info_path.exists():
- return []
- infos = []
- with info_path.open("r", encoding="utf8") as fp:
- data = json.load(fp)
- for p in data["plugins"]:
- try:
- infos.append(PluginInfo.parse_obj(p))
- except ValidationError:
- pass
- return sorted(infos, key=lambda i: i.package_name)
-
-
-def dump_plugin_info(infos: List[PluginInfo]):
- info_path.parent.mkdir(parents=True, exist_ok=True)
- infos_dict = {
- "plugins": [info.dict(exclude={"enabled", "locked"}) for info in infos]
- }
- with info_path.open("w", encoding="utf8") as fp:
- json.dump(
- infos_dict,
- fp,
- indent=4,
- ensure_ascii=False,
- )
-
-
-driver = get_driver()
-
-
-@driver.on_startup
-def update_plugin_info():
- plugins = get_loaded_plugins()
- infos: List[PluginInfo] = []
- for p in plugins:
- info = PluginInfo(package_name=p.name)
- info.extra["unique_name"] = (
- p.name.replace("nonebot_plugin_", "")
- .replace("nonebot-plugin-", "")
- .replace("nonebot_", "")
- .replace("nonebot-", "")
- )
- if metadata := p.metadata:
- info.name = metadata.name
- info.description = metadata.description
- info.usage = metadata.usage
- info.extra.update(metadata.extra)
- else:
- info.name = get_plugin_attr(p, "__help__plugin_name__") or p.name
- info.description = get_plugin_attr(p, "__des__")
- info.usage = get_plugin_attr(p, "__usage__")
- info.extra["example"] = get_plugin_attr(p, "__example__")
- info.extra["notice"] = get_plugin_attr(p, "__notice__")
- if info.name and info.usage and info.description:
- infos.append(info)
- infos_set = set(infos)
- for info in load_plugin_info():
- infos_set.add(info)
- infos = list(infos_set)
- infos = sorted(infos, key=lambda i: i.package_name)
- dump_plugin_info(infos)
-
-
-def get_plugin_attr(plugin: Plugin, attr: str):
- try:
- return plugin.module.__getattribute__(attr)
- except:
- return ""
-
-
-def get_plugins(event: MessageEvent) -> List[PluginInfo]:
- infos = load_plugin_info()
- conv = {
- "user": [event.user_id],
- "group": [event.group_id] if isinstance(event, GroupMessageEvent) else [],
- }
- plugins_read = plugin_manager.get_plugin(conv, 4)
- plugins_write = plugin_manager.get_plugin(conv, 6)
- plugins_exec = plugin_manager.get_plugin(conv, 1)
- plugins: List[PluginInfo] = []
- for info in infos:
- if plugins_read.get(info.package_name, False):
- info.enabled = plugins_exec.get(info.package_name, False)
- info.locked = not plugins_write.get(info.package_name, False)
- plugins.append(info)
- return plugins
diff --git a/src/plugins/manager/__init__.py b/src/plugins/manager/__init__.py
index ba14047..2d256fc 100644
--- a/src/plugins/manager/__init__.py
+++ b/src/plugins/manager/__init__.py
@@ -1,116 +1,184 @@
-from typing import Dict, List
+from typing import Annotated
-from nonebot.rule import Rule
-from nonebot.params import CommandArg
from nonebot import on_command, require
+from nonebot.adapters import Bot, Event, Message
+from nonebot.exception import IgnoredException
+from nonebot.log import logger
+from nonebot.matcher import Matcher
+from nonebot.message import run_preprocessor
+from nonebot.params import CommandArg, Depends
+from nonebot.permission import SUPERUSER, Permission
from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import Bot, MessageEvent, GroupMessageEvent, Message
-require("nonebot_plugin_manager")
-from nonebot_plugin_manager.manager import plugin_manager
+require("nonebot_plugin_session")
-from ..help.plugin import get_plugins
+from nonebot_plugin_session import EventSession, SessionId, SessionIdType, SessionLevel
+
+from src.utils.plugin_manager import ManageType, plugin_manager
__plugin_meta__ = PluginMetadata(
name="插件管理",
description="启用/禁用插件",
- usage="启用插件/插件 插件名",
+ usage="启用插件/禁用插件 插件名",
extra={
"example": "禁用插件 setu",
- "notice": "仅群管理员或超级用户可用",
+ "notice": "仅管理员或超级用户可用",
},
)
-def manager_rule(bot: Bot, event: MessageEvent) -> bool:
- return isinstance(event, GroupMessageEvent) and (
- str(event.user_id) in bot.config.superusers
- or event.sender.role in ["admin", "owner"]
- )
+def _is_private(session: EventSession) -> bool:
+ return session.level == SessionLevel.LEVEL1
+
+
+PERM_EDIT = SUPERUSER | Permission(_is_private)
+PERM_GLOBAL = SUPERUSER
+
+
+try:
+ from nonebot.adapters.onebot.v11.permission import GROUP_ADMIN, GROUP_OWNER
+
+ PERM_EDIT |= GROUP_ADMIN | GROUP_OWNER
+except ImportError:
+ pass
+
+
+block = on_command("禁用插件", block=True, priority=11, permission=PERM_EDIT)
+unblock = on_command("启用插件", block=True, priority=11, permission=PERM_EDIT)
+block_gl = on_command("全局禁用插件", block=True, priority=11, permission=PERM_GLOBAL)
+unblock_gl = on_command("全局启用插件", block=True, priority=11, permission=PERM_GLOBAL)
+set_mode = on_command("设置插件模式", block=True, priority=11, permission=PERM_GLOBAL)
+
+
+UserId = Annotated[str, SessionId(SessionIdType.GROUP, include_bot_type=False)]
-block = on_command("禁用插件", block=True, rule=Rule(manager_rule))
-unblock = on_command("启用插件", block=True, rule=Rule(manager_rule))
+@run_preprocessor
+async def _(matcher: Matcher, user_id: UserId):
+ plugin = matcher.plugin_name
-Conv = Dict[str, List[int]]
+ if plugin and not plugin_manager.check(plugin, user_id):
+ reason = f"Manager has blocked '{plugin}'"
+ logger.debug(reason)
+ raise IgnoredException(reason)
-def get_conv(event: MessageEvent) -> Conv:
- return {
- "user": [event.user_id],
- "group": [event.group_id] if isinstance(event, GroupMessageEvent) else [],
- }
+def IsSuperUser():
+ async def dependency(bot: Bot, event: Event):
+ return event.get_user_id() in bot.config.superusers
+
+ return Depends(dependency)
@block.handle()
-async def _(bot: Bot, event: MessageEvent, msg: Message = CommandArg()):
+async def _(
+ matcher: Matcher,
+ user_id: UserId,
+ msg: Message = CommandArg(),
+ is_superuser: bool = IsSuperUser(),
+):
keyword = msg.extract_plain_text().strip()
if not keyword:
return
- plugins = get_plugins(event)
- plugin = None
- for p in plugins[::-1]:
- if keyword.lower() in (
- p.package_name.lower(),
- p.name.lower(),
- p.extra.get("unique_name", "_").lower(),
- ):
- plugin = p
- break
- if not plugin:
- await block.finish(f"插件 {keyword} 不存在！")
-
- conv: Conv = get_conv(event)
- if str(event.user_id) in bot.config.superusers:
- plugins_write = plugin_manager.get_plugin(perm=6)
- else:
- plugins_write = plugin_manager.get_plugin(conv, 6)
- if not plugins_write.get(plugin.package_name, False):
- await block.finish(f"插件 {plugin.name} 已关闭编辑权限！")
-
- if conv["group"]:
- conv["user"] = []
- result = plugin_manager.block_plugin([plugin.package_name], conv)
- if result.get(plugin.package_name, False):
- res = f"插件 {plugin.name} 禁用成功"
+ plugin_name = plugin_manager.find(keyword)
+ if not plugin_name:
+ await matcher.finish(f"插件 {keyword} 不存在！")
+
+ plugin_config = plugin_manager.get_config(plugin_name)
+ if not plugin_config:
+ await matcher.finish(f"插件 {keyword} 不存在！")
+
+ mode = plugin_config.mode
+ if not ((mode & 2) or is_superuser):
+ await matcher.finish(f"插件 {plugin_name} 已关闭编辑权限！")
+
+ if plugin_manager.block(plugin_name, user_id):
+ await matcher.finish(f"插件 {plugin_name} 禁用成功")
else:
- res = f"插件 {plugin.name} 不存在或已关闭编辑权限！"
- await block.finish(res)
+ await matcher.finish(f"插件 {plugin_name} 禁用失败")
@unblock.handle()
-async def _(bot: Bot, event: MessageEvent, msg: Message = CommandArg()):
+async def _(
+ matcher: Matcher,
+ user_id: UserId,
+ msg: Message = CommandArg(),
+ is_superuser: bool = IsSuperUser(),
+):
keyword = msg.extract_plain_text().strip()
if not keyword:
return
- plugins = get_plugins(event)
- plugin = None
- for p in plugins[::-1]:
- if keyword.lower() in (
- p.package_name.lower(),
- p.name.lower(),
- p.extra.get("unique_name", "_").lower(),
- ):
- plugin = p
- break
- if not plugin:
- await unblock.finish(f"插件 {keyword} 不存在！")
-
- conv: Conv = get_conv(event)
- if str(event.user_id) in bot.config.superusers:
- plugins_write = plugin_manager.get_plugin(perm=6)
+ plugin_name = plugin_manager.find(keyword)
+ if not plugin_name:
+ await matcher.finish(f"插件 {keyword} 不存在！")
+
+ plugin_config = plugin_manager.get_config(plugin_name)
+ if not plugin_config:
+ await matcher.finish(f"插件 {keyword} 不存在！")
+
+ mode = plugin_config.mode
+ if not ((mode & 2) or is_superuser):
+ await matcher.finish(f"插件 {plugin_name} 已关闭编辑权限！")
+
+ if plugin_manager.unblock(plugin_name, user_id):
+ await matcher.finish(f"插件 {plugin_name} 启用成功")
else:
- plugins_write = plugin_manager.get_plugin(conv, 6)
- if not plugins_write.get(plugin.package_name, False):
- await unblock.finish(f"插件 {plugin.name} 已关闭编辑权限！")
-
- if conv["group"]:
- conv["user"] = []
- result = plugin_manager.unblock_plugin([plugin.package_name], conv)
- if result.get(plugin.package_name, False):
- res = f"插件 {plugin.name} 启用成功"
+ await matcher.finish(f"插件 {plugin_name} 启用失败")
+
+
+@block_gl.handle()
+async def _(matcher: Matcher, msg: Message = CommandArg()):
+ keyword = msg.extract_plain_text().strip()
+ if not keyword:
+ return
+
+ plugin_name = plugin_manager.find(keyword)
+ if not plugin_name:
+ await matcher.finish(f"插件 {keyword} 不存在！")
+
+ if plugin_manager.change_manage_type(plugin_name, ManageType.WHITE):
+ await matcher.finish(f"插件 {plugin_name} 已设为白名单模式")
+ else:
+ await matcher.finish(f"插件 {plugin_name} 设置失败")
+
+
+@unblock_gl.handle()
+async def _(matcher: Matcher, msg: Message = CommandArg()):
+ keyword = msg.extract_plain_text().strip()
+ if not keyword:
+ return
+
+ plugin_name = plugin_manager.find(keyword)
+ if not plugin_name:
+ await matcher.finish(f"插件 {keyword} 不存在！")
+
+ if plugin_manager.change_manage_type(plugin_name, ManageType.BLACK):
+ await matcher.finish(f"插件 {plugin_name} 已设为黑名单模式")
+ else:
+ await matcher.finish(f"插件 {plugin_name} 设置失败")
+
+
+@set_mode.handle()
+async def _(
+ matcher: Matcher,
+ msg: Message = CommandArg(),
+):
+ message = msg.extract_plain_text().strip()
+ if not message:
+ return
+ args = message.split()
+ if len(args) != 2 or not args[1].isdigit():
+ await matcher.finish("参数错误！")
+ keyword, mode = args
+ mode = int(mode)
+
+ plugin_name = plugin_manager.find(keyword)
+ if not plugin_name:
+ await matcher.finish(f"插件 {keyword} 不存在！")
+
+ if plugin_manager.change_mode(plugin_name, mode):
+ await matcher.finish(f"插件 {plugin_name} 模式已设为 {mode}")
else:
- res = f"插件 {plugin.name} 不存在或已关闭编辑权限！"
- await unblock.finish(res)
+ await matcher.finish(f"插件 {plugin_name} 设置失败")
diff --git a/src/plugins/mc/__init__.py b/src/plugins/mc/__init__.py
deleted file mode 100644
index 4850df3..0000000
--- a/src/plugins/mc/__init__.py
+++ /dev/null
@@ -1,215 +0,0 @@
-from typing import Type
-from argparse import Namespace
-from nonebot.matcher import Matcher
-from nonebot.plugin import PluginMetadata
-from nonebot.rule import Rule, ArgumentParser
-from nonebot import on_command, on_shell_command, on_notice
-from nonebot.params import CommandArg, ShellCommandArgs, Depends
-from nonebot.adapters.onebot.v11 import (
- Event,
- GroupMessageEvent,
- PokeNotifyEvent,
- Message,
- MessageSegment,
-)
-
-from .monitor import *
-from .dynmap_list import *
-from .config import Config
-from .dynmap_source import get_status, send_message
-from .data_source import get_mcstatus, get_mc_uuid, get_crafatar, get_mcmodel
-
-
-__plugin_meta__ = PluginMetadata(
- name="Minecraft",
- description="Minecraft相关功能",
- usage=(
- "1、mcstatus {url}，MC服务器状态查询\n"
- "2、mc avatar/head/body/skin/cape/model {id}，获取MC用户的 头像/头/身体/皮肤/披风/全身动图\n"
- "3、dynmap bind {url}，动态地图相关功能"
- ),
- config=Config,
- extra={
- "example": "mcstatus mczju.tpddns.cn\nmcskin hsds",
- },
-)
-
-
-mc = on_command("mc", block=True, priority=12)
-
-
-@mc.handle()
-async def _(arg: Message = CommandArg()):
- msg = arg.extract_plain_text().strip()
- if not msg:
- await mc.finish()
- if msg.startswith("status"):
- addr = msg.replace("status", "", 1).strip()
- if addr:
- status = await get_mcstatus(addr)
- if status:
- await mc.finish(status)
- else:
- await mc.finish("出错了，请稍后再试")
- else:
- types = ["avatar", "head", "body", "skin", "cape", "model"]
- for t in types:
- if msg.startswith(t):
- username = msg.replace(t, "", 1).strip()
- if username:
- uuid = await get_mc_uuid(username)
- if not uuid:
- await mc.finish("出错了，请稍后再试")
- if uuid == "none":
- await mc.finish("找不到该用户")
- if t == "model":
- await mc.send("生成中，请耐心等待。。。")
- result = await get_mcmodel(uuid)
- else:
- result = await get_crafatar(t, uuid)
- if result:
- await mc.finish(MessageSegment.image(result))
- else:
- await mc.finish("出错了，请稍后再试")
- await mc.finish()
-
-
-async def bind(matcher: Matcher, user_id: str, url: str, **kwargs):
- result = await bind_dynmap(user_id, url)
- if not result:
- await matcher.finish("出错了，请检查链接或稍后再试")
- await matcher.finish(
- "成功绑定动态地图，使用“dynmap chaton/chatoff”开启或关闭聊天转发；使用“dynmap status”查看在线状态"
- )
-
-
-async def unbind(matcher: Matcher, user_id: str, **kwargs):
- if not unbind_dynmap(user_id):
- await matcher.finish("尚未绑定动态地图")
- await matcher.finish("成功解绑动态地图")
-
-
-async def chaton(matcher: Matcher, user_id: str, **kwargs):
- if not open_chat(user_id):
- await matcher.finish("尚未绑定动态地图")
- await matcher.finish("聊天转发已打开")
-
-
-async def chatoff(matcher: Matcher, user_id: str, **kwargs):
- if not close_chat(user_id):
- await matcher.finish("尚未绑定动态地图")
- await matcher.finish("聊天转发已关闭")
-
-
-async def pokeon(matcher: Matcher, user_id: str, **kwargs):
- if not open_poke(user_id):
- await matcher.finish("尚未绑定动态地图")
- await matcher.finish("已打开戳一戳状态显示")
-
-
-async def pokeoff(matcher: Matcher, user_id: str, **kwargs):
- if not close_poke(user_id):
- await matcher.finish("尚未绑定动态地图")
- await matcher.finish("已关闭戳一戳状态显示")
-
-
-async def status(matcher: Type[Matcher], user_id: str, **kwargs):
- url = get_update_url(user_id)
- if not url:
- await matcher.finish("尚未绑定动态地图")
- status = await get_status(url)
- if not status:
- await matcher.finish("出错了，请稍后再试")
- await matcher.finish(status)
-
-
-async def login(matcher: Matcher, user_id: str, username: str, password: str, **kwargs):
- set_user(user_id, username, password)
- await matcher.finish("登录成功")
-
-
-async def logout(matcher: Matcher, user_id: str, **kwargs):
- del_user(user_id)
- await matcher.finish("注销成功")
-
-
-async def send(matcher: Matcher, user_id: str, message: str, **kwargs):
- if not get_login_status(user_id):
- await matcher.finish("尚未登录，使用“dynmap login ”登录")
- if not await send_message(get_dynmap_list()[user_id], message):
- await matcher.finish("发送消息失败")
-
-
-dynmap_parser = ArgumentParser("dynmap")
-
-dynmap_subparsers = dynmap_parser.add_subparsers()
-
-bind_parser = dynmap_subparsers.add_parser("bind", aliases=("绑定"))
-bind_parser.add_argument("url")
-bind_parser.set_defaults(func=bind)
-
-unbind_parser = dynmap_subparsers.add_parser("unbind", aliases=("解绑"))
-unbind_parser.set_defaults(func=unbind)
-
-login_parser = dynmap_subparsers.add_parser("login", aliases=("登录"))
-login_parser.add_argument("username")
-login_parser.add_argument("password")
-login_parser.set_defaults(func=login)
-
-logout_parser = dynmap_subparsers.add_parser("logout", aliases=("退出"))
-logout_parser.set_defaults(func=logout)
-
-send_parser = dynmap_subparsers.add_parser("send", aliases=("发送"))
-send_parser.add_argument("message")
-send_parser.set_defaults(func=send)
-
-status_parser = dynmap_subparsers.add_parser("status", aliases=("状态"))
-status_parser.set_defaults(func=status)
-
-pokeon_parser = dynmap_subparsers.add_parser("pokeon")
-pokeon_parser.set_defaults(func=pokeon)
-
-pokeoff_parser = dynmap_subparsers.add_parser("pokeoff")
-pokeoff_parser.set_defaults(func=pokeoff)
-
-chaton_parser = dynmap_subparsers.add_parser("chaton")
-chaton_parser.set_defaults(func=chaton)
-
-chatoff_parser = dynmap_subparsers.add_parser("chatoff")
-chatoff_parser.set_defaults(func=chatoff)
-
-dynmap = on_shell_command("dynmap", parser=dynmap_parser, block=True, priority=15)
-
-
-def get_id(event: Event):
- if isinstance(event, GroupMessageEvent) or isinstance(event, PokeNotifyEvent):
- return "group_" + str(event.group_id)
- else:
- return "private_" + str(event.get_user_id())
-
-
-@dynmap.handle()
-async def _(args: Namespace = ShellCommandArgs(), user_id: str = Depends(get_id)):
- args.matcher = dynmap
- args.user_id = user_id
-
- if hasattr(args, "func"):
- await args.func(**vars(args))
-
-
-async def _poke_status(event: Event) -> bool:
- if isinstance(event, PokeNotifyEvent) and event.is_tome():
- user_id = get_id(event)
- dynmap_list = get_dynmap_list()
- if user_id in dynmap_list:
- if dynmap_list[user_id]["poke"]:
- return True
- return False
-
-
-poke_status = on_notice(Rule(_poke_status), priority=15, block=False)
-
-
-@poke_status.handle()
-async def _(event: Event, user_id: str = Depends(get_id)):
- await status(poke_status, user_id)
diff --git a/src/plugins/mc/config.py b/src/plugins/mc/config.py
deleted file mode 100644
index b3b2746..0000000
--- a/src/plugins/mc/config.py
+++ /dev/null
@@ -1,5 +0,0 @@
-from pydantic import BaseModel, Extra
-
-
-class Config(BaseModel, extra=Extra.ignore):
- dynmap_cron: list = ["*/2", "*", "*", "*", "*", "*"]
diff --git a/src/plugins/mc/dynmap_list.py b/src/plugins/mc/dynmap_list.py
deleted file mode 100644
index 72bc95e..0000000
--- a/src/plugins/mc/dynmap_list.py
+++ /dev/null
@@ -1,154 +0,0 @@
-import json
-import httpx
-import traceback
-from typing import Dict
-from pathlib import Path
-from datetime import datetime
-from nonebot.log import logger
-
-from .dynmap_source import get_status
-
-data_path = Path("data/mc")
-if not data_path.exists():
- data_path.mkdir(parents=True)
-dynmap_path = data_path / "dynmap_list.json"
-
-
-def _load_dynmap_list() -> Dict[str, dict]:
- try:
- return json.load(dynmap_path.open("r", encoding="utf-8"))
- except FileNotFoundError:
- return {}
-
-
-_dynmap_list = _load_dynmap_list()
-
-
-def dump_dynmap_list():
- json.dump(
- _dynmap_list,
- dynmap_path.open("w", encoding="utf-8"),
- indent=4,
- separators=(",", ": "),
- ensure_ascii=False,
- )
-
-
-def get_dynmap_list():
- return _dynmap_list.copy()
-
-
-def get_dynmap_url(user_id: str) -> str:
- if user_id in _dynmap_list:
- return _dynmap_list[user_id].get("url", "")
- return ""
-
-
-def get_update_url(user_id: str) -> str:
- if user_id in _dynmap_list:
- return _dynmap_list[user_id].get("update_url", "")
- return ""
-
-
-def get_poke_status(user_id: str) -> bool:
- if user_id in _dynmap_list:
- return _dynmap_list[user_id].get("poke", False)
- return False
-
-
-def get_login_status(user_id: str) -> bool:
- if user_id in _dynmap_list:
- return _dynmap_list[user_id].get("username", "") and _dynmap_list[user_id].get(
- "password", ""
- )
- return False
-
-
-async def fetch_update_url(url: str) -> str:
- url_config = url + "/up/configuration"
- try:
- async with httpx.AsyncClient() as client:
- resp = await client.get(url_config)
- result = resp.json()
- world = result["defaultworld"]
- return f"{url}/up/world/{world}"
- except:
- logger.debug(traceback.format_exc())
- return ""
-
-
-async def bind_dynmap(user_id: str, url: str) -> bool:
- url = url.strip("/#")
- update_url = await fetch_update_url(url)
- if not update_url:
- return False
- status = await get_status(update_url)
- if not status:
- return False
- _dynmap_list[user_id] = {
- "url": url,
- "update_url": update_url,
- "chat": False,
- "poke": False,
- "last_update": int(datetime.now().timestamp() * 1000),
- }
- dump_dynmap_list()
- return True
-
-
-def unbind_dynmap(user_id: str) -> bool:
- if not get_dynmap_url(user_id):
- return False
- _dynmap_list.pop(user_id)
- dump_dynmap_list()
- return True
-
-
-def open_chat(user_id: str) -> bool:
- if not get_dynmap_url(user_id):
- return False
- _dynmap_list[user_id]["chat"] = True
- dump_dynmap_list()
- return True
-
-
-def close_chat(user_id: str) -> bool:
- if not get_dynmap_url(user_id):
- return False
- _dynmap_list[user_id]["chat"] = False
- dump_dynmap_list()
- return True
-
-
-def open_poke(user_id: str) -> bool:
- if not get_dynmap_url(user_id):
- return False
- _dynmap_list[user_id]["poke"] = True
- dump_dynmap_list()
- return True
-
-
-def close_poke(user_id: str) -> bool:
- if not get_dynmap_url(user_id):
- return False
- _dynmap_list[user_id]["poke"] = False
- dump_dynmap_list()
- return True
-
-
-def set_user(user_id: str, username: str, password: str) -> bool:
- if not get_dynmap_url(user_id):
- return False
- _dynmap_list[user_id]["username"] = username
- _dynmap_list[user_id]["password"] = password
- dump_dynmap_list()
- return True
-
-
-def del_user(user_id: str) -> bool:
- if not get_dynmap_url(user_id):
- return False
- _dynmap_list[user_id].pop("username")
- _dynmap_list[user_id].pop("password")
- dump_dynmap_list()
- return True
diff --git a/src/plugins/mc/dynmap_source.py b/src/plugins/mc/dynmap_source.py
deleted file mode 100644
index b2b0ad8..0000000
--- a/src/plugins/mc/dynmap_source.py
+++ /dev/null
@@ -1,52 +0,0 @@
-import httpx
-from nonebot.log import logger
-from datetime import datetime, timedelta
-
-
-async def get_dynmap_updates(url: str):
- try:
- stamp = (datetime.now() - timedelta(minutes=2)).timestamp()
- url += "/" + str(int(stamp * 1000))
- async with httpx.AsyncClient(timeout=20) as client:
- resp = await client.get(url)
- result = resp.json()
- return result
- except Exception as e:
- logger.warning(f"Error in get_dynmap_updates({url}): {e}")
- return None
-
-
-async def get_status(url: str) -> str:
- result = await get_dynmap_updates(url)
- if not result:
- return ""
- players = result["players"]
- players = [p["account"] for p in players]
- players = ", ".join(players)
- tick = result["servertime"]
- tick = tick - 18000
- tick = tick + 24000 if tick < 0 else tick
- tick = tick * 36
- stime = "{:02d}:{:02d}:{:02d}.{:01d}".format(
- tick // 10 // 60 // 60, tick // 10 // 60 % 60, tick // 10 % 60, tick % 10
- )
- storm = result["hasStorm"]
- thunder = result["isThundering"]
- weather = "☀" if not storm else "⛈" if thunder else "🌧"
- status = f"当前在线：{players}\n服务器时间：{stime}\n服务器天气：{weather}"
- return status
-
-
-async def send_message(config, msg):
- try:
- login_url = config["url"] + "/up/login"
- send_url = config["url"] + "/up/sendmessage"
- info = {"j_username": config["username"], "j_password": config["password"]}
- data = {"name": "", "message": msg}
- async with httpx.AsyncClient(timeout=20) as client:
- await client.post(login_url, data=info)
- await client.post(send_url, json=data)
- return True
- except Exception as e:
- logger.warning(f"Error in send_message({config}, {msg}): {e}")
- return False
diff --git a/src/plugins/mc/monitor.py b/src/plugins/mc/monitor.py
deleted file mode 100644
index 8f3ae14..0000000
--- a/src/plugins/mc/monitor.py
+++ /dev/null
@@ -1,114 +0,0 @@
-import re
-from lxml import etree
-from typing import List, Dict
-from dataclasses import dataclass, field
-from nonebot import get_bot, get_driver
-from nonebot.adapters.onebot.v11 import Bot
-from nonebot_plugin_apscheduler import scheduler
-
-from .dynmap_source import get_dynmap_updates
-from .dynmap_list import get_dynmap_list
-
-from .config import Config
-
-mc_config = Config.parse_obj(get_driver().config.dict())
-
-
-@dataclass
-class Recorder:
- count: int = 0
- last_time: int = 0
- chats: List[dict] = field(default_factory=list)
-
-
-chat_recorders: Dict[str, Recorder] = {}
-
-
-def user_type(user_id: str):
- p_group = r"group_(\d+)"
- p_private = r"private_(\d+)"
- match = re.fullmatch(p_group, user_id)
- if match:
- return "group", match.group(1)
- match = re.fullmatch(p_private, user_id)
- if match:
- return "private", match.group(1)
- return "", user_id
-
-
-async def send_bot_msg(user_id: str, msg):
- type, id = user_type(user_id)
- bot = get_bot()
- assert isinstance(bot, Bot)
- if type == "group":
- await bot.send_group_msg(group_id=int(id), message=msg)
- elif type == "private":
- await bot.send_private_msg(user_id=int(id), message=msg)
-
-
-async def update_dynmap():
- dynmap_list = get_dynmap_list()
- for user_id, config in dynmap_list.items():
- if not config["chat"]:
- continue
-
- url = config["update_url"]
- result = await get_dynmap_updates(url)
- if not result:
- continue
-
- updates = result["updates"]
- if not updates:
- continue
-
- if user_id not in chat_recorders:
- chat_recorders[user_id] = Recorder(
- count=1, last_time=updates[-1]["timestamp"]
- )
- continue
- recorder = chat_recorders[user_id]
- recorder.count += 1
-
- chats = []
- for update in updates:
- if update["type"] == "chat" and update["timestamp"] > recorder.last_time:
- chats.append(update)
- recorder.last_time = updates[-1]["timestamp"]
-
- if not chats:
- continue
- recorder.chats.extend(chats)
-
-
-async def dynmap_monitor():
- await update_dynmap()
- for user_id, recorder in chat_recorders.items():
- if recorder.count < 6:
- continue
- else:
- recorder.count = 0
- msgs = []
- for chat in recorder.chats:
- name = chat["playerName"]
- name = etree.HTML(name, etree.HTMLParser()).xpath("string(.)").strip()
- message = chat["message"]
- msgs.append(f"[dynmap] {name}: {message}")
- recorder.chats = []
- if msgs:
- msg = "\n".join(msgs)
- await send_bot_msg(user_id, msg)
-
-
-dynmap_cron = mc_config.dynmap_cron
-scheduler.add_job(
- dynmap_monitor,
- "cron",
- second=dynmap_cron[0],
- minute=dynmap_cron[1],
- hour=dynmap_cron[2],
- day=dynmap_cron[3],
- month=dynmap_cron[4],
- year=dynmap_cron[5],
- id="dynmap_monitor",
- coalesce=True,
-)
diff --git a/src/plugins/mc_tools/__init__.py b/src/plugins/mc_tools/__init__.py
new file mode 100644
index 0000000..80e46d5
--- /dev/null
+++ b/src/plugins/mc_tools/__init__.py
@@ -0,0 +1,69 @@
+from nonebot import on_command, require
+from nonebot.adapters import Message
+from nonebot.matcher import Matcher
+from nonebot.params import CommandArg
+from nonebot.plugin import PluginMetadata
+
+require("nonebot_plugin_alconna")
+require("nonebot_plugin_htmlrender")
+
+from nonebot_plugin_alconna import Image, UniMessage
+
+from .data_source import get_crafatar, get_mc_uuid, get_mcmodel, get_mcstatus
+
+__plugin_meta__ = PluginMetadata(
+ name="MC工具",
+ description="Minecraft相关功能",
+ usage=(
+ "1、mcstatus {url}，MC服务器状态查询\n"
+ "2、mc avatar/head/body/skin/cape/model {id}，"
+ "获取MC用户的 头像/头/身体/皮肤/披风/全身动图\n"
+ ),
+ extra={
+ "example": "mcstatus mczju.tpddns.cn\nmcskin hsds",
+ },
+)
+
+
+mc = on_command("mc", block=True, priority=12)
+
+
+@mc.handle()
+async def _(matcher: Matcher, arg: Message = CommandArg()):
+ msg = arg.extract_plain_text().strip()
+ if not msg:
+ await matcher.finish()
+ if msg.startswith("status"):
+ addr = msg.replace("status", "", 1).strip()
+ if addr:
+ result = await get_mcstatus(addr)
+ if result:
+ icon, status = result
+ message = UniMessage()
+ if icon:
+ message += Image(raw=icon)
+ message += status
+ await message.send()
+ else:
+ await matcher.finish("出错了，请稍后再试")
+ else:
+ types = ["avatar", "head", "body", "skin", "cape", "model"]
+ for t in types:
+ if msg.startswith(t):
+ username = msg.replace(t, "", 1).strip()
+ if username:
+ uuid = await get_mc_uuid(username)
+ if not uuid:
+ await matcher.finish("出错了，请稍后再试")
+ if uuid == "none":
+ await matcher.finish("找不到该用户")
+ if t == "model":
+ await matcher.send("生成中，请耐心等待。。。")
+ result = await get_mcmodel(uuid)
+ else:
+ result = await get_crafatar(t, uuid)
+ if result:
+ await UniMessage.image(raw=result).send()
+ else:
+ await matcher.finish("出错了，请稍后再试")
+ await matcher.finish()
diff --git a/src/plugins/mc/data_source.py b/src/plugins/mc_tools/data_source.py
similarity index 83%
rename from src/plugins/mc/data_source.py
rename to src/plugins/mc_tools/data_source.py
index 286545e..4393a22 100644
--- a/src/plugins/mc/data_source.py
+++ b/src/plugins/mc_tools/data_source.py
@@ -1,18 +1,17 @@
-import httpx
-import base64
-import jinja2
import asyncio
+import base64
import traceback
-from PIL import Image
from io import BytesIO
from pathlib import Path
-from typing import List, Optional
-from PIL.Image import Image as IMG
-from mcstatus import MinecraftServer
+from typing import List, Optional, Tuple
-from nonebot_plugin_htmlrender import get_new_page
-from nonebot.adapters.onebot.v11 import Message, MessageSegment
+import httpx
+import jinja2
+from mcstatus import JavaServer
from nonebot.log import logger
+from nonebot_plugin_htmlrender import get_new_page
+from PIL import Image
+from PIL.Image import Image as IMG
dir_path = Path(__file__).parent
template_path = dir_path / "template"
@@ -30,7 +29,7 @@ async def get_mc_uuid(username: str) -> str:
if not result:
return "none"
return result.get("id", "none")
- except:
+ except Exception:
return ""
@@ -54,12 +53,12 @@ async def get_crafatar(type: str, uuid: str) -> Optional[bytes]:
resp = await client.get(url)
result = resp.content
return result
- except:
+ except Exception:
return None
def load_file(name):
- with (template_path / name).open("r", encoding="utf-8") as f:
+ with open(template_path / name, encoding="utf-8") as f:
return f.read()
@@ -102,14 +101,14 @@ async def get_mcmodel(uuid: str) -> Optional[BytesIO]:
optimize=False,
)
return output
- except:
+ except Exception:
logger.warning(traceback.format_exc())
return None
-async def get_mcstatus(addr: str) -> Optional[Message]:
+async def get_mcstatus(addr: str) -> Optional[Tuple[Optional[bytes], str]]:
try:
- server = MinecraftServer.lookup(addr)
+ server = JavaServer.lookup(addr)
status = await server.async_status()
version = status.version.name
protocol = status.version.protocol
@@ -120,18 +119,15 @@ async def get_mcstatus(addr: str) -> Optional[Message]:
description = status.description
latency = status.latency
favicon = status.favicon
- except:
+ except Exception:
logger.warning(traceback.format_exc())
return None
- msg = Message()
+ image = None
if favicon:
- msg.append(
- MessageSegment.image(
- "base64://" + favicon.replace("data:image/png;base64,", "")
- )
- )
- msg.append(
+ image = base64.b64decode(favicon.replace("data:image/png;base64,", "").encode())
+
+ msg = (
f"服务端版本：{version}\n"
f"协议版本：{protocol}\n"
f"当前人数：{players_online}/{players_max}\n"
@@ -139,4 +135,4 @@ async def get_mcstatus(addr: str) -> Optional[Message]:
f"描述文本：{description}\n"
f"游戏延迟：{latency}ms"
)
- return msg
+ return image, msg
diff --git a/src/plugins/mc/template/jquery.min.js b/src/plugins/mc_tools/template/jquery.min.js
similarity index 100%
rename from src/plugins/mc/template/jquery.min.js
rename to src/plugins/mc_tools/template/jquery.min.js
diff --git a/src/plugins/mc/template/skin.html b/src/plugins/mc_tools/template/skin.html
similarity index 100%
rename from src/plugins/mc/template/skin.html
rename to src/plugins/mc_tools/template/skin.html
diff --git a/src/plugins/mc/template/skin.min.js b/src/plugins/mc_tools/template/skin.min.js
similarity index 100%
rename from src/plugins/mc/template/skin.min.js
rename to src/plugins/mc_tools/template/skin.min.js
diff --git a/src/plugins/mc/template/three.min.js b/src/plugins/mc_tools/template/three.min.js
similarity index 100%
rename from src/plugins/mc/template/three.min.js
rename to src/plugins/mc_tools/template/three.min.js
diff --git a/src/plugins/nbnhhsh/__init__.py b/src/plugins/nbnhhsh/__init__.py
new file mode 100644
index 0000000..6062ac9
--- /dev/null
+++ b/src/plugins/nbnhhsh/__init__.py
@@ -0,0 +1,34 @@
+import re
+
+from nonebot import on_command
+from nonebot.adapters import Message
+from nonebot.params import CommandArg
+from nonebot.plugin import PluginMetadata
+
+from .data_source import get_nbnhhsh
+
+__plugin_meta__ = PluginMetadata(
+ name="能不能好好说话",
+ description="拼音首字母缩写翻译工具",
+ usage="nbnhhsh/缩写 xxx",
+ extra={
+ "example": "缩写 xswl",
+ },
+)
+
+nbnhhsh = on_command("nbnhhsh", aliases={"缩写"}, block=True, priority=13)
+
+
+@nbnhhsh.handle()
+async def _(msg: Message = CommandArg()):
+ keyword = msg.extract_plain_text().strip()
+ if not keyword:
+ await nbnhhsh.finish()
+ if not re.fullmatch(r"[a-zA-Z]+", keyword):
+ await nbnhhsh.finish()
+
+ res = await get_nbnhhsh(keyword)
+ if res:
+ await nbnhhsh.finish(res)
+ else:
+ await nbnhhsh.finish("找不到相关的缩写")
diff --git a/src/plugins/nbnhhsh/data_source.py b/src/plugins/nbnhhsh/data_source.py
new file mode 100644
index 0000000..4d7f173
--- /dev/null
+++ b/src/plugins/nbnhhsh/data_source.py
@@ -0,0 +1,15 @@
+import httpx
+
+
+async def get_nbnhhsh(keyword: str) -> str:
+ url = "https://lab.magiconch.com/api/nbnhhsh/guess"
+ headers = {"referer": "https://lab.magiconch.com/nbnhhsh/"}
+ data = {"text": keyword}
+ async with httpx.AsyncClient() as client:
+ resp = await client.post(url=url, headers=headers, data=data)
+ res = resp.json()
+ results = []
+ for i in res:
+ if "trans" in i and i["trans"]:
+ results.append(f"{i['name']} => {'，'.join(i['trans'])}")
+ return "\n".join(results)
diff --git a/src/plugins/nhd_girl/__init__.py b/src/plugins/nhd_girl/__init__.py
new file mode 100644
index 0000000..df9db14
--- /dev/null
+++ b/src/plugins/nhd_girl/__init__.py
@@ -0,0 +1,62 @@
+import random
+import re
+from typing import Dict, List
+
+from nonebot import on_message, require
+from nonebot.matcher import Matcher
+from nonebot.params import EventPlainText
+from nonebot.rule import Rule
+from nonebot.typing import T_State
+
+require("nonebot_plugin_userinfo")
+
+from nonebot_plugin_userinfo import EventUserInfo, UserInfo
+
+entries: Dict[str, List[str]] = {
+ r"[Nn][Hh][Dd]娘[ ,，]*求交往": [
+ "你是个好人，这张好人卡请收下吧~",
+ ],
+ r"[Nn][Hh][Dd]娘[ ,，]*求约会": [
+ "才不要呢！约会什么的最讨厌了！",
+ "好的 {nickname}，三分钟后来八舍楼下等我哦！",
+ ],
+ r"[Nn][Hh][Dd]娘[ ,，]*(你|早上|晚上|中午|下午)好": [
+ "你好~今天NHD娘也是活力满满地在工作着呢！",
+ "你好~这里是最勤劳最可爱的NHD娘~",
+ "你……你好！呜！被看到没睡醒的样子了>_<",
+ ],
+ r"[Nn][Hh][Dd]娘[ ,，]*(求|我要)米": [
+ "哼！人家才不知道什么叫米呢！",
+ "求米什么的最讨厌了！烦死了烦死了烦死了！",
+ "呜呜，为什么你们这些人总是米米米的，人家弄不懂啦~",
+ "呵呵，你们不知道，我跟马爷谈笑风生，你们这些求米的人啊，naive！",
+ ],
+ r"[Nn][Hh][Dd]娘[ ,，]*(马爷|麻耶)": [
+ "你说的是 sssdjay 吗？马爷可是永远的博士之光呢！NHD娘最喜欢他了！嗯嗯！",
+ ],
+}
+
+
+def nhd_girl_rule() -> Rule:
+ def checker(state: T_State, message: str = EventPlainText()) -> bool:
+ if not message:
+ return False
+ for pattern, replies in entries.items():
+ if re.fullmatch(pattern, message):
+ reply = random.choice(replies)
+ state["reply"] = reply
+ return True
+ return False
+
+ return Rule(checker)
+
+
+nhd_girl = on_message(nhd_girl_rule(), block=True, priority=15)
+
+
+@nhd_girl.handle()
+async def _(matcher: Matcher, state: T_State, user_info: UserInfo = EventUserInfo()):
+ reply: str = state["reply"]
+ user_name = user_info.user_name
+ reply = reply.format(nickname=user_name)
+ await matcher.finish(reply)
diff --git a/src/plugins/nhd_rss/__init__.py b/src/plugins/nhd_rss/__init__.py
new file mode 100644
index 0000000..f581a30
--- /dev/null
+++ b/src/plugins/nhd_rss/__init__.py
@@ -0,0 +1,11 @@
+from nonebot import require
+
+require("nonebot_plugin_saa")
+require("nonebot_plugin_htmlrender")
+require("nonebot_plugin_apscheduler")
+
+from nonebot_plugin_saa import enable_auto_select_bot
+
+enable_auto_select_bot()
+
+from . import pusher as pusher
diff --git a/src/plugins/nhd_rss/config.py b/src/plugins/nhd_rss/config.py
new file mode 100644
index 0000000..ba68b25
--- /dev/null
+++ b/src/plugins/nhd_rss/config.py
@@ -0,0 +1,13 @@
+from typing import List
+
+from nonebot import get_plugin_config
+from pydantic import BaseModel
+
+
+class Config(BaseModel):
+ nhd_rss_url: str = ""
+ nhd_rss_interval: int = 300
+ nhd_rss_targets: List[dict] = []
+
+
+rss_config = get_plugin_config(Config)
diff --git a/src/plugins/nhd_rss/data_source.py b/src/plugins/nhd_rss/data_source.py
new file mode 100644
index 0000000..36c1d15
--- /dev/null
+++ b/src/plugins/nhd_rss/data_source.py
@@ -0,0 +1,61 @@
+import re
+from dataclasses import dataclass
+from datetime import datetime
+from typing import List
+
+import dateparser
+import httpx
+from lxml import etree
+from nonebot.log import logger
+
+from .config import rss_config
+
+
+@dataclass
+class NHDRSSEntry:
+ title: str
+ subtitle: str
+ size: str
+ author: str
+ category: str
+ pubDate: datetime
+
+
+async def get_rss_entries() -> List[NHDRSSEntry]:
+ try:
+ async with httpx.AsyncClient() as client:
+ resp = await client.get(rss_config.nhd_rss_url, timeout=20)
+ content = resp.content
+ except Exception as e:
+ logger.warning(f"Error in get_rss_entries: {e}")
+ return []
+ root = etree.HTML(content, etree.XMLParser())
+ items = root.xpath("//item")
+
+ entries = []
+ for item in items:
+ try:
+ raw_title = str(item.xpath("./title/text()")[0])
+ raw_title, size, _ = re.split(
+ r"\[(\d{1,4}\.\d{2} [KMGTP]B)\]$", raw_title, maxsplit=1
+ )
+ title, subtitle, _ = re.split(r"\[(.*)\]$", raw_title, maxsplit=1)
+ author = str(item.xpath("./author/text()")[0]).split("@")[0]
+ category = str(item.xpath("./category/text()")[0])
+ pubDate = dateparser.parse(str(item.xpath("./pubDate/text()")[0]))
+ if not pubDate:
+ continue
+ entries.append(
+ NHDRSSEntry(
+ title=title,
+ subtitle=subtitle,
+ size=size,
+ author=author,
+ category=category,
+ pubDate=pubDate,
+ )
+ )
+ except Exception as e:
+ logger.warning(f"RSS item parse failed: {e}")
+ pass
+ return entries
diff --git a/src/plugins/nhd_rss/pusher.py b/src/plugins/nhd_rss/pusher.py
new file mode 100644
index 0000000..c313eed
--- /dev/null
+++ b/src/plugins/nhd_rss/pusher.py
@@ -0,0 +1,36 @@
+from datetime import datetime, timedelta, timezone
+
+from nonebot.log import logger
+from nonebot_plugin_apscheduler import scheduler
+from nonebot_plugin_saa import Image, MessageFactory, PlatformTarget
+
+from .config import rss_config
+from .data_source import get_rss_entries
+from .render import rss_to_image
+
+targets = [PlatformTarget.deserialize(target) for target in rss_config.nhd_rss_targets]
+last_update: datetime = datetime.now().astimezone(timezone(timedelta(hours=8)))
+
+
+@scheduler.scheduled_job(
+ "interval", seconds=rss_config.nhd_rss_interval, id="nhd_rss_schedule"
+)
+async def _():
+ entries = await get_rss_entries()
+ entries = sorted(entries, key=lambda x: x.pubDate)
+ entries = [e for e in entries if e.pubDate > last_update]
+ if not entries:
+ return
+ global last_update
+ last_update = entries[-1].pubDate
+
+ for entry in entries:
+ image = await rss_to_image(entry)
+ if not image:
+ continue
+ msg = MessageFactory([Image(image)])
+ for target in targets:
+ try:
+ await msg.send_to(target)
+ except Exception as e:
+ logger.warning(f"Error sending msg '{msg}' to target '{target}': {e}")
diff --git a/src/plugins/nhd_rss/render.py b/src/plugins/nhd_rss/render.py
new file mode 100644
index 0000000..be33c6f
--- /dev/null
+++ b/src/plugins/nhd_rss/render.py
@@ -0,0 +1,27 @@
+from pathlib import Path
+from typing import Optional
+
+import jinja2
+from nonebot.log import logger
+from nonebot_plugin_htmlrender import html_to_pic
+
+from .data_source import NHDRSSEntry
+
+dir_path = Path(__file__).parent
+template_path = dir_path / "template"
+env = jinja2.Environment(
+ loader=jinja2.FileSystemLoader(template_path), enable_async=True
+)
+
+
+async def rss_to_image(rss: NHDRSSEntry) -> Optional[bytes]:
+ try:
+ template = env.get_template("nhd.html")
+ html = await template.render_async(rss=rss)
+ return await html_to_pic(
+ html,
+ viewport={"width": 300, "height": 100},
+ template_path=f"file://{template_path.absolute()}",
+ )
+ except Exception as e:
+ logger.warning(f"Error in get_rss_entries: {e}")
diff --git a/src/plugins/rss/template/catsprites/Anime.png b/src/plugins/nhd_rss/template/catsprites/Anime.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/Anime.png
rename to src/plugins/nhd_rss/template/catsprites/Anime.png
diff --git a/src/plugins/rss/template/catsprites/Documentaries.png b/src/plugins/nhd_rss/template/catsprites/Documentaries.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/Documentaries.png
rename to src/plugins/nhd_rss/template/catsprites/Documentaries.png
diff --git a/src/plugins/rss/template/catsprites/E-learning.png b/src/plugins/nhd_rss/template/catsprites/E-learning.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/E-learning.png
rename to src/plugins/nhd_rss/template/catsprites/E-learning.png
diff --git a/src/plugins/rss/template/catsprites/Games.png b/src/plugins/nhd_rss/template/catsprites/Games.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/Games.png
rename to src/plugins/nhd_rss/template/catsprites/Games.png
diff --git a/src/plugins/rss/template/catsprites/HQ Audio.png b/src/plugins/nhd_rss/template/catsprites/HQ Audio.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/HQ Audio.png
rename to src/plugins/nhd_rss/template/catsprites/HQ Audio.png
diff --git a/src/plugins/rss/template/catsprites/Movies.png b/src/plugins/nhd_rss/template/catsprites/Movies.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/Movies.png
rename to src/plugins/nhd_rss/template/catsprites/Movies.png
diff --git a/src/plugins/rss/template/catsprites/Music Videos.png b/src/plugins/nhd_rss/template/catsprites/Music Videos.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/Music Videos.png
rename to src/plugins/nhd_rss/template/catsprites/Music Videos.png
diff --git a/src/plugins/rss/template/catsprites/Other.png b/src/plugins/nhd_rss/template/catsprites/Other.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/Other.png
rename to src/plugins/nhd_rss/template/catsprites/Other.png
diff --git a/src/plugins/rss/template/catsprites/Sports.png b/src/plugins/nhd_rss/template/catsprites/Sports.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/Sports.png
rename to src/plugins/nhd_rss/template/catsprites/Sports.png
diff --git a/src/plugins/rss/template/catsprites/TV Series.png b/src/plugins/nhd_rss/template/catsprites/TV Series.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/TV Series.png
rename to src/plugins/nhd_rss/template/catsprites/TV Series.png
diff --git a/src/plugins/rss/template/catsprites/TV Shows.png b/src/plugins/nhd_rss/template/catsprites/TV Shows.png
similarity index 100%
rename from src/plugins/rss/template/catsprites/TV Shows.png
rename to src/plugins/nhd_rss/template/catsprites/TV Shows.png
diff --git a/src/plugins/rss/template/nhd.html b/src/plugins/nhd_rss/template/nhd.html
similarity index 70%
rename from src/plugins/rss/template/nhd.html
rename to src/plugins/nhd_rss/template/nhd.html
index 79e79f0..1741dbf 100644
--- a/src/plugins/rss/template/nhd.html
+++ b/src/plugins/nhd_rss/template/nhd.html
@@ -78,54 +78,45 @@
}
- {% set info_ = info['title'] | split_nhd_title(rss.url) %}
- {% set category = info_.get('category') %}
- {% if not category %} {% set category = info['tags'] %} {% endif %}
- 
+ 
- {% set subtitle = info_.get('subtitle') %}
- {% if subtitle %}
- {% endif %}
- {% set size_num = info_.get('size_num') %}
- {% if size_num %}
+ {% set size_num, size_unit = rss.size.split(' ') %}
- {% endif %}
- {% set author = info_.get('author') %}
- {% if not author %} {% set author = info['author'] %} {% endif %}
+ {% set author = rss.author %}
{% if author == 'anonymous' %} {% set author = '匿名' %} {% else %} {% set author = '' + author + '' %} {% endif %}
{{ author }}
diff --git a/src/plugins/nhd_rss/template/rss_logo.jpg b/src/plugins/nhd_rss/template/rss_logo.jpg
new file mode 100644
index 0000000..ac2de61
Binary files /dev/null and b/src/plugins/nhd_rss/template/rss_logo.jpg differ
diff --git a/src/plugins/pixiv/__init__.py b/src/plugins/pixiv/__init__.py
deleted file mode 100644
index e745b8e..0000000
--- a/src/plugins/pixiv/__init__.py
+++ /dev/null
@@ -1,37 +0,0 @@
-from nonebot import on_command
-from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import Message
-from nonebot.params import CommandArg, EventToMe
-
-from .config import Config
-from .data_source import get_pixiv
-
-
-__plugin_meta__ = PluginMetadata(
- name="pixiv",
- description="Pixiv搜图、排行榜",
- usage="pixiv {日榜/周榜/月榜/id/关键词}，关键词搜索需要 @我",
- config=Config,
- extra={
- "example": "pixiv 日榜\n@小Q pixiv 伊蕾娜",
- },
-)
-
-
-pixiv = on_command("pixiv", block=True, priority=12)
-
-
-@pixiv.handle()
-async def _(msg: Message = CommandArg(), tome: bool = EventToMe()):
- keyword = msg.extract_plain_text().strip()
- if not keyword:
- await pixiv.finish()
-
- if not keyword.isdigit() and keyword not in ["日榜", "周榜", "月榜"]:
- if not tome:
- await pixiv.finish()
-
- res = await get_pixiv(keyword)
- if not res:
- await pixiv.finish("出错了，请稍后再试")
- await pixiv.finish(res)
diff --git a/src/plugins/pixiv/config.py b/src/plugins/pixiv/config.py
deleted file mode 100644
index 1c10088..0000000
--- a/src/plugins/pixiv/config.py
+++ /dev/null
@@ -1,5 +0,0 @@
-from pydantic import BaseModel, Extra
-
-
-class Config(BaseModel, extra=Extra.ignore):
- pixiv_token = ""
diff --git a/src/plugins/pixiv/data_source.py b/src/plugins/pixiv/data_source.py
deleted file mode 100644
index 3cba2c9..0000000
--- a/src/plugins/pixiv/data_source.py
+++ /dev/null
@@ -1,91 +0,0 @@
-import httpx
-import random
-from typing import Union
-from pixivpy_async import PixivClient, AppPixivAPI
-
-from nonebot import get_driver
-from nonebot.log import logger
-from nonebot.adapters.onebot.v11 import Message, MessageSegment
-
-from .config import Config
-
-global_config = get_driver().config
-pixiv_config = Config.parse_obj(global_config.dict())
-proxy = str(global_config.http_proxy)
-
-
-async def get_pixiv(keyword: str) -> Union[str, Message]:
- try:
- if keyword.isdigit():
- illust = await get_by_id(int(keyword))
- if not illust:
- return "找不到该id的作品"
- illusts = [illust["illust"]]
- elif keyword == "日榜":
- illusts = await get_by_ranking(mode="day")
- elif keyword == "周榜":
- illusts = await get_by_ranking(mode="week")
- elif keyword == "月榜":
- illusts = await get_by_ranking(mode="month")
- else:
- illusts = await get_by_search(keyword)
- if not illusts:
- return "找不到相关的作品"
- if not illusts:
- return "出错了，请稍后再试"
- msg = await to_msg(illusts)
- return msg
- except Exception as e:
- logger.warning(f"Error in get_pixiv({keyword}): {e}")
- return "出错了，请稍后再试"
-
-
-async def to_msg(illusts) -> Message:
- msg = Message()
- async with PixivClient(proxy=proxy, timeout=20) as client:
- aapi = AppPixivAPI(client=client, proxy=proxy)
- await aapi.login(refresh_token=pixiv_config.pixiv_token)
- for illust in illusts:
- try:
- url: str = illust["image_urls"]["large"]
- url = url.replace("_webp", "").replace("i.pximg.net", "i.pixiv.re")
- async with httpx.AsyncClient() as client:
- resp = await client.get(url, timeout=20)
- result = resp.content
- if result:
- msg.append("{} ({})".format(illust["title"], illust["id"]))
- msg.append(MessageSegment.image(result))
- except Exception as e:
- logger.warning(f"Error downloading image: {e}")
- return msg
-
-
-async def get_by_ranking(mode="day", num=3):
- async with PixivClient(proxy=proxy, timeout=20) as client:
- aapi = AppPixivAPI(client=client, proxy=proxy)
- await aapi.login(refresh_token=pixiv_config.pixiv_token)
- illusts = await aapi.illust_ranking(mode)
- illusts = illusts["illusts"]
- random.shuffle(illusts)
- return illusts[0:num]
-
-
-async def get_by_search(keyword, num=3):
- async with PixivClient(proxy=proxy, timeout=20) as client:
- aapi = AppPixivAPI(client=client, proxy=proxy)
- await aapi.login(refresh_token=pixiv_config.pixiv_token)
- illusts = await aapi.search_illust(keyword)
- illusts = illusts["illusts"]
- illusts = sorted(illusts, key=lambda i: i["total_bookmarks"], reverse=True)
- if len(illusts) > num * 3:
- illusts = illusts[0 : int(len(illusts) / 2)]
- random.shuffle(illusts)
- return illusts[0 : min(num, len(illusts))]
-
-
-async def get_by_id(work_id):
- async with PixivClient(proxy=proxy, timeout=20) as client:
- aapi = AppPixivAPI(client=client, proxy=proxy)
- await aapi.login(refresh_token=pixiv_config.pixiv_token)
- illust = await aapi.illust_detail(work_id)
- return illust
diff --git a/src/plugins/pixiv/pixiv_auth.py b/src/plugins/pixiv/pixiv_auth.py
deleted file mode 100644
index 3fbfe1f..0000000
--- a/src/plugins/pixiv/pixiv_auth.py
+++ /dev/null
@@ -1,115 +0,0 @@
-#!/usr/bin/env python
-
-from argparse import ArgumentParser
-from base64 import urlsafe_b64encode
-from hashlib import sha256
-from pprint import pprint
-from secrets import token_urlsafe
-from sys import exit
-from urllib.parse import urlencode
-from webbrowser import open as open_url
-
-import requests
-
-# Latest app version can be found using GET /v1/application-info/android
-USER_AGENT = "PixivAndroidApp/5.0.234 (Android 11; Pixel 5)"
-REDIRECT_URI = "https://app-api.pixiv.net/web/v1/users/auth/pixiv/callback"
-LOGIN_URL = "https://app-api.pixiv.net/web/v1/login"
-AUTH_TOKEN_URL = "https://oauth.secure.pixiv.net/auth/token"
-CLIENT_ID = "MOBrBDS8blbauoSck0ZfDbtuzpyT"
-CLIENT_SECRET = "lsACyCD94FhDUtGTXi3QzcFE2uU1hqtDaKeqrdwj"
-
-
-def s256(data):
- """S256 transformation method."""
-
- return urlsafe_b64encode(sha256(data).digest()).rstrip(b"=").decode("ascii")
-
-
-def oauth_pkce(transform):
- """Proof Key for Code Exchange by OAuth Public Clients (RFC7636)."""
-
- code_verifier = token_urlsafe(32)
- code_challenge = transform(code_verifier.encode("ascii"))
-
- return code_verifier, code_challenge
-
-
-def print_auth_token_response(response):
- data = response.json()
-
- try:
- access_token = data["access_token"]
- refresh_token = data["refresh_token"]
- except KeyError:
- print("error:")
- pprint(data)
- exit(1)
-
- print("access_token:", access_token)
- print("refresh_token:", refresh_token)
- print("expires_in:", data.get("expires_in", 0))
-
-
-def login():
- code_verifier, code_challenge = oauth_pkce(s256)
- login_params = {
- "code_challenge": code_challenge,
- "code_challenge_method": "S256",
- "client": "pixiv-android",
- }
-
- open_url(f"{LOGIN_URL}?{urlencode(login_params)}")
-
- try:
- code = input("code: ").strip()
- except (EOFError, KeyboardInterrupt):
- return
-
- response = requests.post(
- AUTH_TOKEN_URL,
- data={
- "client_id": CLIENT_ID,
- "client_secret": CLIENT_SECRET,
- "code": code,
- "code_verifier": code_verifier,
- "grant_type": "authorization_code",
- "include_policy": "true",
- "redirect_uri": REDIRECT_URI,
- },
- headers={"User-Agent": USER_AGENT},
- )
-
- print_auth_token_response(response)
-
-
-def refresh(refresh_token):
- response = requests.post(
- AUTH_TOKEN_URL,
- data={
- "client_id": CLIENT_ID,
- "client_secret": CLIENT_SECRET,
- "grant_type": "refresh_token",
- "include_policy": "true",
- "refresh_token": refresh_token,
- },
- headers={"User-Agent": USER_AGENT},
- )
- print_auth_token_response(response)
-
-
-def main():
- parser = ArgumentParser()
- subparsers = parser.add_subparsers()
- parser.set_defaults(func=lambda _: parser.print_usage())
- login_parser = subparsers.add_parser("login")
- login_parser.set_defaults(func=lambda _: login())
- refresh_parser = subparsers.add_parser("refresh")
- refresh_parser.add_argument("refresh_token")
- refresh_parser.set_defaults(func=lambda ns: refresh(ns.refresh_token))
- args = parser.parse_args()
- args.func(args)
-
-
-if __name__ == "__main__":
- main()
diff --git a/src/plugins/render/__init__.py b/src/plugins/render/__init__.py
index 4f9cf19..d8fad21 100644
--- a/src/plugins/render/__init__.py
+++ b/src/plugins/render/__init__.py
@@ -1,12 +1,15 @@
-from nonebot.params import CommandArg
from nonebot import on_command, require
+from nonebot.adapters import Message
+from nonebot.params import CommandArg
from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import Message, MessageSegment
-
-from .data_source import t2p, m2p
+require("nonebot_plugin_alconna")
require("nonebot_plugin_htmlrender")
+from nonebot_plugin_alconna import UniMessage
+
+from .data_source import m2p, t2p
+
__plugin_meta__ = PluginMetadata(
name="文本渲染",
description="文本、Markdown转图片",
@@ -29,7 +32,7 @@ async def _(msg: Message = CommandArg()):
img = await t2p(text)
if img:
- await text2pic.finish(MessageSegment.image(img))
+ await UniMessage.image(raw=img).send()
@md2pic.handle()
@@ -40,4 +43,4 @@ async def _(msg: Message = CommandArg()):
img = await m2p(text)
if img:
- await md2pic.finish(MessageSegment.image(img))
+ await UniMessage.image(raw=img).send()
diff --git a/src/plugins/render/data_source.py b/src/plugins/render/data_source.py
index 66aece0..0884b68 100644
--- a/src/plugins/render/data_source.py
+++ b/src/plugins/render/data_source.py
@@ -1,6 +1,7 @@
-import jinja2
from pathlib import Path
-from nonebot_plugin_htmlrender import text_to_pic, md_to_pic
+
+import jinja2
+from nonebot_plugin_htmlrender import md_to_pic, text_to_pic
dir_path = Path(__file__).parent
tpl_path = dir_path / "template"
diff --git a/src/plugins/repeat/__init__.py b/src/plugins/repeat/__init__.py
deleted file mode 100644
index d168e91..0000000
--- a/src/plugins/repeat/__init__.py
+++ /dev/null
@@ -1,115 +0,0 @@
-import re
-from typing import Dict
-from nonebot.typing import T_State
-from nonebot.params import EventMessage
-from nonebot import on_message, get_driver
-from nonebot.adapters.onebot.v11 import (
- GroupMessageEvent,
- Message,
- MessageSegment,
- unescape,
-)
-from nonebot.plugin import PluginMetadata
-
-from .config import Config
-
-repeat_config = Config.parse_obj(get_driver().config.dict())
-
-
-__plugin_meta__ = PluginMetadata(
- name="复读",
- description="复读机",
- usage="重复两次的内容会被复读；只能复读文本和图片；指令及其回复不会被复读",
- config=Config,
-)
-
-
-class Counter:
- def __init__(self):
- self.count = 0
- self.msg = None
-
- def clear(self):
- self.count = 0
- self.msg = None
-
- def add_msg(self, msg: Message):
- for m in msg:
- if m.type not in ["text", "face", "at", "image"]:
- self.clear()
- return
- if m.type == "text":
- m.data["text"] = unescape(m.data["text"])
-
- if self.compare_msg(msg):
- self.count += 1
- self.msg = msg
- else:
- self.count = 1
- self.msg = msg
-
- def compare_msg(self, msg: Message):
- if not self.msg:
- return True
- if len(self.msg) != len(msg):
- return False
- for m1, m2 in zip(self.msg, msg):
- if m1.type != m2.type:
- return False
- if not self.compare_msgseg(m1, m2):
- return False
- return True
-
- def compare_msgseg(self, msg1: MessageSegment, msg2: MessageSegment):
- msg_type = msg1.type
- if msg_type == "text":
- text = msg2.data["text"]
- if (
- not text
- or "此处消息的转义尚未被插件支持" in text
- or "请使用最新版手机QQ体验新功能" in text
- or re.fullmatch(r"\[\S+\]", text)
- ):
- return False
- return msg1.data["text"] == msg2.data["text"]
- elif msg_type == "face":
- return msg1.data["id"] == msg2.data["id"]
- elif msg_type == "at":
- return msg1.data["qq"] == msg2.data["qq"]
- elif msg_type == "image":
- return msg1.data["file"] == msg2.data["file"]
-
-
-msgs: Dict[int, Counter] = {}
-
-
-def repeat_rule(
- event: GroupMessageEvent, state: T_State, msg: Message = EventMessage()
-) -> bool:
- if not msg:
- return False
-
- group_id = event.group_id
- if group_id not in msgs.keys():
- msgs[group_id] = Counter()
- counter = msgs[group_id]
-
- if event.reply:
- counter.clear()
- else:
- counter.add_msg(msg)
-
- if counter.count == repeat_config.repeat_count:
- state["msg"] = counter.msg
- return True
- return False
-
-
-repeat = on_message(repeat_rule, block=False, priority=101)
-
-
-@repeat.handle()
-async def _(state: T_State):
- msg: Message = state["msg"]
- if msg:
- await repeat.finish(msg)
diff --git a/src/plugins/repeat/config.py b/src/plugins/repeat/config.py
deleted file mode 100644
index 25c8f07..0000000
--- a/src/plugins/repeat/config.py
+++ /dev/null
@@ -1,5 +0,0 @@
-from pydantic import BaseModel, Extra
-
-
-class Config(BaseModel, extra=Extra.ignore):
- repeat_count: int = 2
diff --git a/src/plugins/rss/__init__.py b/src/plugins/rss/__init__.py
deleted file mode 100644
index 2395ec7..0000000
--- a/src/plugins/rss/__init__.py
+++ /dev/null
@@ -1,107 +0,0 @@
-from typing import List
-from argparse import Namespace
-from nonebot.matcher import Matcher
-from nonebot import on_shell_command
-from nonebot.rule import ArgumentParser
-from nonebot.plugin import PluginMetadata
-from nonebot.params import ShellCommandArgs, Depends
-from nonebot.adapters.onebot.v11 import MessageEvent, GroupMessageEvent
-
-from .monitor import *
-from .rss_class import RSS
-from .config import Config
-from .data_source import update_rss_info
-from .rss_list import (
- get_rss_list,
- clear_rss_list,
- add_rss_list,
- del_rss_list,
- DupeError,
-)
-
-__plugin_meta__ = PluginMetadata(
- name="RSS订阅",
- description="订阅rss链接并以图片形式发出",
- usage=(
- "添加订阅：rss add {订阅名} {RSSHub路径/完整URL}\n"
- "取消订阅：rss del {订阅名}\n"
- "订阅列表：rss list\n"
- "清空订阅：rss clear"
- ),
- config=Config,
- extra={
- "example": "rss add 订阅1 /bilibili/user/dynamic/282994",
- },
-)
-
-
-async def add_rss(matcher: Matcher, user_id: str, name: str, url: str, **kwargs):
- new_rss = RSS(name, url)
- status = await update_rss_info(new_rss)
- if not status:
- await matcher.finish("获取RSS信息失败，请检查链接或稍后再试")
- try:
- add_rss_list(user_id, new_rss)
- except DupeError:
- await matcher.finish("已存在该名称的订阅，请更换名称或删除之前的订阅")
- await matcher.finish(f"成功添加订阅 {name}：{new_rss.title}")
-
-
-async def del_rss(matcher: Matcher, user_id: str, name: str, **kwargs):
- try:
- del_rss_list(user_id, name)
- except DupeError:
- await matcher.finish("不存在该名称的订阅")
- await matcher.finish(f"成功删除订阅 {name}")
-
-
-async def list_rss(matcher: Matcher, user_id: str, **kwargs):
- rss_list: List[RSS] = get_rss_list(user_id)
- if not rss_list:
- await matcher.finish("目前还没有任何订阅")
- msg = "已订阅以下内容:\n"
- for r in rss_list:
- msg += f"\n{r.name} ({r.url})"
- await matcher.finish(msg)
-
-
-async def clear_rss(matcher: Matcher, user_id: str, **kwargs):
- clear_rss_list(user_id)
- await matcher.finish("订阅列表已清空")
-
-
-rss_parser = ArgumentParser("rss")
-
-rss_subparsers = rss_parser.add_subparsers()
-
-add_parser = rss_subparsers.add_parser("add", aliases=("添加"))
-add_parser.add_argument("name")
-add_parser.add_argument("url")
-add_parser.set_defaults(func=add_rss)
-
-del_parser = rss_subparsers.add_parser("del", aliases=("删除"))
-del_parser.add_argument("name")
-del_parser.set_defaults(func=del_rss)
-
-list_parser = rss_subparsers.add_parser("list", aliases=("列表"))
-list_parser.set_defaults(func=list_rss)
-
-clear_parser = rss_subparsers.add_parser("clear", aliases=("清空"))
-clear_parser.set_defaults(func=clear_rss)
-
-rss = on_shell_command("rss", parser=rss_parser, block=True, priority=12)
-
-
-def get_id(event: MessageEvent):
- if isinstance(event, GroupMessageEvent):
- return "group_" + str(event.group_id)
- else:
- return "private_" + str(event.user_id)
-
-
-@rss.handle()
-async def _(args: Namespace = ShellCommandArgs(), user_id: str = Depends(get_id)):
- args.matcher = rss
- args.user_id = user_id
- if hasattr(args, "func"):
- await args.func(**vars(args))
diff --git a/src/plugins/rss/config.py b/src/plugins/rss/config.py
deleted file mode 100644
index 4bebf14..0000000
--- a/src/plugins/rss/config.py
+++ /dev/null
@@ -1,6 +0,0 @@
-from pydantic import BaseModel, Extra
-
-
-class Config(BaseModel, extra=Extra.ignore):
- rss_update_cron: list = ["0", "*/5", "*", "*", "*", "*"]
-
diff --git a/src/plugins/rss/data_source.py b/src/plugins/rss/data_source.py
deleted file mode 100644
index fe7eb2e..0000000
--- a/src/plugins/rss/data_source.py
+++ /dev/null
@@ -1,81 +0,0 @@
-import httpx
-import feedparser
-from typing import List
-from urllib.parse import quote
-
-from nonebot.log import logger
-
-from .rss_class import RSS
-
-
-async def get_rss_info(url: str) -> dict:
- try:
- async with httpx.AsyncClient() as client:
- resp = await client.get(url, timeout=20)
- result = resp.text
- return feedparser.parse(result)
- except Exception as e:
- logger.warning(f"Error in get_rss_info({url}): {e}")
- return {}
-
-
-async def update_rss_info(rss: RSS) -> bool:
- info = await get_rss_info(rss.url)
- if not info:
- return False
- rss.title = info["feed"].get("title", "")
- rss.link = info["feed"].get("link", "")
- try:
- rss.logo = info["feed"]["image"]["href"]
- except:
- rss.logo = (
- f"https://ui-avatars.com/api/?background=random&name={quote(rss.title)}"
- )
- rss.rights = info["feed"].get("rights", "")
- return True
-
-
-async def update_rss(rss: RSS) -> List[dict]:
- info = await get_rss_info(rss.url)
- if not info:
- return []
- new_entries = []
- entries = info.get("entries")
- if not entries:
- return []
- for entry in entries[::-1]:
- try:
- time = RSS.parse_time(entry["published_parsed"])
- if time <= rss.last_update:
- continue
- title = entry["title"]
- summary = entry["summary"]
- link = entry["link"]
- authors = []
- tags = []
- try:
- for author in entry["authors"]:
- authors.append(author["name"])
- for tag in entry["tags"]:
- tags.append(tag["term"])
- except:
- pass
- new_entries.append(
- {
- "time": time.strftime("%c"),
- "title": title,
- "summary": summary,
- "link": link,
- "author": ", ".join(authors),
- "tags": ", ".join(tags),
- }
- )
- except:
- continue
- try:
- newest_time = RSS.parse_time(entries[0]["published_parsed"])
- if newest_time > rss.last_update:
- rss.last_update = newest_time
- except:
- rss.last_update = RSS.time_now()
- return new_entries
diff --git a/src/plugins/rss/monitor.py b/src/plugins/rss/monitor.py
deleted file mode 100644
index 6e5f3af..0000000
--- a/src/plugins/rss/monitor.py
+++ /dev/null
@@ -1,60 +0,0 @@
-import re
-from nonebot import get_bot, get_driver
-from nonebot.adapters.onebot.v11 import Bot
-from nonebot_plugin_apscheduler import scheduler
-
-from .render import rss_to_msg
-from .data_source import update_rss
-from .rss_list import get_rss_list, get_user_ids, dump_rss_list
-
-from .config import Config
-
-rss_config = Config.parse_obj(get_driver().config.dict())
-
-
-def user_type(user_id: str):
- p_group = r"group_(\d+)"
- p_private = r"private_(\d+)"
- match = re.fullmatch(p_group, user_id)
- if match:
- return "group", match.group(1)
- match = re.fullmatch(p_private, user_id)
- if match:
- return "private", match.group(1)
- return "", user_id
-
-
-async def rss_monitor():
- bot = get_bot()
- assert isinstance(bot, Bot)
- user_ids = get_user_ids()
- for user_id in user_ids:
- user_rss_list = get_rss_list(user_id)
- for rss in user_rss_list:
- entries = await update_rss(rss)
- for entry in entries:
- msg = await rss_to_msg(rss, entry)
- if not msg:
- continue
- type, id = user_type(user_id)
- if type == "group":
- await bot.send_group_msg(group_id=int(id), message=msg)
- elif type == "private":
- await bot.send_private_msg(user_id=int(id), message=msg)
- dump_rss_list()
-
-
-rss_cron = rss_config.rss_update_cron
-scheduler.add_job(
- rss_monitor,
- "cron",
- second=rss_cron[0],
- minute=rss_cron[1],
- hour=rss_cron[2],
- day=rss_cron[3],
- month=rss_cron[4],
- year=rss_cron[5],
- id="rss_monitor",
- coalesce=True,
- misfire_grace_time=30,
-)
diff --git a/src/plugins/rss/render.py b/src/plugins/rss/render.py
deleted file mode 100644
index fced249..0000000
--- a/src/plugins/rss/render.py
+++ /dev/null
@@ -1,113 +0,0 @@
-import re
-import httpx
-import base64
-import jinja2
-import mimetypes
-from pathlib import Path
-from typing import Optional
-from nonebot.adapters.onebot.v11 import Message, MessageSegment
-from nonebot_plugin_htmlrender import html_to_pic
-
-from .rss_class import RSS
-
-dir_path = Path(__file__).parent
-template_path = dir_path / "template"
-env = jinja2.Environment(
- loader=jinja2.FileSystemLoader(template_path), enable_async=True
-)
-
-
-async def rss_to_msg(rss: RSS, info: dict) -> Optional[Message]:
- msg = Message()
- img = await rss_to_image(rss, info)
- if not img:
- return None
- msg.append(MessageSegment.image(img))
- # msg.append(info["link"])
- return msg
-
-
-async def rss_to_image(rss: RSS, info: dict) -> bytes:
- html = await rss_to_html(rss, info)
- html = await replace_url(html, rss.link)
- return await html_to_pic(html, viewport={"width": 300, "height": 100})
-
-
-async def rss_to_html(rss: RSS, info: dict) -> str:
- template = env.get_template(f"{rss.style}.html")
- return await template.render_async(rss=rss, info=info)
-
-
-async def replace_url(text: str, base_url: str) -> str:
- pattern = r"![]()
"
- urls = re.findall(pattern, text, re.DOTALL)
- for url in urls:
- if url.startswith("data:image"):
- continue
- url_new = RSS.parse_url(url, base_url)
- b64 = await url_to_b64(url_new)
- text = text.replace(url, b64, 1)
- return text
-
-
-async def url_to_b64(url: str) -> str:
- result = await download_img(url)
- if not result:
- return url
- type = mimetypes.guess_type(url)[0]
- if not type:
- type = "image"
- return f"data:{type};base64,{base64.b64encode(result).decode()}"
-
-
-async def download_img(url: str) -> Optional[bytes]:
- try:
- async with httpx.AsyncClient() as client:
- resp = await client.get(url, timeout=20)
- result = resp.read()
- return result
- except:
- return None
-
-
-def split_nhd_title(text, url) -> Optional[dict]:
- text = text.strip()
- pattern = r""
- pattern_fail = r""
-
- p_url = r"\[(?P.*?)\]"
- p_title = r"(?P.*?)"
- p_subtitle = r"\[(?P<subtitle>.*?)\]"
- p_size = r"\[(?P<size_num>[\d\.]+)\s*(?P<size_unit>[GMKTB]+)\]"
- p_author = r"\[(?P<author>\S+)\]"
-
- if "icat" in url:
- pattern += p_url
- pattern += p_title
- pattern_fail = pattern
- if "ismalldescr" in url:
- pattern += p_subtitle
- if "isize" in url:
- pattern += p_size
- pattern_fail += p_size
- if "iuplder" in url:
- pattern += p_author
- pattern_fail += p_author
-
- match = re.match(pattern, text)
- if not match:
- match = re.match(pattern_fail, text)
- if match:
- return match.groupdict()
-
-
-def load_category_img(category) -> str:
- img_path = template_path / "catsprites" / f"{category}.png"
- if not img_path.exists():
- return category
- with (img_path).open("rb") as f:
- return "data:image/png;base64," + base64.b64encode(f.read()).decode()
-
-
-env.filters["split_nhd_title"] = split_nhd_title
-env.filters["load_category_img"] = load_category_img
diff --git a/src/plugins/rss/rss_class.py b/src/plugins/rss/rss_class.py
deleted file mode 100644
index d5bac56..0000000
--- a/src/plugins/rss/rss_class.py
+++ /dev/null
@@ -1,97 +0,0 @@
-import re
-import time
-import pytz
-from typing import Union
-from pytz import timezone
-from datetime import datetime
-
-TZ = timezone("Asia/Shanghai")
-RSSHUB_PREFIX = "https://rsshub.app"
-
-
-class RSS:
- def __init__(
- self,
- name: str,
- url: str,
- title: str = "",
- link: str = "",
- logo: str = "",
- rights: str = "",
- style: str = "main",
- last_update: Union[datetime, time.struct_time, str] = "",
- ):
- self.name = name
- self.url = self.parse_url(url, RSSHUB_PREFIX) if url else ""
- self.title = title
- self.link = link
- self.logo = logo
- self.rights = rights
- self.style = style
- self.last_update = (
- self.parse_time(last_update) if last_update else self.time_now()
- )
-
- @staticmethod
- def parse_url(url: str, base_url: str) -> str:
- if re.match(r"https?://", url, re.IGNORECASE):
- return url
- if url[0] == "/":
- url = base_url + url
- else:
- url = base_url + "/" + url
- return url
-
- @staticmethod
- def parse_time(raw_time: Union[datetime, time.struct_time, str]) -> datetime:
- if isinstance(raw_time, datetime):
- return raw_time.astimezone(TZ)
- elif isinstance(raw_time, time.struct_time):
- return (
- datetime.fromtimestamp(time.mktime(raw_time))
- .replace(tzinfo=pytz.utc)
- .astimezone(TZ)
- )
- elif isinstance(raw_time, str):
- return datetime.fromisoformat(raw_time).astimezone(TZ)
-
- @staticmethod
- def time_now() -> datetime:
- return datetime.now().astimezone(TZ)
-
- @classmethod
- def from_json(cls, json_dict: dict):
- name = json_dict.get("name", "")
- url = json_dict.get("url", "")
- title = json_dict.get("title", "")
- link = json_dict.get("link", "")
- logo = json_dict.get("logo", "")
- rights = json_dict.get("rights", "")
- style = json_dict.get("style", "main")
- last_update = json_dict.get("last_update")
- if last_update:
- last_update = datetime.fromisoformat(last_update)
- else:
- last_update = cls.time_now()
- return cls(
- name,
- url,
- title=title,
- link=link,
- logo=logo,
- rights=rights,
- style=style,
- last_update=last_update,
- )
-
- def to_json(self) -> dict:
- return {
- "name": self.name,
- "url": self.url,
- "title": self.title,
- "link": self.link,
- "logo": self.logo,
- "rights": self.rights,
- "style": self.style,
- "last_update": self.last_update.isoformat(),
- }
diff --git a/src/plugins/rss/rss_list.py b/src/plugins/rss/rss_list.py
deleted file mode 100644
index 3eb6bac..0000000
--- a/src/plugins/rss/rss_list.py
+++ /dev/null
@@ -1,77 +0,0 @@
-import json
-from pathlib import Path
-from typing import List, Dict
-
-from .rss_class import RSS
-
-data_path = Path("data/rss")
-if not data_path.exists():
- data_path.mkdir(parents=True)
-rss_path = data_path / "rss_list.json"
-
-
-class DupeError(Exception):
- pass
-
-
-def _load_rss_list() -> Dict[str, List[RSS]]:
- try:
- rss_list = {}
- json_list: dict = json.load(rss_path.open("r", encoding="utf-8"))
- for user_id, user_rss_list in json_list.items():
- rss_list[user_id] = [RSS.from_json(rss) for rss in user_rss_list]
- return rss_list
- except FileNotFoundError:
- return {}
-
-
-_rss_list = _load_rss_list()
-
-
-def dump_rss_list():
- rss_path.parent.mkdir(parents=True, exist_ok=True)
- json_list = {}
- for user_id, user_rss_list in _rss_list.items():
- json_list[user_id] = [rss.to_json() for rss in user_rss_list]
- json.dump(
- json_list,
- rss_path.open("w", encoding="utf-8"),
- indent=4,
- separators=(",", ": "),
- ensure_ascii=False,
- )
-
-
-def get_user_ids() -> List[str]:
- return list(_rss_list.keys())
-
-
-def get_rss_list(user_id: str) -> List[RSS]:
- return _rss_list.get(user_id, []).copy()
-
-
-def add_rss_list(user_id: str, new_rss: RSS):
- user_sub_list = get_rss_list(user_id)
- names = [rss.name for rss in user_sub_list]
- if new_rss.name in names:
- raise DupeError
- if user_id not in _rss_list:
- _rss_list[user_id] = []
- _rss_list[user_id].append(new_rss)
- dump_rss_list()
-
-
-def del_rss_list(user_id: str, name: str):
- user_sub_list = get_rss_list(user_id)
- for rss in user_sub_list:
- if rss.name == name:
- _rss_list[user_id].remove(rss)
- dump_rss_list()
- return
- raise DupeError
-
-
-def clear_rss_list(user_id: str):
- if user_id in _rss_list:
- _rss_list.pop(user_id)
- dump_rss_list()
diff --git a/src/plugins/rss/template/main.html b/src/plugins/rss/template/main.html
deleted file mode 100644
index eb98f03..0000000
--- a/src/plugins/rss/template/main.html
+++ /dev/null
@@ -1,96 +0,0 @@
-<html>
- <head>
- <style>
- body {
- margin: 15pt;
- }
- hr {
- color: lightgray;
- background-color: lightgray;
- border-width: 0;
- height: 1px;
- margin-top: 10pt;
- margin-bottom: 10pt;
- margin-left: 5pt;
- margin-right: 20pt;
- }
- h2 {
- margin-right: 15pt;
- }
- .image {
- overflow: scroll;
- display: flex;
- flex-direction: column;
- width: max-content;
- max-width: 600pt;
- }
- .head {
- display: flex;
- margin-top: 5pt;
- margin-bottom: 5pt;
- margin-right: 15pt;
- padding: 10pt;
- background-color: whitesmoke;
- }
- .avatar {
- float: left;
- }
- .avatar img {
- margin-right: 10pt;
- width: 20pt;
- height: 20pt;
- }
- .name {
- font-size: 15pt;
- }
- .info {
- font-size: 12pt;
- color: gray;
- }
- .info p {
- margin-top: 5px;
- margin-bottom: 5px;
- }
- .summary {
- display: flex;
- flex-direction: column;
- font-size: 12pt;
- margin-right: 15pt;
- }
- .rights {
- text-align: center;
- font-size: 10pt;
- margin-top: 20pt;
- margin-right: 15pt;
- }
- </style>
- </head>
-
- <body>
- <div class="image">
- <div class="head">
- <div class="avatar">
- <img src="{{ rss.logo }}" />
- </div>
- <div class="name">{{ rss.name }} - {{ rss.title }}</div>
- </div>
- <h2>{{ info['title'] }}</h2>
- <div class="info">
- <p>
- {{ info['time'] }}
- {% if info['author'] %}
- | {{ info['author'] }}
- {% endif %}
- {% if info['tags'] %}
- | {{ info['tags'] }}
- {% endif %}
- </p>
- </div>
- <hr />
- <div class="summary">{{ info['summary'] }}</div>
- {% if rss.rights %}
- <div class="rights">{{ rss.rights }}</div>
- {% endif %}
- </div>
- </body>
-</html>
diff --git a/src/plugins/setu/__init__.py b/src/plugins/setu/__init__.py
index 4cdb4d2..09bbc29 100644
--- a/src/plugins/setu/__init__.py
+++ b/src/plugins/setu/__init__.py
@@ -1,41 +1,46 @@
-import asyncio
from typing import Type
-from nonebot import on_command
-from nonebot.rule import to_me
+
+from nonebot import on_command, require
+from nonebot.adapters import Message
from nonebot.matcher import Matcher
from nonebot.params import CommandArg
from nonebot.permission import SUPERUSER
from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import Bot, Message, MessageSegment
+from nonebot.rule import to_me
-from .data_source import get_setu
+require("nonebot_plugin_alconna")
+from nonebot_plugin_alconna import UniMessage
+
+from .data_source import get_setu
__plugin_meta__ = PluginMetadata(
name="涩图",
description="随机涩图",
- usage="@我 setu [keyword]",
+ usage="@我 setu {keyword}",
extra={
- "example": "@小Q setu 伊蕾娜",
+ "example": "@小Q setu 小萝莉",
},
)
-setu = on_command("setu", aliases={"涩图", "色图"}, rule=to_me(), block=True, priority=14)
+setu = on_command(
+ "setu", aliases={"涩图", "色图"}, rule=to_me(), block=True, priority=14
+)
setu_ = on_command("r18", permission=SUPERUSER, rule=to_me(), block=True, priority=14)
@setu.handle()
-async def _(bot: Bot, msg: Message = CommandArg()):
- await handle(bot, setu, msg)
+async def _(msg: Message = CommandArg()):
+ await handle(setu, msg)
@setu_.handle()
-async def _(bot: Bot, msg: Message = CommandArg()):
- await handle(bot, setu_, msg, r18=True)
+async def _(msg: Message = CommandArg()):
+ await handle(setu_, msg, r18=True)
-async def handle(bot: Bot, matcher: Type[Matcher], msg: Message, r18=False):
+async def handle(matcher: Type[Matcher], msg: Message, r18=False):
keyword = msg.extract_plain_text().strip()
res = await get_setu(keyword=keyword, r18=r18)
if not res:
@@ -43,9 +48,4 @@ async def handle(bot: Bot, matcher: Type[Matcher], msg: Message, r18=False):
if isinstance(res, str):
await matcher.finish(res)
else:
- result = await matcher.send(MessageSegment.image(res))
- msg_id = result["message_id"]
- loop = asyncio.get_running_loop()
- loop.call_later(
- 100, lambda: asyncio.ensure_future(bot.delete_msg(message_id=msg_id))
- )
+ await UniMessage.image(raw=res).send()
diff --git a/src/plugins/setu/data_source.py b/src/plugins/setu/data_source.py
index cca35be..4a4c2cd 100644
--- a/src/plugins/setu/data_source.py
+++ b/src/plugins/setu/data_source.py
@@ -1,8 +1,6 @@
-import httpx
-import random
-from PIL import Image
-from io import BytesIO
from typing import Optional, Union
+
+import httpx
from nonebot.log import logger
@@ -21,7 +19,7 @@ async def get_setu(keyword="", r18=False) -> Optional[Union[str, bytes]]:
result = resp.json()
if result["error"]:
logger.warning("lolicon error: " + result["error"])
- return None
+ return
if result["data"]:
setu_url = result["data"][0]["urls"]["regular"]
logger.info("Get setu url: " + setu_url)
@@ -29,17 +27,9 @@ async def get_setu(keyword="", r18=False) -> Optional[Union[str, bytes]]:
async with httpx.AsyncClient() as client:
resp = await client.get(setu_url, timeout=20)
result = resp.content
-
- # 随机涂黑一个像素点
- image = Image.open(BytesIO(result)).convert("RGB")
- x = random.randint(0, image.width - 1)
- y = random.randint(0, image.height - 1)
- image.putpixel((x, y), (0, 0, 0))
- output = BytesIO()
- image.save(output, format="jpeg")
- return output.getvalue()
+ return result
else:
return "找不到相关的涩图"
except Exception as e:
logger.warning(f"Error in get_setu({keyword}): {e}")
- return None
+ return
diff --git a/src/plugins/speak/__init__.py b/src/plugins/speak/__init__.py
deleted file mode 100644
index 405c140..0000000
--- a/src/plugins/speak/__init__.py
+++ /dev/null
@@ -1,34 +0,0 @@
-from nonebot import on_command
-from nonebot.rule import to_me
-from nonebot.params import CommandArg
-from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import Message, MessageSegment
-
-from .config import Config
-from .data_source import get_voice
-
-__plugin_meta__ = PluginMetadata(
- name="语音合成",
- description="文字转语音，支持中文/日文",
- usage="@我 说 {text}",
- config=Config,
- extra={
- "example": "@小Q 说你是猪",
- },
-)
-
-
-speak = on_command("speak", aliases={"说"}, block=True, rule=to_me(), priority=11)
-
-
-@speak.handle()
-async def _(msg: Message = CommandArg()):
- text = msg.extract_plain_text().strip()
- if not text:
- await speak.finish()
-
- voice = await get_voice(text)
- if voice:
- await speak.finish(MessageSegment.record(voice))
- else:
- await speak.finish("出错了，请稍后再试")
diff --git a/src/plugins/speak/config.py b/src/plugins/speak/config.py
deleted file mode 100644
index 502566b..0000000
--- a/src/plugins/speak/config.py
+++ /dev/null
@@ -1,7 +0,0 @@
-from pydantic import BaseModel, Extra
-
-
-class Config(BaseModel, extra=Extra.ignore):
- tencent_secret_id: str = ""
- tencent_secret_key: str = ""
- tts_project_id: str = ""
diff --git a/src/plugins/speak/data_source.py b/src/plugins/speak/data_source.py
deleted file mode 100644
index e188631..0000000
--- a/src/plugins/speak/data_source.py
+++ /dev/null
@@ -1,123 +0,0 @@
-import re
-import json
-import uuid
-import httpx
-import langid
-import traceback
-from io import BytesIO
-from typing import Optional, Union
-from pydub import AudioSegment
-from pydub.silence import detect_silence
-
-from tencentcloud.common import credential
-from tencentcloud.tts.v20190823 import tts_client, models
-
-from nonebot import get_driver
-from nonebot.log import logger
-
-from .config import Config
-
-tts_config = Config.parse_obj(get_driver().config.dict())
-
-
-async def get_voice(text, type=0) -> Optional[Union[str, BytesIO]]:
- try:
- if langid.classify(text)[0] == "ja":
- voice = await get_ai_voice(text, type)
- else:
- voice = await get_tx_voice(text, type)
- return voice
- except:
- logger.warning(traceback.format_exc())
- return None
-
-
-async def get_tx_voice(text, type=0) -> str:
- cred = credential.Credential(
- tts_config.tencent_secret_id, tts_config.tencent_secret_key
- )
- client = tts_client.TtsClient(cred, "ap-shanghai")
- req = models.TextToVoiceRequest()
-
- if type == 0:
- voice_type = 101016
- else:
- voice_type = 101010
-
- params = {
- "Text": text,
- "SessionId": str(uuid.uuid1()),
- "Volume": 5,
- "Speed": 0,
- "ProjectId": int(tts_config.tts_project_id),
- "ModelType": 1,
- "VoiceType": voice_type,
- }
- req.from_json_string(json.dumps(params))
- resp = client.TextToVoice(req)
- return f"base64://{resp.Audio}"
-
-
-async def get_ai_voice(text, type=0) -> Optional[BytesIO]:
- mp3_url = await get_ai_voice_url(text, type)
- if not mp3_url:
- return None
-
- async with httpx.AsyncClient() as client:
- resp = await client.get(mp3_url)
- result = resp.content
-
- return await split_voice(BytesIO(result))
-
-
-async def get_ai_voice_url(text, type=0) -> str:
- url = "https://cloud.ai-j.jp/demo/aitalk_demo.php"
- if type == 0:
- params = {
- "callback": "callback",
- "speaker_id": 555,
- "text": text,
- "ext": "mp3",
- "volume": 2.0,
- "speed": 1,
- "pitch": 1,
- "range": 1,
- "webapi_version": "v5",
- }
- else:
- params = {
- "callback": "callback",
- "speaker_id": 1214,
- "text": text,
- "ext": "mp3",
- "volume": 2.0,
- "speed": 1,
- "pitch": 1,
- "range": 1,
- "anger": 0,
- "sadness": 0,
- "joy": 0,
- "webapi_version": "v5",
- }
-
- async with httpx.AsyncClient() as client:
- resp = await client.get(url, params=params)
- result = resp.text
-
- match_obj = re.search(r'"url":"(.*?)"', result)
- if match_obj:
- mp3_url = "https:" + match_obj.group(1).replace(r"\/", "/")
- return mp3_url
- return ""
-
-
-async def split_voice(input) -> Optional[BytesIO]:
- sound = AudioSegment.from_file(input)
- silent_ranges = detect_silence(sound, min_silence_len=500, silence_thresh=-40)
- if len(silent_ranges) >= 1:
- first_silent_end = silent_ranges[0][1] - 300
- result = sound[first_silent_end:] + AudioSegment.silent(300)
- output = BytesIO()
- result.export(output, format="mp3")
- return output
- return None
diff --git a/src/plugins/tarot/__init__.py b/src/plugins/tarot/__init__.py
index 4525606..9ac4a9a 100644
--- a/src/plugins/tarot/__init__.py
+++ b/src/plugins/tarot/__init__.py
@@ -1,9 +1,14 @@
-from nonebot import on_command
+from nonebot import on_command, require
+from nonebot.matcher import Matcher
from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import MessageEvent, Message, MessageSegment
-from .data_source import get_tarot
+require("nonebot_plugin_alconna")
+require("nonebot_plugin_userinfo")
+
+from nonebot_plugin_alconna import Image, UniMessage
+from nonebot_plugin_userinfo import EventUserInfo, UserInfo
+from .data_source import get_tarot
__plugin_meta__ = PluginMetadata(
name="塔罗牌",
@@ -13,18 +18,21 @@
tarot = on_command(
- "塔罗牌", aliases={"单张塔罗牌", "塔罗牌占卜", "draw 单张塔罗牌", "draw 塔罗牌"}, block=True, priority=13
+ "塔罗牌",
+ aliases={"单张塔罗牌", "塔罗牌占卜", "draw 单张塔罗牌", "draw 塔罗牌"},
+ block=True,
+ priority=13,
)
@tarot.handle()
-async def _(event: MessageEvent):
- username = event.sender.card or event.sender.nickname
- message = Message(f"来看看 {username} 抽到了什么：")
+async def _(matcher: Matcher, userinfo: UserInfo = EventUserInfo()):
+ username = userinfo.user_displayname or userinfo.user_name
+ msg = UniMessage(f"来看看 {username} 抽到了什么：")
try:
img, meaning = await get_tarot()
- except:
- await tarot.finish("出错了，请稍后再试")
- message.append(MessageSegment.image(img))
- message.append(meaning)
- await tarot.finish(message)
+ except Exception:
+ await matcher.finish("出错了，请稍后再试")
+ msg += Image(raw=img)
+ msg += meaning
+ await msg.send()
diff --git a/src/plugins/tarot/data_source.py b/src/plugins/tarot/data_source.py
index e034724..0be7708 100644
--- a/src/plugins/tarot/data_source.py
+++ b/src/plugins/tarot/data_source.py
@@ -1,10 +1,11 @@
import json
import random
-from PIL import Image
from io import BytesIO
from pathlib import Path
from typing import Tuple
+from PIL import Image
+
dir_path = Path(__file__).parent
tarot_path = dir_path / "resources"
image_path = tarot_path / "images"
diff --git a/src/plugins/test/__init__.py b/src/plugins/test/__init__.py
index 120e4bc..8e642a5 100644
--- a/src/plugins/test/__init__.py
+++ b/src/plugins/test/__init__.py
@@ -1,6 +1,6 @@
from nonebot import on_command
-from nonebot.rule import to_me
from nonebot.permission import SUPERUSER
+from nonebot.rule import to_me
test = on_command("test", rule=to_me(), block=True, permission=SUPERUSER, priority=10)
diff --git a/src/plugins/tex/__init__.py b/src/plugins/tex/__init__.py
index f547b3c..3658afc 100644
--- a/src/plugins/tex/__init__.py
+++ b/src/plugins/tex/__init__.py
@@ -1,7 +1,12 @@
-from nonebot import on_command
+from nonebot import on_command, require
+from nonebot.adapters import Message
+from nonebot.matcher import Matcher
from nonebot.params import CommandArg
from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import Message, MessageSegment
+
+require("nonebot_plugin_alconna")
+
+from nonebot_plugin_alconna import UniMessage
from .data_source import tex2pic
@@ -10,7 +15,9 @@
description="支持行内公式和少量行间公式",
usage="tex {equation}",
extra={
- "example": "tex a + b = c\ntex \\begin{bmatrix} a & b \\\\ c & d \\end{bmatrix}",
+ "example": (
+ "tex a + b = c\ntex \\begin{bmatrix} a & b \\\\ c & d \\end{bmatrix}"
+ ),
},
)
@@ -19,13 +26,13 @@
@tex.handle()
-async def _(msg: Message = CommandArg()):
+async def _(matcher: Matcher, msg: Message = CommandArg()):
equation = msg.extract_plain_text().strip().strip("$")
if not equation:
- await tex.finish()
+ await matcher.finish()
image = await tex2pic(equation)
if image:
- await tex.finish(MessageSegment.image(image))
+ await UniMessage.image(raw=image).send()
else:
- await tex.finish("出错了，请检查公式或稍后再试")
+ await matcher.finish("出错了，请检查公式或稍后再试")
diff --git a/src/plugins/tex/data_source.py b/src/plugins/tex/data_source.py
index d6608b5..1b90fc9 100644
--- a/src/plugins/tex/data_source.py
+++ b/src/plugins/tex/data_source.py
@@ -1,9 +1,10 @@
import os
-import jinja2
-import tempfile
import subprocess
+import tempfile
from pathlib import Path
from typing import Optional
+
+import jinja2
from nonebot.log import logger
dir_path = Path(__file__).parent / "template"
@@ -51,7 +52,7 @@ async def tex2pic(equation, fmt="png", border=2, resolution=1000) -> Optional[by
"ppm": "",
}
if fmt in formats.keys():
- convert_cmd = f"pdftoppm -r %d -%s %s > %s" % (
+ convert_cmd = "pdftoppm -r %d -%s %s > %s" % (
resolution,
formats[fmt],
tmp_pdf,
diff --git a/src/plugins/welcome/__init__.py b/src/plugins/welcome/__init__.py
deleted file mode 100644
index a870cd1..0000000
--- a/src/plugins/welcome/__init__.py
+++ /dev/null
@@ -1,23 +0,0 @@
-from pathlib import Path
-from nonebot import on_notice
-from nonebot.rule import Rule
-from nonebot.adapters.onebot.v11 import (
- MessageSegment,
- Bot,
- GroupIncreaseNoticeEvent,
-)
-
-dir_path = Path(__file__).parent
-welcome_path = dir_path / "resources" / "welcome.jpg"
-
-
-async def welcome_rule(bot: Bot, event: GroupIncreaseNoticeEvent) -> bool:
- return event.user_id != bot.self_id
-
-
-welcome = on_notice(rule=Rule(welcome_rule), priority=10)
-
-
-@welcome.handle()
-async def _():
- await welcome.finish(MessageSegment.image(welcome_path.read_bytes()))
diff --git a/src/plugins/welcome/resources/welcome.jpg b/src/plugins/welcome/resources/welcome.jpg
deleted file mode 100644
index e8f9b4e..0000000
Binary files a/src/plugins/welcome/resources/welcome.jpg and /dev/null differ
diff --git a/src/plugins/what/__init__.py b/src/plugins/what/__init__.py
deleted file mode 100644
index c66e5db..0000000
--- a/src/plugins/what/__init__.py
+++ /dev/null
@@ -1,87 +0,0 @@
-import re
-from nonebot.plugin import PluginMetadata
-from nonebot import on_keyword, on_command
-from nonebot.params import CommandArg, EventPlainText, EventToMe
-from nonebot.adapters.onebot.v11 import Message
-
-from .data_source import get_content
-
-__plugin_meta__ = PluginMetadata(
- name="百科",
- description="缩写查询、梗百科",
- usage=(
- "1. 百科 {keyword}，来源为nbnhhsh、小鸡词典、百度百科\n"
- "2. {keyword} 是啥/是什么，来源为nbnhhsh、小鸡词典\n"
- "3. 缩写 {keyword}，来源为nbnhhsh"
- ),
- extra={
- "example": "缩写 xswl\nxswl 是啥\n百科 洛天依",
- },
-)
-
-
-commands = {"是啥", "是什么", "是谁"}
-what = on_keyword(commands, priority=14)
-baike = on_command("baike", aliases={"what", "百科"}, block=True, priority=13)
-nbnhhsh = on_command("nbnhhsh", aliases={"缩写"}, block=True, priority=13)
-
-
-@what.handle()
-async def _(msg: str = EventPlainText(), to_me: bool = EventToMe()):
- def split_command(msg):
- for command in commands:
- if command in msg:
- prefix, suffix = re.split(command, msg)
- return prefix, suffix
- return "", ""
-
- msg = msg.strip().strip(".>,?!。，（）()[]【】")
- prefix_words = ["这", "这个", "那", "那个", "你", "我", "他", "它"]
- suffix_words = ["意思", "梗", "玩意", "鬼"]
- prefix, suffix = split_command(msg)
- if (not prefix or prefix in prefix_words) or (
- suffix and suffix not in suffix_words
- ):
- what.block = False
- await what.finish()
- keyword = prefix
-
- if to_me:
- res = await get_content(keyword)
- else:
- res = await get_content(keyword, sources=["jiki", "nbnhhsh"])
-
- if res:
- what.block = True
- await what.finish(res)
- else:
- what.block = False
- await what.finish()
-
-
-@baike.handle()
-async def _(msg: Message = CommandArg()):
- keyword = msg.extract_plain_text().strip()
- if not keyword:
- await baike.finish()
-
- res = await get_content(keyword)
- if res:
- await baike.finish(res)
- else:
- await baike.finish("找不到相关的条目")
-
-
-@nbnhhsh.handle()
-async def _(msg: Message = CommandArg()):
- keyword = msg.extract_plain_text().strip()
- if not keyword:
- await nbnhhsh.finish()
- if not re.fullmatch(r"[a-zA-Z]+", keyword):
- await nbnhhsh.finish()
-
- res = await get_content(keyword, sources=["nbnhhsh"])
- if res:
- await nbnhhsh.finish(res)
- else:
- await nbnhhsh.finish("找不到相关的缩写")
diff --git a/src/plugins/what/data_source.py b/src/plugins/what/data_source.py
deleted file mode 100644
index 9e7e009..0000000
--- a/src/plugins/what/data_source.py
+++ /dev/null
@@ -1,121 +0,0 @@
-import re
-import httpx
-from lxml import etree
-from thefuzz import fuzz
-from baike import getBaike
-from urllib.parse import quote
-from typing import List, Tuple, Union
-
-from nonebot.log import logger
-from nonebot.adapters.onebot.v11 import Message, MessageSegment
-
-
-async def get_nbnhhsh(keyword: str) -> Tuple[str, str]:
- url = "https://lab.magiconch.com/api/nbnhhsh/guess"
- headers = {"referer": "https://lab.magiconch.com/nbnhhsh/"}
- data = {"text": keyword}
- async with httpx.AsyncClient() as client:
- resp = await client.post(url=url, headers=headers, data=data)
- res = resp.json()
- title = ""
- result = []
- for i in res:
- if "trans" in i:
- if i["trans"]:
- title = i["name"]
- result.append(f"{i['name']} => {'，'.join(i['trans'])}")
- result = "\n".join(result)
- if fuzz.ratio(title.lower(), keyword.lower()) < 90:
- return "", ""
- return title, result
-
-
-async def get_jiki(keyword: str) -> Tuple[str, Union[str, Message]]:
- search_url = "https://jikipedia.com/search?phrase={}".format(quote(keyword))
- async with httpx.AsyncClient() as client:
- resp = await client.get(url=search_url)
- result = resp.text
-
- if "对不起！小鸡词典暂未收录该词条" in result:
- return "", ""
-
- dom = etree.HTML(result, etree.HTMLParser())
- card_urls = dom.xpath(
- "//div[contains(@class, 'masonry')]/div/div/div/a[contains(@class, 'title-container')]/@href"
- )
- if not card_urls:
- return "", ""
-
- card_url = card_urls[0]
- async with httpx.AsyncClient() as client:
- resp = await client.get(url=card_url)
- result = resp.text
-
- dom = etree.HTML(result, etree.HTMLParser())
- title = dom.xpath(
- "//div[@class='section card-middle']/div[@class='title-container']/div/h1/text()"
- )[0]
- content = dom.xpath(
- "//div[@class='section card-middle']/div[@class='content']/div"
- )[0]
- content = content.xpath("string(.)").strip()
- img_urls = dom.xpath(
- "//div[@class='section card-middle']/div/div/div[@class='show-images']/img/@src"
- )
- if fuzz.ratio(str(title).lower(), keyword.lower()) < 90:
- return "", ""
-
- msg = Message()
- msg.append(title + ":\n---------------\n")
- msg.append(content)
- for img_url in img_urls:
- msg.append(MessageSegment.image(file=img_url))
- return title, msg
-
-
-async def get_baidu(keyword: str) -> Tuple[str, str]:
- content = getBaike(keyword)
- if not content:
- return "", ""
-
- match_obj = re.match(r"(.*?)(（.*?）?)\n(.*)", content)
- if not match_obj:
- return "", ""
-
- title = match_obj.group(1)
- subtitle = match_obj.group(2)
- text = match_obj.group(3)
- if fuzz.ratio(title.lower(), keyword.lower()) < 90:
- return "", ""
-
- msg = title
- if subtitle:
- msg += subtitle
- msg += ":\n---------------\n" + text
- return title, msg
-
-
-sources_func = {"jiki": get_jiki, "baidu": get_baidu, "nbnhhsh": get_nbnhhsh}
-
-
-async def get_content(
- keyword: str, sources=["jiki", "baidu", "nbnhhsh"]
-) -> Union[str, Message]:
- result = ""
- msgs: List[Tuple[str, Union[str, Message]]] = []
- for s in sources:
- try:
- title, msg = await sources_func[s](keyword)
- if title and msg:
- msgs.append((title, msg))
- except Exception as e:
- logger.warning(f"Error in get_content({keyword}) using {s}: {e}")
-
- if len(msgs) == 1:
- result = msgs[0][1]
- elif len(msgs) > 1:
- msgs = sorted(
- msgs, key=lambda m: fuzz.ratio(m[0].lower(), keyword.lower()), reverse=True
- )
- result = msgs[0][1]
- return result
diff --git a/src/plugins/wolfram/__init__.py b/src/plugins/wolfram/__init__.py
index 7beee8b..f923e3c 100644
--- a/src/plugins/wolfram/__init__.py
+++ b/src/plugins/wolfram/__init__.py
@@ -1,13 +1,18 @@
import re
-from nonebot import on_command
+
+from nonebot import on_command, require
+from nonebot.adapters import Message
+from nonebot.matcher import Matcher
from nonebot.params import CommandArg
from nonebot.plugin import PluginMetadata
-from nonebot.adapters.onebot.v11 import Message
+
+require("nonebot_plugin_alconna")
+
+from nonebot_plugin_alconna import UniMessage
from .config import Config
from .data_source import get_wolframalpha_simple, get_wolframalpha_text
-
__plugin_meta__ = PluginMetadata(
name="WolframAlpha",
description="WolframAlpha计算知识引擎",
@@ -23,7 +28,7 @@
@wolfram.handle()
-async def _(msg: Message = CommandArg()):
+async def _(matcher: Matcher, msg: Message = CommandArg()):
text = msg.extract_plain_text().strip()
plaintext = False
@@ -34,13 +39,17 @@ async def _(msg: Message = CommandArg()):
break
text = text.replace("-p", "").replace("--plaintext", "").strip()
if not text:
- await wolfram.finish()
+ await matcher.finish()
if plaintext:
res = await get_wolframalpha_text(text)
else:
res = await get_wolframalpha_simple(text)
+
if not res:
- await wolfram.finish("出错了，请稍后再试")
+ await matcher.finish("出错了，请稍后再试")
- await wolfram.finish(res)
+ if isinstance(res, str):
+ await matcher.finish(res)
+ else:
+ await UniMessage.image(raw=res).send()
diff --git a/src/plugins/wolfram/config.py b/src/plugins/wolfram/config.py
index ef8575c..0c04742 100644
--- a/src/plugins/wolfram/config.py
+++ b/src/plugins/wolfram/config.py
@@ -1,5 +1,9 @@
-from pydantic import BaseModel, Extra
+from nonebot import get_plugin_config
+from pydantic import BaseModel
-class Config(BaseModel, extra=Extra.ignore):
- wolframalpha_appid = ""
+class Config(BaseModel):
+ wolframalpha_appid: str = ""
+
+
+wolframalpha_config = get_plugin_config(Config)
diff --git a/src/plugins/wolfram/data_source.py b/src/plugins/wolfram/data_source.py
index 2ca24fe..b1a1150 100644
--- a/src/plugins/wolfram/data_source.py
+++ b/src/plugins/wolfram/data_source.py
@@ -1,21 +1,19 @@
-import httpx
import itertools
import urllib.parse
+from typing import Optional
+
+import httpx
import wolframalpha
-from nonebot import get_driver
from nonebot.log import logger
-from nonebot.adapters.onebot.v11 import MessageSegment
-
-from .config import Config
-wolframalpha_config = Config.parse_obj(get_driver().config.dict())
+from .config import wolframalpha_config
-async def get_wolframalpha_simple(input, params=(), **kwargs):
- data = dict(
- input=input,
- appid=wolframalpha_config.wolframalpha_appid,
- )
+async def get_wolframalpha_simple(input, params=(), **kwargs) -> Optional[bytes]:
+ data = {
+ "input": input,
+ "appid": wolframalpha_config.wolframalpha_appid,
+ }
data = itertools.chain(params, data.items(), kwargs.items())
query = urllib.parse.urlencode(tuple(data))
url = "https://api.wolframalpha.com/v2/simple?" + query
@@ -23,14 +21,12 @@ async def get_wolframalpha_simple(input, params=(), **kwargs):
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=10)
- result = resp.content
- return MessageSegment.image(result)
+ return resp.content
except Exception as e:
logger.warning(f"Error in get_wolframalpha_simple({input}): {e}")
- return None
-async def get_wolframalpha_text(input, params=(), **kwargs):
+async def get_wolframalpha_text(input, params=(), **kwargs) -> str:
try:
client = wolframalpha.Client(wolframalpha_config.wolframalpha_appid)
res = client.query(input, params, **kwargs)
diff --git a/src/utils/__init__.py b/src/utils/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/utils/plugin_manager.py b/src/utils/plugin_manager.py
new file mode 100644
index 0000000..99ec7a2
--- /dev/null
+++ b/src/utils/plugin_manager.py
@@ -0,0 +1,157 @@
+from enum import IntEnum
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+import yaml
+from nonebot import get_driver
+from nonebot.log import logger
+from nonebot.plugin import Plugin, get_loaded_plugins
+from pydantic import BaseModel, field_serializer
+
+config_path = Path("data/plugin_manager.yml")
+
+
+class ManageType(IntEnum):
+ BLACK = 0
+ WHITE = 1
+
+
+class PluginConfig(BaseModel):
+ mode: int = 7
+ manage_type: ManageType = ManageType.BLACK
+ white_list: List[str] = []
+ black_list: List[str] = []
+
+ @field_serializer("manage_type")
+ def get_eunm_value(self, v: ManageType, info) -> int:
+ return v.value
+
+
+class PluginManager:
+ def __init__(self, path: Path = config_path):
+ self.__path = path
+ self.__plugin_list: Dict[str, PluginConfig] = {}
+ self.__plugins: List[Plugin] = []
+
+ def block(self, plugin_name: str, user_id: str) -> bool:
+ if plugin_name not in self.__plugin_list:
+ return False
+ config = self.__plugin_list[plugin_name]
+ manage_type = config.manage_type
+ if manage_type == ManageType.BLACK and user_id not in config.black_list:
+ config.black_list.append(user_id)
+ if manage_type == ManageType.WHITE and user_id in config.white_list:
+ config.white_list.remove(user_id)
+ self.__dump()
+ return True
+
+ def unblock(self, plugin_name: str, user_id: str) -> bool:
+ if plugin_name not in self.__plugin_list:
+ return False
+ config = self.__plugin_list[plugin_name]
+ manage_type = config.manage_type
+ if manage_type == ManageType.WHITE and user_id not in config.white_list:
+ config.white_list.append(user_id)
+ if manage_type == ManageType.BLACK and user_id in config.black_list:
+ config.black_list.remove(user_id)
+ self.__dump()
+ return True
+
+ def change_mode(self, plugin_name: str, mode: int) -> bool:
+ if plugin_name not in self.__plugin_list:
+ return False
+ if not (0 <= mode <= 7):
+ return False
+ self.__plugin_list[plugin_name].mode = mode
+ self.__dump()
+ return True
+
+ def change_manage_type(self, plugin_name: str, manage_type: ManageType) -> bool:
+ if plugin_name not in self.__plugin_list:
+ return False
+ self.__plugin_list[plugin_name].manage_type = manage_type
+ self.__dump()
+ return True
+
+ def get_config(self, plugin_name: str) -> Optional[PluginConfig]:
+ return self.__plugin_list.get(plugin_name)
+
+ def check(self, plugin_name: str, user_id: str) -> bool:
+ if plugin_name not in self.__plugin_list:
+ return False
+ config = self.__plugin_list[plugin_name]
+ if not (config.mode & 1):
+ return False
+ if config.manage_type == ManageType.BLACK:
+ if user_id in config.black_list:
+ return False
+ return True
+ elif config.manage_type == ManageType.WHITE:
+ if user_id in config.white_list:
+ return True
+ return False
+ return False
+
+ def find(self, keyword: str) -> Optional[str]:
+ for p in self.__plugins[::-1]:
+ names = [p.name.lower()]
+ if metadata := p.metadata:
+ names.append(metadata.name.lower())
+ unique_name = metadata.extra.get("unique_name")
+ if not unique_name:
+ unique_name = (
+ p.name.replace("nonebot_plugin_", "")
+ .replace("nonebot-plugin-", "")
+ .replace("nonebot_", "")
+ .replace("nonebot-", "")
+ )
+ names.append(str(unique_name).lower())
+ if keyword.lower() in names:
+ return p.name
+
+ def init(self):
+ raw_list: Dict[str, Any] = {}
+ if self.__path.exists():
+ with self.__path.open("r", encoding="utf-8") as f:
+ try:
+ raw_list = yaml.safe_load(f)
+ except Exception:
+ logger.warning("插件列表解析失败")
+ raise
+ try:
+ plugin_list = {
+ name: PluginConfig.model_validate(config)
+ for name, config in raw_list.items()
+ }
+ except Exception:
+ logger.warning("插件列表解析失败")
+ raise
+
+ self.__plugins = list(get_loaded_plugins())
+ for plugin in self.__plugins:
+ plugin_config = PluginConfig()
+ if not plugin.metadata or not plugin.matcher:
+ plugin_config.mode &= 3
+ self.__plugin_list[plugin.name] = plugin_config
+
+ self.__plugin_list.update(plugin_list)
+ self.__dump()
+
+ def __dump(self):
+ self.__path.parent.mkdir(parents=True, exist_ok=True)
+ plugin_list = {
+ name: config.model_dump() for name, config in self.__plugin_list.items()
+ }
+ with self.__path.open("w", encoding="utf-8") as f:
+ yaml.dump(plugin_list, f, allow_unicode=True)
+
+
+plugin_manager = PluginManager()
+
+driver = get_driver()
+
+
+@driver.on_startup
+def _():
+ plugin_manager.init()
+ logger.info("插件管理器初始化完成")
