Andrey K. Choi
c16ec54891
added: dictionary support for censore message/user management with dict triggers
69 lines
3.7 KiB
Python
69 lines
3.7 KiB
Python
from telegram import Update
|
|
from telegram.constants import ChatType
|
|
from telegram.ext import ContextTypes
|
|
from sqlalchemy import select, func
|
|
from app.db.session import get_session
|
|
from app.db.models import ChatSecurity, SecurityPolicy, PolicyDictionaryLink, SpamDictionary, DictionaryEntry, ModerationLog
|
|
|
|
async def mod_status_cmd(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
|
chat = update.effective_chat
|
|
if chat.type not in (ChatType.GROUP, ChatType.SUPERGROUP):
|
|
await update.effective_message.reply_text("Команду /mod_status нужно запускать в группе.")
|
|
return
|
|
|
|
# Только админам группы
|
|
try:
|
|
m = await ctx.bot.get_chat_member(chat.id, update.effective_user.id)
|
|
if m.status not in ("administrator","creator"):
|
|
return
|
|
except Exception:
|
|
return
|
|
|
|
with get_session() as s:
|
|
cs = s.query(ChatSecurity).filter_by(chat_id=chat.id).first()
|
|
if not cs:
|
|
await update.effective_message.reply_text("Политика не привязана. Откройте /security и нажмите «Привязать к этому чату».")
|
|
return
|
|
p = s.get(SecurityPolicy, cs.policy_id)
|
|
if not p:
|
|
await update.effective_message.reply_text("Политика не найдена (policy_id устарел). Перепривяжите через /security.")
|
|
return
|
|
|
|
# словари, доступные через links (глобальные не считаем, просто ориентир)
|
|
linked_dicts = (
|
|
s.query(SpamDictionary)
|
|
.join(PolicyDictionaryLink, PolicyDictionaryLink.dictionary_id == SpamDictionary.id)
|
|
.filter(PolicyDictionaryLink.policy_id == p.id)
|
|
.all()
|
|
)
|
|
rules_count = 0
|
|
if linked_dicts:
|
|
d_ids = [d.id for d in linked_dicts]
|
|
rules_count = s.query(func.count(DictionaryEntry.id)).filter(DictionaryEntry.dictionary_id.in_(d_ids)).scalar() or 0
|
|
|
|
# блокировки за 15 минут
|
|
from datetime import datetime, timedelta
|
|
since = datetime.utcnow() - timedelta(minutes=15)
|
|
blocked_15m = (s.query(func.count(ModerationLog.id))
|
|
.filter(ModerationLog.chat_id == chat.id,
|
|
ModerationLog.created_at >= since,
|
|
ModerationLog.action.in_(("delete","warn","timeout","ban")))
|
|
.scalar() or 0)
|
|
|
|
# права бота
|
|
bot_member = await ctx.bot.get_chat_member(chat.id, ctx.bot.id)
|
|
can_delete = getattr(bot_member, "can_delete_messages", False)
|
|
can_restrict = getattr(bot_member, "can_restrict_members", False)
|
|
|
|
txt = (
|
|
f"Чат: {chat.title or chat.id}\n"
|
|
f"Модерация: {'ON' if cs.enabled else 'OFF'} (policy: {p.name})\n"
|
|
f"Категории: Profanity={'ON' if p.block_profanity else 'OFF'}, Spam={'ON' if p.block_spam else 'OFF'}, Adult={'ON' if p.block_adult else 'OFF'}, Scam={'ON' if p.block_scam else 'OFF'}\n"
|
|
f"Лимиты: links≤{p.max_links}, mentions≤{p.max_mentions}, rate={p.user_msg_per_minute}/min, duplicate={p.duplicate_window_seconds}s\n"
|
|
f"Права бота: delete={'yes' if can_delete else 'no'}, restrict={'yes' if can_restrict else 'no'}\n"
|
|
f"Привязанных словарей: {len(linked_dicts)} (правил: {rules_count})\n"
|
|
f"Заблокировано за 15 мин: {blocked_15m}\n"
|
|
"Если блокировок 0: проверьте privacy mode, права бота и что mod=ON."
|
|
)
|
|
await update.effective_message.reply_text(txt)
|