🎨 为邀请BOT的条件添加多种选择

为了机器人防止被滥用,或部署的服务器资源不足只能留在部分群使用,或者是私用搭建所写的一个功能
别问为啥懒得写单独认证
懒了.jpg
This commit is contained in:
洛水居室 2022-10-16 18:15:23 +08:00
parent b27f8af9a4
commit 4a0168d311
No known key found for this signature in database
GPG Key ID: C9DE87DA724B88FC
3 changed files with 54 additions and 12 deletions

View File

@ -25,6 +25,9 @@ ADMINS=[{ "username": "", "user_id": -1 }]
# 文章推送群组 可选配置项 # 文章推送群组 可选配置项
# CHANNELS=[{ "name": "", "chat_id": 1}] # CHANNELS=[{ "name": "", "chat_id": 1}]
# 是否允许机器人邀请到其他群 默认不允许 如果允许 可以允许全部人或有认证选项 可选配置项
# JOIN_GROUPS = "NO_ALLOW"
# 群验证功能 可选配置项 # 群验证功能 可选配置项
# VERIFY_GROUPS=[] # VERIFY_GROUPS=[]

View File

@ -1,3 +1,4 @@
from enum import Enum
from pathlib import Path from pathlib import Path
from typing import ( from typing import (
List, List,
@ -11,11 +12,17 @@ from pydantic import BaseModel, BaseSettings, validator
from utils.const import PROJECT_ROOT from utils.const import PROJECT_ROOT
__all__ = ["BotConfig", "config"] __all__ = ["BotConfig", "config", "JoinGroups"]
dotenv.load_dotenv() dotenv.load_dotenv()
class JoinGroups(str, Enum):
NO_ALLOW = "NO_ALLOW"
ALLOW_AUTH_USER = "ALLOW_AUTH_USER"
ALLOW_ALL = "ALLOW_ALL"
class BotConfig(BaseSettings): class BotConfig(BaseSettings):
debug: bool = False debug: bool = False
@ -38,6 +45,7 @@ class BotConfig(BaseSettings):
channels: List["ConfigChannel"] = [] channels: List["ConfigChannel"] = []
admins: List["ConfigUser"] = [] admins: List["ConfigUser"] = []
verify_groups: List[Union[int, str]] = [] verify_groups: List[Union[int, str]] = []
join_groups: Optional[JoinGroups] = JoinGroups.NO_ALLOW
logger_width: int = 180 logger_width: int = 180
logger_log_path: str = "./logs" logger_log_path: str = "./logs"

View File

@ -1,34 +1,65 @@
from telegram import Update from telegram import Update
from telegram.ext import CallbackContext from telegram.ext import CallbackContext
from core.admin import BotAdminService from core.admin.services import BotAdminService
from core.config import config, JoinGroups
from core.cookies.error import CookiesNotFoundError
from core.cookies.services import CookiesService
from core.plugin import Plugin, handler from core.plugin import Plugin, handler
from core.user.error import UserNotFoundError
from core.user.services import UserService
from utils.log import logger from utils.log import logger
class BotJoiningGroupsVerification(Plugin): class BotJoiningGroupsVerification(Plugin):
def __init__(self, bot_admin_service: BotAdminService = None): def __init__(
self,
bot_admin_service: BotAdminService = None,
user_service: UserService = None,
cookies_service: CookiesService = None,
):
self.cookies_service = cookies_service
self.user_service = user_service
self.bot_admin_service = bot_admin_service self.bot_admin_service = bot_admin_service
@handler.message.new_chat_members(priority=1) @handler.message.new_chat_members(priority=1)
async def new_member(self, update: Update, context: CallbackContext) -> None: async def new_member(self, update: Update, context: CallbackContext) -> None:
if config.join_groups == JoinGroups.ALLOW_ALL:
return None
message = update.effective_message message = update.effective_message
chat = message.chat chat = message.chat
from_user = message.from_user from_user = message.from_user
quit_status = False
for new_chat_members_user in message.new_chat_members: for new_chat_members_user in message.new_chat_members:
if new_chat_members_user.id == context.bot.id: if new_chat_members_user.id == context.bot.id:
logger.info(f"有人邀请BOT进入群 {chat.title}[{chat.id}]") logger.info(f"有人邀请BOT进入群 {chat.title}[{chat.id}]")
quit_status = True quit_status = True
if from_user is not None: if from_user is not None:
logger.info(f"用户 {from_user.full_name}[{from_user.id}] 在群 {chat.title}[{chat.id}] 邀请BOT") logger.info(f"用户 {from_user.full_name}[{from_user.id}] 在群 {chat.title}[{chat.id}] 邀请BOT")
admin_list = await self.bot_admin_service.get_admin_list() if config.join_groups == JoinGroups.NO_ALLOW:
if from_user.id in admin_list: try:
await context.bot.send_message(message.chat_id, "感谢邀请小派蒙到本群!请使用 /help 查看咱已经学会的功能。") admin_list = await self.bot_admin_service.get_admin_list()
quit_status = False if from_user.id in admin_list:
quit_status = False
else:
logger.warning("不是管理员邀请!退出群聊")
except Exception as exc:
logger.error(f"获取信息出现错误 {repr(exc)}")
elif config.join_groups == JoinGroups.ALLOW_AUTH_USER:
try:
user_info = await self.user_service.get_user_by_id(chat.id)
await self.cookies_service.get_cookies(user_info.user_id, user_info.region)
except (UserNotFoundError, CookiesNotFoundError):
logger.warning(f"用户 {from_user.full_name}[{from_user.id}] 邀请请求被拒绝")
except Exception as exc:
logger.error(f"获取信息出现错误 {repr(exc)}")
else:
quit_status = False
else:
quit_status = True
else: else:
logger.info(f"未知用户 在群 {chat.title}[{chat.id}] 邀请BOT") logger.info(f"未知用户 在群 {chat.title}[{chat.id}] 邀请BOT")
if quit_status: if quit_status:
logger.warning("不是管理员邀请!退出群聊。") await context.bot.send_message(message.chat_id, "派蒙不想进去!不是旅行者的邀请!")
await context.bot.send_message(message.chat_id, "派蒙不想进去!不是旅行者的邀请!") await context.bot.leave_chat(chat.id)
await context.bot.leave_chat(chat.id) else:
await context.bot.send_message(message.chat_id, "感谢邀请小派蒙到本群!请使用 /help 查看咱已经学会的功能。")