ngfchl
80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
"""群管理工具：权限校验、目标用户解析"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from telegram import Update
|
|
from telegram.constants import ChatMemberStatus
|
|
from telegram.ext import ContextTypes
|
|
|
|
import config
|
|
from db.models import Action, Ban, Message
|
|
from services.message_utils import schedule_delete
|
|
|
|
logger = logging.getLogger("spam_guard")
|
|
|
|
ADMIN_STATUSES = {ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.OWNER}
|
|
|
|
|
|
async def is_admin(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
|
|
"""判断命令发送者是否为群管理员或配置中的超级管理员。"""
|
|
user = update.effective_user
|
|
chat = update.effective_chat
|
|
if not user or not chat:
|
|
return False
|
|
if user.id in config.ADMIN_USER_IDS:
|
|
return True
|
|
try:
|
|
member = await context.bot.get_chat_member(chat.id, user.id)
|
|
return member.status in ADMIN_STATUSES
|
|
except Exception as e:
|
|
logger.warning(f"管理员校验失败: user={user.id}, chat={chat.id}, err={e}")
|
|
return False
|
|
|
|
|
|
async def require_admin(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
|
|
"""命令权限守卫。返回 True 表示允许继续。"""
|
|
if await is_admin(update, context):
|
|
return True
|
|
reply = await update.message.reply_text("⛔ 只有群管理员可以使用这个命令")
|
|
schedule_delete(update.message)
|
|
schedule_delete(reply)
|
|
return False
|
|
|
|
|
|
async def resolve_target_user(update: Update, args: list[str]) -> tuple[int | None, str]:
|
|
"""解析管理命令目标用户。
|
|
|
|
支持：
|
|
1. 回复某条消息后执行命令
|
|
2. /cmd 用户ID
|
|
3. /cmd @username，从历史消息/操作/封禁记录中反查 user_id
|
|
"""
|
|
# 回复消息优先，最准确
|
|
if update.message and update.message.reply_to_message and update.message.reply_to_message.from_user:
|
|
u = update.message.reply_to_message.from_user
|
|
username = u.username or u.first_name or str(u.id)
|
|
return u.id, username
|
|
|
|
if not args:
|
|
return None, ""
|
|
|
|
target = args[0].strip().lstrip("@")
|
|
if target.isdigit():
|
|
return int(target), target
|
|
|
|
# Telegram Bot API 不能直接用 username 查群成员，只能从历史记录里反查
|
|
msg = await Message.filter(username=target).order_by("-created_at").first()
|
|
if msg:
|
|
return msg.user_id, target
|
|
|
|
action = await Action.filter(username=target, user_id__not_isnull=True).order_by("-created_at").first()
|
|
if action:
|
|
return action.user_id, target
|
|
|
|
ban = await Ban.filter(username=target).order_by("-created_at").first()
|
|
if ban:
|
|
return ban.user_id, target
|
|
|
|
return None, target
|