Heroku-Modules/universal.py

# meta developer: @HoroshoVse
# requires: yt-dlp

from .. import loader, utils
import os
import asyncio
import tempfile
import json
import shutil
import sys

@loader.tds
class YIDownloaderMod(loader.Module):
    """Универсальный скачиватель видео"""
    strings = {"name": "YI downloader"}

    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<code>{utils.escape_html(err_msg)}</code>")
                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>"
            
            try:
                await message.client.send_file(
                    message.chat_id,
                    video_path,
                    caption=caption,
                    reply_to=message.reply_to_msg_id or message.id
                )
                await status.delete()
            except Exception as e:
                await status.edit(f"❌ Ошибка отправки в Telegram: {str(e)}")
            finally:
                if os.path.exists(video_path):
                    os.remove(video_path)
            
        except Exception as e:
            try:
                await status.edit(f"❌ Ошибка: {str(e)}")
            except:
                pass
    
    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',
                    '--merge-output-format', 'mp4',
                    '--max-filesize', f"{self.config['max_size']}M",
                    '-o', output_template,
                    url
                ]
                
                if source.lower() == 'tiktok':
                    cmd.extend(['--cookies', 'tiktok_cookies.txt', '--impersonate', ''])
                elif source.lower() == 'instagram':
                    cmd.extend(['--cookies', 'instagram_cookies.txt', '--impersonate', ''])
                
                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, source)
                        return final_path, info
            
            return None, "Файл не найден после скачивания"
            
        except Exception as e:
            return None, str(e)
    
    async def get_video_info(self, url, source):
        try:
            cmd = [sys.executable, '-m', 'yt_dlp', '--dump-json', '--no-playlist', url]
            if source.lower() in ['tiktok', 'instagram']:
                cmd.extend(['--impersonate', ''])
            
            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 {}