Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2025-12-15 06:12:53 +08:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

feat✨: add dialogue

Browse Source
This commit is contained in:
HibiKier 2024-03-27 20:09:30 +08:00
parent d37280b296
commit d96fd8191d
7 changed files with 237 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
zhenxun/builtin_plugins/scheduler/auto_update_group.py
View File

@ -4,7 +4,7 @@ from nonebot_plugin_apscheduler import scheduler
from zhenxun.models.friend_user import FriendUser
from zhenxun.models.group_console import GroupConsole
from zhenxun.services.log import logger
from zhenxun.utils.platform import PlatformManage
from zhenxun.utils.platform import PlatformUtils
# 自动更新群组信息
@ -18,7 +18,7 @@ async def _():
_used_group = []
for bot in bots.values():
try:
await PlatformManage.update_group(bot)
await PlatformUtils.update_group(bot)
except Exception as e:
logger.error(f"Bot: {bot.self_id} 自动更新群组信息", e=e)
logger.info("自动更新群组成员信息成功...")
@ -34,7 +34,7 @@ async def _():
bots = nonebot.get_bots()
for bot in bots.values():
try:
await PlatformManage.update_friend(bot)
await PlatformUtils.update_friend(bot)
except Exception as e:
logger.error(f"自动更新好友信息错误", "自动更新好友", e=e)
logger.info("自动更新好友信息成功...")

6
zhenxun/builtin_plugins/superuser/broadcast/_data_source.py
View File

