yanistyle
c10df870ab
Модуль для загрузки из популярный соц сетей видео (tiktok,YouTube,vk,Instagram)
271 lines
10 KiB
Python
271 lines
10 KiB
Python
# meta developer: @origsama
|
||
from .. import loader, utils
|
||
import requests
|
||
import re
|
||
import os
|
||
import subprocess
|
||
import tempfile
|
||
import urllib.parse
|
||
import json
|
||
|
||
@loader.tds
|
||
class SocialDownloaderMod(loader.Module):
|
||
"""Скачивание видео из TikTok, Instagram, YouTube, VK"""
|
||
strings = {"name": "SocialDownloader"}
|
||
|
||
def __init__(self):
|
||
self.config = loader.ModuleConfig(
|
||
loader.ConfigValue(
|
||
"max_size",
|
||
50,
|
||
"Максимальный размер видео в МБ",
|
||
validator=loader.validators.Integer()
|
||
)
|
||
)
|
||
|
||
async def dlcmd(self, message):
|
||
"""<ссылка> - Скачать видео из TikTok/Instagram/YouTube/VK"""
|
||
args = utils.get_args_raw(message)
|
||
if not args:
|
||
await utils.answer(message,
|
||
"❌ Укажи ссылку на видео\n\n"
|
||
"Поддерживаемые сайты:\n"
|
||
"• TikTok: https://vm.tiktok.com/...\n"
|
||
"• Instagram: https://www.instagram.com/p/...\n"
|
||
"• YouTube: https://youtu.be/...\n"
|
||
"• VK: https://vk.com/video...\n"
|
||
"• Twitter/X: https://twitter.com/...\n"
|
||
"• Reddit: https://reddit.com/..."
|
||
)
|
||
return
|
||
|
||
# Определяем источник
|
||
source = self.detect_source(args)
|
||
if not source:
|
||
await utils.answer(message, "❌ Неподдерживаемая ссылка")
|
||
return
|
||
|
||
status = await utils.answer(message, f"📥 Скачиваю с {source}...")
|
||
|
||
try:
|
||
# Скачиваем видео
|
||
video_path, info = await self.download_video(args, source)
|
||
|
||
if not video_path:
|
||
await status.edit("❌ Не удалось скачать видео")
|
||
return
|
||
|
||
# Формируем описание
|
||
caption = f"<b>📹 Видео с {source}</b>\n\n"
|
||
if info.get('title'):
|
||
caption += f"📝 {info['title']}\n"
|
||
if info.get('author'):
|
||
caption += f"👤 {info['author']}\n"
|
||
if info.get('duration'):
|
||
caption += f"⏱ {info['duration']}\n"
|
||
|
||
caption += f"\n🔗 <a href='{args}'>Оригинал</a>"
|
||
|
||
# Отправляем
|
||
await status.delete()
|
||
|
||
await message.client.send_file(
|
||
message.chat_id,
|
||
video_path,
|
||
caption=caption,
|
||
reply_to=message.reply_to_msg_id or message.id
|
||
)
|
||
|
||
# Чистим
|
||
os.remove(video_path)
|
||
|
||
except Exception as e:
|
||
await status.edit(f"❌ Ошибка: {str(e)}")
|
||
|
||
def detect_source(self, url):
|
||
"""Определяет источник по ссылке"""
|
||
url = url.lower()
|
||
|
||
sources = {
|
||
'tiktok': ['tiktok.com', 'vm.tiktok.com'],
|
||
'instagram': ['instagram.com', 'instagr.am'],
|
||
'youtube': ['youtube.com', 'youtu.be', 'm.youtube.com'],
|
||
'vk': ['vk.com', 'vkontakte.ru'],
|
||
'twitter': ['twitter.com', 'x.com'],
|
||
'reddit': ['reddit.com', 'redd.it'],
|
||
'facebook': ['facebook.com', 'fb.com'],
|
||
'pinterest': ['pinterest.com', 'pin.it']
|
||
}
|
||
|
||
for source, domains in sources.items():
|
||
for domain in domains:
|
||
if domain in url:
|
||
return source.capitalize()
|
||
|
||
return None
|
||
|
||
async def download_video(self, url, source):
|
||
"""Скачивает видео через yt-dlp"""
|
||
try:
|
||
# Создаем временную папку
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
output_template = os.path.join(temp_dir, '%(title)s.%(ext)s')
|
||
|
||
# Формируем команду
|
||
cmd = [
|
||
'yt-dlp',
|
||
'-f', 'best[filesize<50M]', # Лучшее качество до 50 МБ
|
||
'--max-filesize', f"{self.config['max_size']}M",
|
||
'-o', output_template,
|
||
url
|
||
]
|
||
|
||
# Добавляем специфичные опции
|
||
if source.lower() == 'tiktok':
|
||
cmd.extend(['--cookies', 'tiktok_cookies.txt']) # Опционально
|
||
elif source.lower() == 'instagram':
|
||
cmd.extend(['--cookies', 'instagram_cookies.txt'])
|
||
|
||
# Запускаем
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
|
||
if result.returncode != 0:
|
||
return None, {}
|
||
|
||
# Ищем скачанный файл
|
||
for file in os.listdir(temp_dir):
|
||
file_path = os.path.join(temp_dir, file)
|
||
if os.path.isfile(file_path):
|
||
# Копируем в постоянное место
|
||
import shutil
|
||
final_path = os.path.join(tempfile.gettempdir(), file)
|
||
shutil.copy2(file_path, final_path)
|
||
|
||
# Получаем информацию
|
||
info = await self.get_video_info(url)
|
||
|
||
return final_path, info
|
||
|
||
return None, {}
|
||
|
||
except Exception as e:
|
||
print(f"Download error: {e}")
|
||
return None, {}
|
||
|
||
async def get_video_info(self, url):
|
||
"""Получает информацию о видео"""
|
||
try:
|
||
cmd = [
|
||
'yt-dlp',
|
||
'--dump-json',
|
||
'--no-playlist',
|
||
url
|
||
]
|
||
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
|
||
if result.returncode == 0:
|
||
data = json.loads(result.stdout)
|
||
|
||
duration = data.get('duration', 0)
|
||
minutes = duration // 60
|
||
seconds = duration % 60
|
||
|
||
return {
|
||
'title': data.get('title', ''),
|
||
'author': data.get('uploader', data.get('channel', '')),
|
||
'duration': f"{minutes}:{seconds:02d}" if duration else ''
|
||
}
|
||
|
||
return {}
|
||
except:
|
||
return {}
|
||
|
||
async def tiktokcmd(self, message):
|
||
"""<ссылка> - Скачать видео с TikTok"""
|
||
args = utils.get_args_raw(message)
|
||
if not args:
|
||
await utils.answer(message, "❌ Укажи ссылку на TikTok")
|
||
return
|
||
|
||
await self.dlcmd(message)
|
||
|
||
async def instacmd(self, message):
|
||
"""<ссылка> - Скачать видео/фото с Instagram"""
|
||
args = utils.get_args_raw(message)
|
||
if not args:
|
||
await utils.answer(message, "❌ Укажи ссылку на Instagram")
|
||
return
|
||
|
||
await self.dlcmd(message)
|
||
|
||
async def ytcmd(self, message):
|
||
"""<ссылка> [качество] - Скачать видео с YouTube"""
|
||
args = utils.get_args_raw(message)
|
||
if not args:
|
||
await utils.answer(message, "❌ Укажи ссылку на YouTube")
|
||
return
|
||
|
||
# Разбираем качество если указано
|
||
parts = args.split()
|
||
url = parts[0]
|
||
quality = parts[1] if len(parts) > 1 else 'best'
|
||
|
||
await utils.answer(message, f"📥 Скачиваю с YouTube ({quality})...")
|
||
|
||
try:
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
output_template = os.path.join(temp_dir, '%(title)s.%(ext)s')
|
||
|
||
cmd = [
|
||
'yt-dlp',
|
||
'-f', f'best[height<={quality}]' if quality != 'best' else 'best',
|
||
'--max-filesize', f"{self.config['max_size']}M",
|
||
'-o', output_template,
|
||
url
|
||
]
|
||
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
|
||
if result.returncode != 0:
|
||
await utils.answer(message, "❌ Ошибка скачивания")
|
||
return
|
||
|
||
for file in os.listdir(temp_dir):
|
||
file_path = os.path.join(temp_dir, file)
|
||
if os.path.isfile(file_path):
|
||
import shutil
|
||
final_path = os.path.join(tempfile.gettempdir(), file)
|
||
shutil.copy2(file_path, final_path)
|
||
|
||
info = await self.get_video_info(url)
|
||
|
||
caption = f"<b>🎥 YouTube</b>\n"
|
||
if info.get('title'):
|
||
caption += f"📝 {info['title']}\n"
|
||
if info.get('author'):
|
||
caption += f"👤 {info['author']}\n"
|
||
if info.get('duration'):
|
||
caption += f"⏱ {info['duration']}\n"
|
||
caption += f"\n⚙️ {quality}"
|
||
|
||
await message.client.send_file(
|
||
message.chat_id,
|
||
final_path,
|
||
caption=caption
|
||
)
|
||
|
||
os.remove(final_path)
|
||
return
|
||
|
||
except Exception as e:
|
||
await utils.answer(message, f"❌ Ошибка: {str(e)}")
|
||
|
||
async def vkcmd(self, message):
|
||
"""<ссылка> - Скачать видео с VK"""
|
||
args = utils.get_args_raw(message)
|
||
if not args:
|
||
await utils.answer(message, "❌ Укажи ссылку на VK видео")
|
||
return
|
||
|
||
await self.dlcmd(message) |