➡️Вот мой код:
👉
import os
import logging
import re
import asyncio
from telegram import Update, InputMediaPhoto, InputMediaVideo, InputMediaDocument
from telegram.ext import (
ApplicationBuilder,
ContextTypes,
MessageHandler,
filters,
)
from dotenv import load_dotenv
from contextlib import suppress

Загрузка переменных окружения из .env файла
load_dotenv()

BOT_TOKEN = os.getenv("BOT_TOKEN")
CHANNEL_ID = int(os.getenv("CHANNEL_ID"))
ALLOWED_USER_ID = int(os.getenv("ALLOWED_USER_ID"))
MAX_CAPTION_LENGTH = 1024
GROUP_DELAY = 3 # Увеличено для надежного сбора группы
MAX_RETRIES = 3 # Максимальное количество повторных попыток отправки
RETRY_DELAY = 2 # Задержка между повторными попытками

Настройка логирования
logging.basicConfig(
filename="bot.log", # Добавлено логирование в файл для лучшего отслеживания
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
logger = logging.getLogger(name)

def clean_text(text):
"""Удаляет упоминания и хэштеги из текста."""
return re.sub(r'@\w+|#\w+', '', text, flags=re.IGNORECASE)

Словарь для хранения группированных сообщений
grouped_messages = {}

Словарь для отслеживания запланированных задач
scheduled_group_tasks = {}

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Обработчик всех входящих сообщений."""
if update.effective_user.id != ALLOWED_USER_ID:
await update.message.reply_text("Извините, вы не имеете доступа к использованию этого бота.")
return

message = update.message

Проверка, является ли сообщение частью группового альбома
if message.media_group_id:
group_id = message.media_group_id


if group_id not in grouped_messages:
    grouped_messages[group_id] = []

grouped_messages[group_id].append(message)

if group_id not in scheduled_group_tasks:
    # Запланировать обработку группы спустя GROUP_DELAY секунд
    task = asyncio.create_task(process_group_after_delay(group_id, context))
    scheduled_group_tasks[group_id] = task
else:
# Обработка одиночных сообщений
await send_single_message(message, context)

Удаление сообщения пользователя
try:
await delete_message_with_retries(context.bot, message.chat_id, message.message_id)
except Exception as e:
logger.error(f"Не удалось удалить сообщение: {e}")
async def process_group_after_delay(group_id, context):
"""Фоновая задача для обработки групповых сообщений после задержки."""
await asyncio.sleep(GROUP_DELAY)
messages = grouped_messages.pop(group_id, [])
scheduled_group_tasks.pop(group_id, None)

if not messages:
logger.info(f"No messages found for group {group_id}.")
return

Сортировка сообщений по message_id для сохранения порядка
messages.sort(key=lambda msg: msg.message_id)

media = []
caption_set = False
caption = ''

for msg in messages:
media_file = None
if msg.photo:
if msg.caption_html and not caption_set:
cleaned_caption = clean_text(msg.caption_html)
media_file = InputMediaPhoto(
media=msg.photo[-1].file_id,
caption=cleaned_caption,
parse_mode='HTML'
)
caption_set = True
caption = cleaned_caption
else:
media_file = InputMediaPhoto(media=msg.photo[-1].file_id)
elif msg.video:
if msg.caption_html and not caption_set:
cleaned_caption = clean_text(msg.caption_html)
media_file = InputMediaVideo(
media=msg.video.file_id,
caption=cleaned_caption,
parse_mode='HTML'
)
caption_set = True
caption = cleaned_caption
else:
media_file = InputMediaVideo(media=msg.video.file_id)
elif msg.document:
if msg.caption_html and not caption_set:
cleaned_caption = clean_text(msg.caption_html)
media_file = InputMediaDocument(
media=msg.document.file_id,
caption=cleaned_caption,
parse_mode='HTML'
)
caption_set = True
caption = cleaned_caption
else:
media_file = InputMediaDocument(media=msg.document.file_id)
else:
logger.warning(f"Unsupported media type in message {msg.message_id}")
continue


if media_file:
    media.append(media_file)
if not media:
logger.info(f"No valid media files to send for group {group_id}.")
return

Проверка длины подписи
if caption and len(caption) > MAX_CAPTION_LENGTH:
try:
await send_with_retries(
context.bot.send_message,
chat_id=CHANNEL_ID,
text=caption[:MAX_CAPTION_LENGTH],
parse_mode='HTML'
)
caption = None
for m in media:
m.caption = None # Удаляем подпись с остальных медиа
except Exception as e:
logger.error(f"Не удалось отправить сокращенную подпись: {e}")

Отправка группового медиа в канал
try:
if media:
await send_media_group_with_retries(context.bot, CHANNEL_ID, media)
logger.info(f"Sent media group {group_id} with {len(media)} items.")
except Exception as e:
logger.error(f"Не удалось отправить групповое медиа {group_id}: {e}")
async def send_single_message(message, context):
"""Отправка одиночного сообщения в канал."""

Очистка подписи
caption = clean_text(message.caption_html or '')

Проверка длины подписи
if len(caption) > MAX_CAPTION_LENGTH:
try:
await send_with_retries(
context.bot.send_message,
chat_id=CHANNEL_ID,
text=caption[:MAX_CAPTION_LENGTH],
parse_mode='HTML'
)
caption = None
except Exception as e:
logger.error(f"Не удалось отправить сокращенную подпись: {e}")

Обработка и отправка медиа или текстового сообщения
try:
if message.photo:
await send_with_retries(
context.bot.send_photo,
chat_id=CHANNEL_ID,
photo=message.photo[-1].file_id,
caption=caption,
parse_mode='HTML'
)
elif message.video:
await send_with_retries(
context.bot.send_video,
chat_id=CHANNEL_ID,
video=message.video.file_id,
caption=caption,
parse_mode='HTML'
)
elif message.document:
await send_with_retries(
context.bot.send_document,
chat_id=CHANNEL_ID,
document=message.document.file_id,
caption=caption,
parse_mode='HTML'
)
elif message.audio: # Добавлено для обработки аудио
await send_with_retries(
context.bot.send_audio,
chat_id=CHANNEL_ID,
audio=message.audio.file_id,
caption=caption,
parse_mode='HTML'
)
elif message.voice: # Добавлено для обработки голосовых сообщений
await send_with_retries(
context.bot.send_voice,
chat_id=CHANNEL_ID,
voice=message.voice.file_id,
caption=caption,
parse_mode='HTML'
)
elif message.video_note: # Добавлено для обработки кружочков
await send_with_retries(
context.bot.send_video_note,
chat_id=CHANNEL_ID,
video_note=message.video_note.file_id
)
elif message.text: # Обработка текста, включая только смайлики
await send_with_retries(
context.bot.send_message,
chat_id=CHANNEL_ID,
text=message.text,
parse_mode='HTML'
)
except Exception as e:
logger.error(f"Не удалось отправить одиночное сообщение: {e}")
async def send_with_retries(send_func, **kwargs):
"""Функция для отправки сообщений с несколькими попытками."""
for attempt in range(1, MAX_RETRIES + 1):
try:
return await send_func(**kwargs)
except Exception as e:
logger.warning(f"Попытка {attempt} не удалась: {e}")
if attempt < MAX_RETRIES:
await asyncio.sleep(RETRY_DELAY)
else:
raise

async def send_media_group_with_retries(bot, chat_id, media):
"""Функция для отправки медиа-группы с несколькими попытками."""
for attempt in range(1, MAX_RETRIES + 1):
try:
return await bot.send_media_group(chat_id=chat_id, media=media)
except Exception as e:
logger.warning(f"Попытка {attempt} отправки медиа-группы не удалась: {e}")
if attempt < MAX_RETRIES:
await asyncio.sleep(RETRY_DELAY)
else:
raise

async def delete_message_with_retries(bot, chat_id, message_id):
"""Функция для удаления сообщения с несколькими попытками."""
for attempt in range(1, MAX_RETRIES + 1):
try:
return await bot.delete_message(chat_id=chat_id, message_id=message_id)
except Exception as e:
logger.warning(f"Попытка {attempt} удаления сообщения не удалась: {e}")
if attempt < MAX_RETRIES:
await asyncio.sleep(RETRY_DELAY)
else:
raise

async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Обработчик ошибок."""
logger.error(msg="Exception while handling an update:", exc_info=context.error)

def main():
"""Основная функция для запуска бота."""
application = ApplicationBuilder().token(BOT_TOKEN).build()

Обработчик всех сообщений
message_handler = MessageHandler(filters.ALL, handle_message)
application.add_handler(message_handler)

Обработчик ошибок
application.add_error_handler(error_handler)

Запуск бота
application.run_polling()
if name == 'main':
main()
👈
🔓Твои предложения и рекомендации по улучшению и оптимизации кода🙏❤️