Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2026-10-07 21:00:21 +08:00
Code Issues Actions 8 Packages Projects Releases Wiki Activity

✨ 部分功能适配QQ BOT (#1701)

Browse Source
This commit is contained in:
HibiKier
2024-10-18 18:57:55 +08:00
committed by GitHub
parent 7567bfb732
commit 3d3a2a5f7f
29 changed files with 528 additions and 310 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zhenxun/utils/common_utils.py
+19 -3
View File
@@ -1,3 +1,6 @@
from nonebot.adapters import Bot
from nonebot_plugin_uninfo import Uninfo, Session, SupportScope, get_interface
from zhenxun.services.log import logger
from zhenxun.configs.config import BotConfig
from zhenxun.models.task_info import TaskInfo
@@ -6,9 +9,10 @@ from zhenxun.models.group_console import GroupConsole
class CommonUtils:
@classmethod
async def task_is_block(cls, module: str, group_id: str | None) -> bool:
async def task_is_block(
cls, session: Uninfo | Bot, module: str, group_id: str | None = None
) -> bool:
"""判断被动技能是否可以发送
参数:
@@ -18,6 +22,19 @@ class CommonUtils:
返回:
bool: 是否可以发送
"""
if isinstance(session, Bot):
if interface := get_interface(session):
info = interface.basic_info()
if info["scope"] == SupportScope.qq_api:
logger.info("q官bot放弃所有被动技能发言...")
"""q官bot放弃所有被动技能发言"""
return False
if session.scene == SupportScope.qq_api:
"""q官bot放弃所有被动技能发言"""
logger.info("q官bot放弃所有被动技能发言...")
return False
if not group_id and isinstance(session, Session):
group_id = session.group.id if session.group else None
if task := await TaskInfo.get_or_none(module=module):
"""被动全局状态"""
if not task.status:
@@ -39,7 +56,6 @@ class CommonUtils:
class SqlUtils:
@classmethod
def random(cls, query, limit: int = 1) -> str:
db_class_name = BotConfig.get_sql_type()
zhenxun/utils/depends/__init__.py
+5 -7
View File
@@ -1,10 +1,10 @@
from typing import Any
from nonebot.internal.params import Depends
from nonebot.matcher import Matcher
from nonebot.params import Command
from nonebot.matcher import Matcher
from nonebot_plugin_uninfo import Uninfo
from nonebot.internal.params import Depends
from nonebot_plugin_session import EventSession
from nonebot_plugin_userinfo import EventUserInfo, UserInfo
from zhenxun.configs.config import Config
from zhenxun.utils.message import MessageUtils
@@ -49,10 +49,8 @@ def UserName():
用户名称
"""
async def dependency(user_info: UserInfo = EventUserInfo()):
return (
user_info.user_displayname or user_info.user_remark or user_info.user_name
) or ""
async def dependency(user_info: Uninfo):
return user_info.user.nick or user_info.user.name or ""
return Depends(dependency)
zhenxun/utils/http_utils.py
+5
View File
@@ -202,6 +202,11 @@ class AsyncHttpx:
**kwargs,
)
@classmethod
async def get_content(cls, url: str, **kwargs) -> bytes | None:
res = await cls.get(url, **kwargs)
return res.content if res and res.status_code == 200 else None
@classmethod
async def download_file(
cls,
zhenxun/utils/platform.py
+47 -25
View File
@@ -7,20 +7,24 @@ import nonebot
from pydantic import BaseModel
from nonebot.adapters import Bot
from nonebot.utils import is_coroutine_callable
from nonebot_plugin_uninfo import get_interface
from nonebot_plugin_alconna import SupportScope
from nonebot.adapters.dodo import Bot as DodoBot
from nonebot.adapters.onebot.v11 import Bot as v11Bot
from nonebot.adapters.onebot.v12 import Bot as v12Bot
from nonebot_plugin_uninfo import Uninfo, get_interface
from nonebot.adapters.kaiheila import Bot as KaiheilaBot
from nonebot_plugin_alconna.uniseg import Target, Receipt, UniMessage
from zhenxun.services.log import logger
from zhenxun.configs.config import BotConfig
from zhenxun.utils.message import MessageUtils
from zhenxun.utils.http_utils import AsyncHttpx
from zhenxun.models.friend_user import FriendUser
from zhenxun.utils.exception import NotFindSuperuser
from zhenxun.models.group_console import GroupConsole
driver = nonebot.get_driver()
class UserData(BaseModel):
name: str
@@ -40,6 +44,18 @@ class UserData(BaseModel):
class PlatformUtils:
@classmethod
def is_qbot(cls, session: Uninfo) -> bool:
"""判断bot是否为qq官bot
参数:
session: Uninfo
返回:
bool: 是否为官bot
"""
return session.scope == SupportScope.qq_api
@classmethod
async def ban_user(cls, bot: Bot, user_id: str, group_id: str, duration: int):
"""禁言
@@ -244,30 +260,27 @@ class PlatformUtils:
return None
@classmethod
async def get_user_avatar(cls, user_id: str, platform: str) -> bytes | None:
async def get_user_avatar(
cls, user_id: str, platform: str, appid: str | None = None
) -> bytes | None:
"""快捷获取用户头像
参数:
user_id: 用户id
platform: 平台
"""
url = None
if platform == "qq":
url = f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=160"
async with httpx.AsyncClient() as client:
for _ in range(3):
try:
return (await client.get(url)).content
except Exception:
logger.error(
"获取用户头像错误",
"Util",
target=user_id,
platform=platform,
)
return None
if user_id.isdigit():
url = f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=160"
else:
url = f"https://q.qlogo.cn/qqapp/{appid}/{user_id}/100"
return await AsyncHttpx.get_content(url) if url else None
@classmethod
def get_user_avatar_url(cls, user_id: str, platform: str) -> str | None:
def get_user_avatar_url(
cls, user_id: str, platform: str, appid: str | None = None
) -> str | None:
"""快捷获取用户头像url
参数:
@@ -275,8 +288,13 @@ class PlatformUtils:
platform: 平台
"""
if platform == "qq":
return f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=160"
return None
return (
f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=160"
if user_id.isdigit()
else f"https://q.qlogo.cn/qqapp/{appid}/{user_id}/100"
)
else:
return None
@classmethod
async def get_group_avatar(cls, gid: str, platform: str) -> bytes | None:
@@ -371,7 +389,7 @@ class PlatformUtils:
return len(create_list)
@classmethod
def get_platform(cls, bot: Bot) -> str | None:
def get_platform(cls, t: Bot | Uninfo) -> str:
"""获取平台
参数:
@@ -380,9 +398,13 @@ class PlatformUtils:
返回:
str | None: 平台
"""
if interface := get_interface(bot):
info = interface.basic_info()
platform = info["scope"].lower()
if isinstance(t, Bot):
if interface := get_interface(t):
info = interface.basic_info()
platform = info["scope"].lower()
return "qq" if platform.startswith("qq") else platform
else:
platform = t.basic["scope"].lower()
return "qq" if platform.startswith("qq") else platform
return "unknown"
@@ -548,7 +570,7 @@ async def broadcast_group(
bot: Bot | list[Bot] | None = None,
bot_id: str | set[str] | None = None,
ignore_group: set[int] | None = None,
check_func: Callable[[str], Awaitable] | None = None,
check_func: Callable[[Bot, str], Awaitable] | None = None,
log_cmd: str | None = None,
platform: Literal["qq", "dodo", "kaiheila"] | None = None,
):
@@ -611,9 +633,9 @@ async def broadcast_group(
is_run = False
if check_func:
if is_coroutine_callable(check_func):
is_run = await check_func(group.group_id)
is_run = await check_func(_bot, group.group_id)
else:
is_run = check_func(group.group_id)
is_run = check_func(_bot, group.group_id)
if not is_run:
logger.debug(
"广播方法检测运行方法为 False, 已跳过...",
zhenxun/utils/rules.py
+14 -7
View File
@@ -1,10 +1,12 @@
from nonebot.internal.rule import Rule
from nonebot.adapters import Bot, Event
from nonebot.permission import SUPERUSER
from nonebot_plugin_session import EventSession, SessionLevel
from nonebot_plugin_uninfo import Uninfo
from nonebot_plugin_session import EventSession
from zhenxun.configs.config import Config
from zhenxun.models.level_user import LevelUser
from zhenxun.utils.platform import PlatformUtils
def admin_check(a: int | str, key: str | None = None) -> Rule:
@@ -19,33 +21,38 @@ def admin_check(a: int | str, key: str | None = None) -> Rule:
Rule: Rule
"""
async def _rule(bot: Bot, event: Event, session: EventSession) -> bool:
async def _rule(bot: Bot, event: Event, session: Uninfo) -> bool:
if await SUPERUSER(bot, event):
return True
if session.id1 and session.id2:
if PlatformUtils.is_qbot(session):
"""官bot接口，放弃所有权限检查"""
return False
if session.id and session.group:
level = a
if isinstance(a, str) and key:
level = Config.get_config(a, key)
if level is not None:
return bool(
await LevelUser.check_level(session.id1, session.id2, int(level))
await LevelUser.check_level(
session.id, session.group.id, int(level)
)
)
return False
return Rule(_rule)
def ensure_group(session: EventSession) -> bool:
def ensure_group(session: Uninfo) -> bool:
"""
是否在群聊中
参数:
session: session
session: Uninfo
返回:
bool: bool
"""
return session.level in [SessionLevel.LEVEL2, SessionLevel.LEVEL3]
return bool(session.group)
def ensure_private(session: EventSession) -> bool: