yanistyle/Heroku-Modules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Обновить universal.py

Browse Source
This commit is contained in:
yanistyle
2026-09-01 03:01:32 +03:00
parent 4c30612e81
commit a95c096214
1 changed files with 31 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
universal.py
+31 -117
View File
@@ -1,20 +1,17 @@
# meta developer: @origsama
# requires: yt-dlp curl_cffi
# meta developer: @HoroshoVse
# requires: yt-dlp
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"}
class YIDownloaderMod(loader.Module):
"""Универсальный скачиватель видео"""
strings = {"name": "YI downloader"}
def __init__(self):
self.config = loader.ModuleConfig(
@@ -67,19 +64,25 @@ class SocialDownloaderMod(loader.Module):
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)
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:
await status.edit(f"❌ Ошибка: {str(e)}")
try:
await status.edit(f"❌ Ошибка: {str(e)}")
except:
pass
def detect_source(self, url):
url = url.lower()
@@ -106,16 +109,17 @@ class SocialDownloaderMod(loader.Module):
cmd = [
sys.executable, '-m', 'yt_dlp',
'-f', 'bestvideo+bestaudio/best',
'-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'])
cmd.extend(['--cookies', 'tiktok_cookies.txt', '--downloader', 'default', '--compat-options', 'no-impersonate'])
elif source.lower() == 'instagram':
cmd.extend(['--cookies', 'instagram_cookies.txt'])
cmd.extend(['--cookies', 'instagram_cookies.txt', '--downloader', 'default', '--compat-options', 'no-impersonate'])
process = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
@@ -131,18 +135,20 @@ class SocialDownloaderMod(loader.Module):
final_path = os.path.join(tempfile.gettempdir(), file)
shutil.copy2(file_path, final_path)
info = await self.get_video_info(url)
info = await self.get_video_info(url, source)
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):
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(['--downloader', 'default', '--compat-options', 'no-impersonate'])
process = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
@@ -150,7 +156,6 @@ class SocialDownloaderMod(loader.Module):
if process.returncode == 0:
data = json.loads(stdout.decode('utf-8'))
duration = data.get('duration', 0)
minutes = duration // 60
seconds = duration % 60
@@ -163,95 +168,4 @@ class SocialDownloaderMod(loader.Module):
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')
format_str = f'bestvideo[height<={quality}]+bestaudio/best[height<={quality}]/best' if quality != 'best' else 'bestvideo+bestaudio/best'
cmd = [
sys.executable, '-m', 'yt_dlp',
'-f', format_str,
'--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<code>{utils.escape_html(err_msg)}</code>")
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"<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)