kerrad/telegram-helper-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

all type message can process

Browse Source
This commit is contained in:
KatykhinAA
2024-07-20 22:41:48 +03:00
parent 0b7f718f8a
commit 0704e6b3fe
7 changed files with 257 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
helper_bot/utils/__pycache__/helper_func.cpython-312.pyc
View File

Binary file not shown.

164
helper_bot/utils/helper_func.py
View File

@@ -1,7 +1,8 @@
import os
from datetime import datetime, timedelta
from aiogram import types
from aiogram.types import InputMediaPhoto, FSInputFile
from aiogram.types import InputMediaPhoto, FSInputFile, InputMediaVideo, InputMediaAudio
from helper_bot.utils.base_dependency_factory import BaseDependencyFactory
from logs.custom_logger import logger
@@ -37,9 +38,9 @@ def get_text_message(post_text: str, first_name: str, username: str):
return f'Пост из ТГ:\n{post_text}\n\nАвтор поста: {first_name} @{username}'
async def download_photo(message: types.Message, file_id: str):
async def download_file(message: types.Message, file_id: str):
"""
Скачивает фото по file_id из Telegram.
Скачивает файл по file_id из Telegram.
Args:
message: сообщение
@@ -50,9 +51,12 @@ async def download_photo(message: types.Message, file_id: str):
Путь к сохраненному файлу, если файл был скачан успешно, иначе None
"""
try:
os.makedirs("files", exist_ok=True)
os.makedirs("files/photos", exist_ok=True)
os.makedirs("files/videos", exist_ok=True)
file = await message.bot.get_file(file_id)
file_path = file.file_path
await message.bot.download_file(file_path=file_path, destination=file_path)
file_path = os.path.join("files", file.file_path)
await message.bot.download_file(file_path=file.file_path, destination=file_path)
return file_path
except Exception as e:
logger.error(f"Ошибка скачивания фотографии: {e}")
@@ -71,17 +75,37 @@ async def prepare_media_group_from_middlewares(album, post_caption: str = ''):
Список InputMediaPhoto (MediaGroup).
"""
media_group = []
# Циклом проходимся по собранному миддлварью объекту album
for i, message in enumerate(album):
file_id = message.photo[-1].file_id
# Если это последняя фото в массиве, то добавляем подпись. Остальные фото просто преобразуем в InputMediaPhoto
# и формируем объект MediaGroup
if i == len(album) - 1:
media_group.append(InputMediaPhoto(media=file_id, caption=post_caption))
if message.photo:
file_id = message.photo[-1].file_id
media_type = 'photo'
elif message.video:
file_id = message.video.file_id
media_type = 'video'
elif message.audio:
file_id = message.audio.file_id
media_type = 'audio'
else:
media_group.append(InputMediaPhoto(media=file_id))
# Если нет фото, видео или аудио, пропускаем сообщение
continue
# Формируем объект MediaGroup с учетом типа медиа
if i == len(album) - 1:
if media_type == 'photo':
media_group.append(InputMediaPhoto(media=file_id, caption=post_caption))
elif media_type == 'video':
media_group.append(InputMediaVideo(media=file_id, caption=post_caption))
elif media_type == 'audio':
media_group.append(InputMediaAudio(media=file_id, caption=post_caption))
else:
if media_type == 'photo':
media_group.append(InputMediaPhoto(media=file_id))
elif media_type == 'video':
media_group.append(InputMediaVideo(media=file_id))
elif media_type == 'audio':
media_group.append(InputMediaAudio(media=file_id))
return media_group # Возвращаем MediaGroup
@@ -89,6 +113,31 @@ async def add_in_db_media(sent_message):
"""
Идентификатор медиа-группы
Args:
sent_message: sent_message объект из Telegram API
Returns:
Список InputFile (FSInputFile).
"""
media_group_message_id = sent_message[-1].message_id # Получаем идентификатор медиа-группы
for i, message in enumerate(sent_message):
if message.photo:
file_id = message.photo[-1].file_id
file_path = await download_file(message, file_id=file_id)
BotDB.add_post_content_in_db(media_group_message_id, message.message_id, file_path, 'photo')
elif message.video:
file_id = message.video.file_id
file_path = await download_file(message, file_id=file_id)
BotDB.add_post_content_in_db(media_group_message_id, message.message_id, file_path, 'video')
else:
# Если нет фото, видео или аудио, или другой контент, пропускаем сообщение
continue
async def add_in_db_media_old(sent_message):
"""
Идентификатор медиа-группы
Args:
sent_message: sent_message объект из Telegram API
@@ -98,7 +147,7 @@ async def add_in_db_media(sent_message):
media_group_message_id = sent_message[-1].message_id # Получаем идентификатор медиа-группы
for i, message in enumerate(sent_message):
file_id = message.photo[-1].file_id
file_path = await download_photo(message, file_id=file_id)
file_path = await download_file(message, file_id=file_id)
if i == 0:
BotDB.add_post_content_in_db(media_group_message_id, message.message_id, file_path)
elif i == len(sent_message) - 1:
@@ -133,7 +182,11 @@ async def send_media_group_to_channel(bot, chat_id: int, post_content: list[tupl
for file_path in post_content:
try:
file = FSInputFile(path=file_path[0])
media.append(types.InputMediaPhoto(media=file))
type = file_path[1]
if type == 'video':
media.append(types.InputMediaVideo(media=file))
if type == 'photo':
media.append(types.InputMediaPhoto(media=file))
except FileNotFoundError:
logger.error(f"Файл не найден: {file_path[0]}")
return
@@ -180,6 +233,70 @@ async def send_photo_message(chat_id, message: types.Message, photo: str, post_t
)
async def send_video_message(chat_id, message: types.Message, video: str, post_text: str = "",
markup: types.ReplyKeyboardMarkup = None):
if markup is None:
await message.bot.send_video(
chat_id=chat_id,
caption=post_text,
video=video
)
else:
await message.bot.send_video(
chat_id=chat_id,
caption=post_text,
video=video,
reply_markup=markup
)
async def send_video_note_message(chat_id, message: types.Message, video_note: str,
markup: types.ReplyKeyboardMarkup = None):
if markup is None:
await message.bot.send_video_note(
chat_id=chat_id,
video_note=video_note
)
else:
await message.bot.send_video_note(
chat_id=chat_id,
video_note=video_note,
reply_markup=markup
)
async def send_audio_message(chat_id, message: types.Message, audio: str, post_text: str,
markup: types.ReplyKeyboardMarkup = None):
if markup is None:
await message.bot.send_audio(
chat_id=chat_id,
caption=post_text,
audio=audio
)
else:
await message.bot.send_audio(
chat_id=chat_id,
caption=post_text,
audio=audio,
reply_markup=markup
)
async def send_voice_message(chat_id, message: types.Message, voice: str,
markup: types.ReplyKeyboardMarkup = None):
if markup is None:
await message.bot.send_voice(
chat_id=chat_id,
voice=voice
)
else:
await message.bot.send_voice(
chat_id=chat_id,
voice=voice,
reply_markup=markup
)
def check_access(user_id: int):
"""Проверка прав на совершение действий"""
return BotDB.is_admin(user_id)
@@ -233,23 +350,6 @@ def get_banned_users_buttons():
return user_ids
def get_help_message_id(media_group_message_id: int, data: dict) -> int:
"""
Получает идентификатор текстового сообщения по идентификатору сообщения группы.
Args:
media_group_message_id: Идентификатор сообщения группы
data: Словарь с данными.
Returns:
Идентификатор сообщения помощи.
"""
if 'help_message_id' in data:
return data['help_message_id'] # Возвращаем help_message_id
else:
return 0
def delete_user_blacklist(user_id: int):
return BotDB.delete_user_blacklist(user_id=user_id)