MyLittleApps/AntiMichaell
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: react with poop before auto-delete

Browse Source
This commit is contained in:
sadtweenk
2026-04-20 16:53:38 +03:00
parent bcd6f3e734
commit eb888c04d6
2 changed files with 101 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
bot.py
View File

@@ -8,12 +8,14 @@ from typing import Optional
from dotenv import load_dotenv
from telethon import TelegramClient, events
from telethon.errors import RPCError
from telethon.tl import functions, types
from telethon.tl.custom.message import Message
BAN_COMMAND = "/вбаннахуй"
UNBAN_COMMAND = "/разбан"
DATA_PATH = Path("data/blocked_users.json")
RECENT_CLEANUP_LIMIT = 50
POOP_REACTION_EMOJI = "💩"
ANIMATION_FRAMES = [
"Запускаю анти-режим.",
@@ -131,11 +133,13 @@ async def animate_ban(message: Message, target_user_id: int, is_new: bool) -> No
async def delete_message_best_effort(
client: TelegramClient, message: Message
) -> bool:
) -> tuple[bool, Optional[str]]:
last_error: Optional[str] = None
try:
await message.delete(revoke=False)
return True
return True, None
except Exception as exc:
last_error = f"{type(exc).__name__}: {exc}"
logging.debug(
"Удаление через message.delete(revoke=False) не сработало: %s", exc
)
@@ -143,8 +147,9 @@ async def delete_message_best_effort(
try:
target = message.input_chat if message.input_chat else message.chat_id
await client.delete_messages(target, [message.id], revoke=False)
return True
return True, None
except Exception as exc:
last_error = f"{type(exc).__name__}: {exc}"
logging.debug(
"Удаление через client.delete_messages(..., revoke=False) не сработало: %s",
exc,
@@ -152,8 +157,9 @@ async def delete_message_best_effort(
try:
await message.delete()
return True
return True, None
except RPCError as exc:
last_error = f"{type(exc).__name__}: {exc}"
logging.warning(
"Не удалось удалить сообщение %s в чате %s: %s",
message.id,
@@ -161,13 +167,57 @@ async def delete_message_best_effort(
exc,
)
except Exception:
last_error = "UnexpectedError"
logging.exception(
"Неожиданная ошибка удаления сообщения %s в чате %s.",
message.id,
message.chat_id,
)
return False
return False, last_error
async def react_with_poop_best_effort(client: TelegramClient, message: Message) -> None:
if message.id is None:
return
try:
peer = message.input_chat if message.input_chat else message.chat_id
await client(
functions.messages.SendReactionRequest(
peer=peer,
msg_id=message.id,
big=False,
add_to_recent=False,
reaction=[types.ReactionEmoji(emoticon=POOP_REACTION_EMOJI)],
)
)
except Exception as exc:
logging.debug(
"Не удалось поставить реакцию %s на сообщение %s: %s",
POOP_REACTION_EMOJI,
message.id,
exc,
)
async def delete_message_with_retries(
client: TelegramClient,
message: Message,
retries: tuple[float, ...] = (0.0, 0.7, 2.0),
) -> tuple[bool, Optional[str]]:
await react_with_poop_best_effort(client, message)
last_error: Optional[str] = None
for delay in retries:
if delay > 0:
await asyncio.sleep(delay)
deleted, error = await delete_message_best_effort(client, message)
if deleted:
return True, None
last_error = error
return False, last_error
async def cleanup_recent_messages_from_user(
@@ -177,12 +227,28 @@ async def cleanup_recent_messages_from_user(
async for message in client.iter_messages(chat_id, from_user=user_id, limit=limit):
if message.id is None:
continue
if await delete_message_best_effort(client, message):
deleted, _ = await delete_message_with_retries(client, message)
if deleted:
deleted_count += 1
return deleted_count
def is_permission_like_delete_error(error_text: Optional[str]) -> bool:
if not error_text:
return False
normalized = error_text.upper()
markers = (
"CHATADMINREQUIRED",
"CHAT_ADMIN_REQUIRED",
"MESSAGEDELETEFORBIDDEN",
"MESSAGE_DELETE_FORBIDDEN",
"RIGHT_FORBIDDEN",
)
return any(marker in normalized for marker in markers)
def load_settings() -> tuple[int, str, str]:
load_dotenv()
api_id_raw = os.getenv("API_ID", "").strip()
@@ -211,6 +277,7 @@ async def main() -> None:
api_id, api_hash, session_name = load_settings()
blocklist = BlocklistStore(DATA_PATH)
await blocklist.load()
notified_delete_failures: set[tuple[int, int]] = set()
client = TelegramClient(session_name, api_id, api_hash)
@@ -237,6 +304,12 @@ async def main() -> None:
status_message = await event.reply("Подготовка...")
added = await blocklist.add(target_user_id)
await animate_ban(status_message, target_user_id, added)
chat = await event.get_chat()
if isinstance(chat, types.Channel) and getattr(chat, "megagroup", False):
await event.reply(
"Важно: это супергруппа. В Telegram API локальное удаление чужих "
"сообщений может быть недоступно без прав администратора."
)
cleaned = await cleanup_recent_messages_from_user(
client,
event.chat_id,
@@ -264,6 +337,9 @@ async def main() -> None:
if event.is_private:
return
if event.is_channel and not event.is_group:
return
sender_id = event.sender_id
if sender_id is None:
return
@@ -271,7 +347,7 @@ async def main() -> None:
if not await blocklist.contains(int(sender_id)):
return
deleted = await delete_message_best_effort(client, event.message)
deleted, error = await delete_message_with_retries(client, event.message)
if not deleted:
logging.warning(
"Пропуск удаления: сообщение %s от пользователя %s в чате %s.",
@@ -279,6 +355,21 @@ async def main() -> None:
sender_id,
event.chat_id,
)
if is_permission_like_delete_error(error):
key = (int(event.chat_id), int(sender_id))
if key not in notified_delete_failures:
notified_delete_failures.add(key)
try:
await client.send_message(
"me",
"Не удалось удалить сообщение по автофильтру.\n"
f"Чат: `{event.chat_id}`\n"
f"Пользователь: `{sender_id}`\n"
f"Причина: `{error}`\n"
"Обычно это ограничение прав/типа чата в Telegram.",
)
except Exception:
logging.exception("Не удалось отправить уведомление в Избранное.")
await client.start()
me = await client.get_me()