78 lines
3.1 KiB
Python
78 lines
3.1 KiB
Python
import asyncio
|
||
from telegram import Update
|
||
from telegram.constants import ChatMemberStatus, ChatType, ParseMode
|
||
from telegram.ext import ContextTypes
|
||
from telegram.error import Forbidden
|
||
from app.bot.messages import (
|
||
JOIN_DM_GROUP, JOIN_DM_CHANNEL, JOIN_PUBLIC_WITH_ID, NEED_START_DM
|
||
)
|
||
|
||
TTL_SEC = 30 # через столько секунд удаляем публичную подсказку
|
||
|
||
async def _auto_delete(ctx: ContextTypes.DEFAULT_TYPE, chat_id: int, message_id: int, delay: int = TTL_SEC):
|
||
try:
|
||
await asyncio.sleep(delay)
|
||
await ctx.bot.delete_message(chat_id=chat_id, message_id=message_id)
|
||
except Exception:
|
||
pass
|
||
|
||
async def on_my_chat_member(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||
"""
|
||
При добавлении/повышении прав.
|
||
1) Пробуем DM актёру (my_chat_member.from_user) с chat_id и инструкцией.
|
||
2) Если DM не вышел (нет from_user или нет Start/Forbidden) — пишем в чат
|
||
подсказку с chat_id и удаляем её через TTL_SEC.
|
||
"""
|
||
mcm = update.my_chat_member
|
||
if not mcm:
|
||
return
|
||
|
||
chat = mcm.chat
|
||
new_status = mcm.new_chat_member.status
|
||
if new_status not in (ChatMemberStatus.MEMBER, ChatMemberStatus.ADMINISTRATOR):
|
||
return
|
||
|
||
title = chat.title or str(chat.id)
|
||
chat_id = chat.id
|
||
|
||
# 1) Пытаемся отправить DM тому, кто совершил действие
|
||
actor = getattr(mcm, "from_user", None)
|
||
dm_sent = False
|
||
if actor:
|
||
try:
|
||
if chat.type in (ChatType.GROUP, ChatType.SUPERGROUP):
|
||
await ctx.bot.send_message(
|
||
actor.id, JOIN_DM_GROUP.format(title=title, chat_id=chat_id),
|
||
parse_mode=ParseMode.MARKDOWN
|
||
)
|
||
elif chat.type == ChatType.CHANNEL:
|
||
await ctx.bot.send_message(
|
||
actor.id, JOIN_DM_CHANNEL.format(title=title, chat_id=chat_id),
|
||
parse_mode=ParseMode.MARKDOWN
|
||
)
|
||
dm_sent = True
|
||
except Forbidden:
|
||
# пользователь не нажал Start — подсказка про Start
|
||
try:
|
||
await ctx.bot.send_message(actor.id, NEED_START_DM)
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
if dm_sent:
|
||
return
|
||
|
||
# 2) DM не удался — публикуем в чат краткий хинт с chat_id, удаляем через TTL
|
||
# (для каналов сработает только если бот уже админ и может постить)
|
||
try:
|
||
msg = await ctx.bot.send_message(
|
||
chat_id=chat_id,
|
||
text=JOIN_PUBLIC_WITH_ID.format(chat_id=chat_id, ttl=TTL_SEC),
|
||
parse_mode=ParseMode.MARKDOWN
|
||
)
|
||
ctx.application.create_task(_auto_delete(ctx, chat_id, msg.message_id, delay=TTL_SEC))
|
||
except Exception:
|
||
# Если и сюда не можем — увы, остаётся ручной путь: /id, /add_group и ЛС
|
||
pass
|