From c10df870ab5a4dfb9285012d6e427c36e1d65bf3 Mon Sep 17 00:00:00 2001 From: yanistyle Date: Mon, 31 Aug 2026 17:00:14 +0300 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?/=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Модуль для загрузки из популярный соц сетей видео (tiktok,YouTube,vk,Instagram) --- universal.py | 271 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 271 insertions(+) create mode 100644 universal.py diff --git a/universal.py b/universal.py new file mode 100644 index 0000000..02c9185 --- /dev/null +++ b/universal.py @@ -0,0 +1,271 @@ +# 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"📹 Видео с {source}\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🔗 Оригинал" + + # Отправляем + 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"🎥 YouTube\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) \ No newline at end of file