@ -11,7 +11,7 @@ from nonebot_plugin_session import EventSession
from zhenxun.models.group_console import GroupConsole
from zhenxun.services.log import logger
from zhenxun.utils.platform import PlatformManage
from zhenxun.utils.platform import PlatformUtils
class BroadcastManage:
@ -36,7 +36,7 @@ class BroadcastManage:
message_list.append(Image(msg.url))
elif isinstance(msg, alc.Text):
message_list.append(Text(msg.text))
group_list, _ = await PlatformManage.get_group_list(bot)
group_list, _ = await PlatformUtils.get_group_list(bot)
if group_list:
error_count = 0
for group in group_list:
@ -44,7 +44,7 @@ class BroadcastManage:
if not await GroupConsole.is_block_task(
group.group_id, "broadcast", group.channel_id
):
target = PlatformManage.get_target(
target = PlatformUtils.get_target(
bot, None, group.group_id, group.channel_id
)
if target:

6
zhenxun/builtin_plugins/superuser/update_fg_info.py
View File

@ -9,7 +9,7 @@ from nonebot_plugin_session import EventSession
from zhenxun.configs.utils import PluginExtraData
from zhenxun.services.log import logger
from zhenxun.utils.enum import PluginType
from zhenxun.utils.platform import PlatformManage
from zhenxun.utils.platform import PlatformUtils
__plugin_meta__ = PluginMetadata(
name="更新群组/好友信息",
@ -54,7 +54,7 @@ async def _(
arparma: Arparma,
):
try:
num = await PlatformManage.update_group(bot)
num = await PlatformUtils.update_group(bot)
logger.info(
f"更新群聊信息完成，共更新了 {num} 个群组的信息!",
arparma.header_result,
@ -75,7 +75,7 @@ async def _(
arparma: Arparma,
):
try:
num = await PlatformManage.update_friend(bot, session.platform)
num = await PlatformUtils.update_friend(bot, session.platform)
logger.info(
f"更新好友信息完成，共更新了 {num} 个好友的信息!",
arparma.header_result,

10
zhenxun/plugins/black_word/utils.py
View File

@ -12,7 +12,7 @@ from zhenxun.models.ban_console import BanConsole
from zhenxun.models.group_member_info import GroupInfoUser
from zhenxun.services.log import logger
from zhenxun.utils.http_utils import AsyncHttpx
from zhenxun.utils.platform import PlatformManage
from zhenxun.utils.platform import PlatformUtils
from zhenxun.utils.utils import cn2py
from .model import BlackWord
@ -173,7 +173,7 @@ async def _add_user_black_word(
plant_text=message,
black_word=black_word,
punish_level=punish_level,
platform=PlatformManage.get_platform(bot),
platform=PlatformUtils.get_platform(bot),
)
logger.info(
f"已将 USER {user_id} GROUP {group_id} 添加至黑名单词汇记录 Black_word：{black_word} Plant_text：{message}"
@ -325,7 +325,7 @@ async def _get_punish(
return ban_4_duration
# 口头警告
elif id_ == 5:
await PlatformManage.send_message(bot, user_id, group_id, warning_result)
await PlatformUtils.send_message(bot, user_id, group_id, warning_result)
logger.info(f"BlackWord 口头警告 USER {user_id}")
return warning_result
return None
@ -341,9 +341,9 @@ async def send_msg(bot: Bot, user_id: str, group_id: str | None, message: str):
message: message
"""
if not user_id:
platform = PlatformManage.get_platform(bot)
platform = PlatformUtils.get_platform(bot)
user_id = bot.config.platform_superusers[platform][0]
await PlatformManage.send_message(bot, user_id, group_id, message)
await PlatformUtils.send_message(bot, user_id, group_id, message)
async def check_text(text: str) -> bool:

162
zhenxun/plugins/dialogue/__init__.py Normal file
View File

@ -0,0 +1,162 @@
import nonebot
from nonebot import on_command
from nonebot.adapters import Bot
from nonebot.permission import SUPERUSER
from nonebot.plugin import PluginMetadata
from nonebot_plugin_alconna import At as alcAt
from nonebot_plugin_alconna import Target
from nonebot_plugin_alconna import Text as alcText
from nonebot_plugin_alconna import UniMsg
from nonebot_plugin_saa import Text
from nonebot_plugin_session import EventSession
from nonebot_plugin_userinfo import EventUserInfo, UserInfo
from zhenxun.configs.utils import PluginExtraData
from zhenxun.models.group_console import GroupConsole
from zhenxun.services.log import logger
from zhenxun.utils.platform import PlatformUtils
from ._data_source import DialogueManage
__plugin_meta__ = PluginMetadata(
name="联系管理员",
description="跨越空间与时间跟管理员对话",
usage="""
[滴滴滴]/滴滴滴- ?[文本] ?[图片]
示例：滴滴滴- 我喜欢你
超级管理员额外命令
/t: 查看当前存储的消息
/t [user_id] [group_id] [文本]: 在group回复指定用户
/t [user_id] [文本]: 私聊用户
/t -1 [group_id] [文本]: 在group内发送消息
/t [id] [文本]: 回复指定id的对话，id在 /t 中获取
示例：/t 73747222 32848432 你好啊
示例：/t 73747222 你好不好
示例：/t -1 32848432 我不太好
示例：/t 0 我收到你的话了
""".strip(),
extra=PluginExtraData(
author="HibiKier", version="0.1", menu_type="联系管理员"
).dict(),
)
config = nonebot.get_driver().config
_dialogue_matcher = on_command("滴滴滴-", priority=5, block=True)
_reply_matcher = on_command("/t", priority=1, permission=SUPERUSER, block=True)
@_dialogue_matcher.handle()
async def _(
bot: Bot,
message: UniMsg,
session: EventSession,
user_info: UserInfo = EventUserInfo(),
):
if session.id1:
message[0] = alcText(str(message[0]).replace("滴滴滴-", "", 1))
platform = PlatformUtils.get_platform(bot)
try:
superuser_id = config.platform_superusers["qq"][0]
if platform == "dodo":
superuser_id = config.platform_superusers["dodo"][0]
if platform == "kaiheila":
superuser_id = config.platform_superusers["kaiheila"][0]
if platform == "discord":
superuser_id = config.platform_superusers["discord"][0]
except IndexError:
await Text("管理员失联啦...").finish()
if not superuser_id:
await Text("管理员失联啦...").finish()
uname = user_info.user_displayname or user_info.user_name
group_name = ""
gid = session.id3 or session.id2
if gid:
if g := await GroupConsole.get(group_id=gid):
group_name = g.group_name
logger.info(
f"发送消息至{platform}管理员: {message}", "滴滴滴-", session=session
)
message.insert(0, "消息:\n")
if gid:
message.insert(0, f"群组: {group_name}({gid})\n")
message.insert(0, f"昵称: {uname}({session.id1})\n")
message.insert(0, f"Id: {DialogueManage._index}\n")
message.insert(0, "*****一份交流报告*****\n")
DialogueManage.add(uname, session.id1, gid, group_name, message, platform)
await message.send(bot=bot, target=Target(superuser_id, private=True))
await Text("已成功发送给管理员啦!").send(reply=True)
else:
await Text("用户id为空...").send()
@_reply_matcher.handle()
async def _(
bot: Bot,
message: UniMsg,
session: EventSession,
user_info: UserInfo = EventUserInfo(),
):
message[0] = alcText(str(message[0]).replace("/t", "", 1).strip())
if session.id1:
msg = message.extract_plain_text()
if not msg:
platform = PlatformUtils.get_platform(bot)
data = DialogueManage._data
if not data:
await Text("暂无待回复消息...").finish()
if platform:
data = [data[d] for d in data if data[d].platform == platform]
for d in data:
await d.message.send(
bot=bot, target=Target(session.id1, private=True)
)
else:
msg = msg.split()
group_id = ""
user_id = ""
if msg[0].isdigit():
if len(msg[0]) < 4:
_id = int(msg[0])
if _id >= 0:
if model := DialogueManage.get(_id):
user_id = model.user_id
group_id = model.group_id
else:
return Text("未获取此id数据").finish()
message[0] = alcText(" ".join(str(message[0]).split(" ")[1:]))
else:
user_id = 0
if msg[1].isdigit():
group_id = msg[1]
message[0] = alcText(
" ".join(str(message[0]).split(" ")[2:])
)
else:
await Text("群组id错误...").finish(at_sender=True)
DialogueManage.remove(_id)
else:
user_id = msg[0]
if msg[1].isdigit() and len(msg[1]) > 5:
group_id = msg[1]
message[0] = alcText(" ".join(str(message[0]).split(" ")[2:]))
else:
group_id = 0
message[0] = alcText(" ".join(str(message[0]).split(" ")[1:]))
else:
await Text("参数错误...").finish(at_sender=True)
if group_id:
if user_id:
message.insert(0, alcAt("user", user_id))
message.insert(1, "\n管理员回复\n=======\n")
await message.send(Target(group_id), bot)
await Text("消息发送成功!").finish(at_sender=True)
elif user_id:
await message.send(Target(user_id, private=True), bot)
await Text("消息发送成功!").finish(at_sender=True)
else:
await Text("群组id与用户id为空...").send()
else:
await Text("用户id为空...").send()

55
zhenxun/plugins/dialogue/_data_source.py Normal file
View File

@ -0,0 +1,55 @@
from typing import Dict
from nonebot_plugin_alconna import UniMsg
from pydantic import BaseModel
class DialogueData(BaseModel):
name: str
"""用户名称"""
user_id: str
"""用户id"""
group_id: str | None
"""群组id"""
group_name: str | None
"""群组名称"""
message: UniMsg
"""UniMsg"""
platform: str | None
"""平台"""
class DialogueManage:
_data: Dict[int, DialogueData] = {}
_index = 0
@classmethod
def add(
cls,
name: str,
uid: str,
gid: str | None,
group_name: str | None,
message: UniMsg,
platform: str | None,
):
cls._data[cls._index] = DialogueData(
name=name,
user_id=uid,
group_id=gid,
group_name=group_name,
message=message,
platform=platform,
)
cls._index += 1
@classmethod
def remove(cls, index: int):
if index in cls._data:
del cls._data[index]
@classmethod
def get(cls, k: int):
return cls._data.get(k)

10
zhenxun/utils/platform.py
View File

@ -25,7 +25,7 @@ from zhenxun.models.group_console import GroupConsole
from zhenxun.services.log import logger
class PlatformManage:
class PlatformUtils:
@classmethod
async def send_message(
@ -97,6 +97,8 @@ class PlatformManage:
return "dodo"
if isinstance(bot, KaiheilaBot):
return "kaiheila"
if isinstance(bot, DiscordBot):
return "discord"
return None
@classmethod
@ -308,9 +310,9 @@ async def broadcast_group(
_used_group = []
for _bot in bot_list:
try:
if platform and platform != PlatformManage.get_platform(_bot):
if platform and platform != PlatformUtils.get_platform(_bot):
continue
group_list, _ = await PlatformManage.get_group_list(_bot)
group_list, _ = await PlatformUtils.get_group_list(_bot)
if group_list:
for group in group_list:
key = f"{group.group_id}:{group.channel_id}"
@ -335,7 +337,7 @@ async def broadcast_group(
)
if is_continue:
continue
target = PlatformManage.get_target(
target = PlatformUtils.get_target(
_bot, None, group.group_id, group.channel_id
)
if target: