# meta developer: @origsama # requires: yt-dlp curl_cffi from .. import loader, utils import requests import re import os import asyncio import tempfile import urllib.parse import json import shutil import sys @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: err_msg = info if isinstance(info, str) else "Неизвестная ошибка" await status.edit(f"❌ Не удалось скачать видео:\n{utils.escape_html(err_msg)}") 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): try: with tempfile.TemporaryDirectory() as temp_dir: output_template = os.path.join(temp_dir, '%(title)s.%(ext)s') cmd = [ sys.executable, '-m', 'yt_dlp', '-f', 'bestvideo+bestaudio/best', '--max-filesize', f"{self.config['max_size']}M", '-o', output_template, url ] if source.lower() == 'tiktok': cmd.extend(['--cookies', 'tiktok_cookies.txt', '--impersonate', 'Chrome']) elif source.lower() == 'instagram': cmd.extend(['--cookies', 'instagram_cookies.txt', '--impersonate', 'Chrome']) process = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if process.returncode != 0: return None, stderr.decode('utf-8', errors='ignore') for file in os.listdir(temp_dir): file_path = os.path.join(temp_dir, file) if os.path.isfile(file_path): 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, str(e) async def get_video_info(self, url): try: cmd = [sys.executable, '-m', 'yt_dlp', '--dump-json', '--no-playlist', '--impersonate', 'Chrome', url] process = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if process.returncode == 0: data = json.loads(stdout.decode('utf-8')) 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 = [ sys.executable, '-m', 'yt_dlp', '-f', f'best[height<={quality}]' if quality != 'best' else 'best', '--max-filesize', f"{self.config['max_size']}M", '-o', output_template, url ] process = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if process.returncode != 0: err_msg = stderr.decode('utf-8', errors='ignore') await utils.answer(message, f"❌ Ошибка скачивания:\n{utils.escape_html(err_msg)}") return for file in os.listdir(temp_dir): file_path = os.path.join(temp_dir, file) if os.path.isfile(file_path): 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)