-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtg_dl_bot.py
151 lines (127 loc) · 7.52 KB
/
tg_dl_bot.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
import re
import os
import json
from moviepy.editor import VideoFileClip
from telethon import TelegramClient, events
from telethon.tl.types import DocumentAttributeVideo, InputMediaUploadedDocument
# Загрузка конфигурации из файла config.json
with open('config.json', 'r') as config_file:
config = json.load(config_file)
# Загрузка данных из файла конфигурации
api_id = config['api_id']
api_hash = config['api_hash']
bot_token = config['bot_token']
target_group_id = config['target_group_id']
client = TelegramClient('bot', api_id, api_hash).start(bot_token=bot_token)
# Загрузка списка ACL из файла
def load_acl():
try:
with open('acl.txt', 'r') as file:
return set(line.strip() for line in file if line.strip())
except FileNotFoundError:
print("Файл acl.txt не найден. Бот будет отвечать всем.")
return set()
acl_list = load_acl()
# Генерация превью для длинных видео
def create_thumbnail(file_path):
try:
video = VideoFileClip(file_path)
thumb_path = file_path.rsplit(".", 1)[0] + ".jpg"
video.save_frame(thumb_path, t=1.0)
return thumb_path
except Exception as e:
print(f"Error creating thumbnail for {file_path}: {e}")
return None
# Проверка пользователя по ACL
def is_user_allowed(user):
# Проверка ID пользователя, никнейма и номера телефона
return (str(user.id) in acl_list or
user.username and f"@{user.username}" in acl_list or
user.phone and f"+{user.phone}" in acl_list)
async def download_and_send_media(event, chat_id, message_id):
try:
# Получаем сообщение с указанным ID
message = await client.get_messages(chat_id, ids=message_id)
if message is None:
await event.respond('Не удалось получить сообщение. Возможно, у бота нет прав на доступ к этому сообщению.')
print('Сообщение не найдено.')
return
print(f"Сообщение найдено: {message.id}")
if message.media:
# Указываем путь для загрузки в /tmp
file_path = await message.download_media(file='/tmp/')
thumb_path = None # Инициализируем переменную заранее
if file_path:
# Проверяем размер файла и создаем превью, если он больше 10 MB
if os.path.getsize(file_path) > 10 * 1024 * 1024: # 10 MB
thumb_path = create_thumbnail(file_path)
# Получаем информацию о видео (продолжительность, ширина, высота)
video = VideoFileClip(file_path)
duration = int(video.duration)
width, height = video.size
video.close()
# Загружаем видео и его превью в Telegram
uploaded_video = await client.upload_file(file_path)
uploaded_thumb = await client.upload_file(thumb_path) if thumb_path else None
# Создаем InputMediaUploadedDocument для корректной отправки видео с превью
attributes = [DocumentAttributeVideo(duration=duration, w=width, h=height, supports_streaming=True)]
media = InputMediaUploadedDocument(
file=uploaded_video,
mime_type='video/mp4',
attributes=attributes,
thumb=uploaded_thumb
)
# Отправляем обработанный медиафайл в личный чат бота с пользователем
await client.send_file(event.chat_id, file=media)
else:
# Если видео меньше 10MB, отправляем оригинал
await client.send_file(event.chat_id, file=file_path)
# Удаляем временные файлы после отправки
os.remove(file_path)
if thumb_path:
os.remove(thumb_path) # Удаляем превью после отправки
print(f"Медиафайл {file_path} успешно загружен и отправлен.")
else:
await event.respond('Не удалось загрузить медиафайл.')
print("Не удалось загрузить медиафайл.")
else:
await event.respond('В указанном сообщении нет медиафайла.')
print("В указанном сообщении нет медиафайла.")
except Exception as e:
await event.respond(f'Ошибка при загрузке медиафайла: {str(e)}')
print(f'Ошибка: {str(e)}')
@client.on(events.NewMessage(pattern=r'https://t\.me/c/\d+/(\d+)', incoming=True))
async def handler(event):
# Убедимся, что бот реагирует только на сообщения из личных чатов
if event.is_private:
sender = await event.get_sender()
if not is_user_allowed(sender):
print(f"Пользователь {sender.id} не имеет доступа.")
return
# Парсинг ссылки
match = re.search(r'https://t\.me/c/(\d+)/(\d+)', event.text)
if match:
group_id = match.group(1)
message_id = int(match.group(2))
# Проверяем, что группа соответствует указанному target_group_id
if group_id != target_group_id:
await event.respond('Ссылки из этой группы не разрешены.')
print(f'Ссылка из группы {group_id} отклонена.')
return
chat_id = int('-100' + group_id) # Преобразование в формат идентификатора группы
print(f"Получен запрос для группы {chat_id} и сообщения {message_id}")
# Проверка на возможность доступа к чату
try:
entity = await client.get_entity(chat_id)
print(f"Доступ к чату {entity.title} подтвержден.")
except Exception as e:
await event.respond(f'Ошибка доступа к чату: {str(e)}')
print(f'Ошибка доступа к чату: {str(e)}')
return
# Вызов функции для загрузки и отправки медиа
await download_and_send_media(event, chat_id, message_id)
else:
await event.respond('Неверная ссылка. Пожалуйста, отправьте ссылку в формате https://t.me/c/ID/MessageID.')
print('Неверная ссылка. Пожалуйста, отправьте ссылку в формате https://t.me/c/ID/MessageID.')
print("Бот запущен...")
client.run_until_disconnected()