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 02:06:25 +03:00
parent c10df870ab
commit 4b44ae9670
1 changed files with 22 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
universal.py
+22 -43
View File
@@ -1,12 +1,14 @@
# meta developer: @origsama # meta developer: @origsama
# requires: yt-dlp
from .. import loader, utils from .. import loader, utils
import requests import requests
import re import re
import os import os
import subprocess import asyncio
import tempfile import tempfile
import urllib.parse import urllib.parse
import json import json
import shutil
@loader.tds @loader.tds
class SocialDownloaderMod(loader.Module): class SocialDownloaderMod(loader.Module):
@@ -39,7 +41,6 @@ class SocialDownloaderMod(loader.Module):
) )
return return
# Определяем источник
source = self.detect_source(args) source = self.detect_source(args)
if not source: if not source:
await utils.answer(message, "❌ Неподдерживаемая ссылка") await utils.answer(message, "❌ Неподдерживаемая ссылка")
@@ -48,14 +49,12 @@ class SocialDownloaderMod(loader.Module):
status = await utils.answer(message, f"📥 Скачиваю с {source}...") status = await utils.answer(message, f"📥 Скачиваю с {source}...")
try: try:
# Скачиваем видео
video_path, info = await self.download_video(args, source) video_path, info = await self.download_video(args, source)
if not video_path: if not video_path:
await status.edit("❌ Не удалось скачать видео") await status.edit("❌ Не удалось скачать видео")
return return
# Формируем описание
caption = f"<b>📹 Видео с {source}</b>\n\n" caption = f"<b>📹 Видео с {source}</b>\n\n"
if info.get('title'): if info.get('title'):
caption += f"📝 {info['title']}\n" caption += f"📝 {info['title']}\n"
@@ -66,7 +65,6 @@ class SocialDownloaderMod(loader.Module):
caption += f"\n🔗 <a href='{args}'>Оригинал</a>" caption += f"\n🔗 <a href='{args}'>Оригинал</a>"
# Отправляем
await status.delete() await status.delete()
await message.client.send_file( await message.client.send_file(
@@ -76,16 +74,13 @@ class SocialDownloaderMod(loader.Module):
reply_to=message.reply_to_msg_id or message.id reply_to=message.reply_to_msg_id or message.id
) )
# Чистим
os.remove(video_path) os.remove(video_path)
except Exception as e: except Exception as e:
await status.edit(f"❌ Ошибка: {str(e)}") await status.edit(f"❌ Ошибка: {str(e)}")
def detect_source(self, url): def detect_source(self, url):
"""Определяет источник по ссылке"""
url = url.lower() url = url.lower()
sources = { sources = {
'tiktok': ['tiktok.com', 'vm.tiktok.com'], 'tiktok': ['tiktok.com', 'vm.tiktok.com'],
'instagram': ['instagram.com', 'instagr.am'], 'instagram': ['instagram.com', 'instagr.am'],
@@ -96,54 +91,45 @@ class SocialDownloaderMod(loader.Module):
'facebook': ['facebook.com', 'fb.com'], 'facebook': ['facebook.com', 'fb.com'],
'pinterest': ['pinterest.com', 'pin.it'] 'pinterest': ['pinterest.com', 'pin.it']
} }
for source, domains in sources.items(): for source, domains in sources.items():
for domain in domains: for domain in domains:
if domain in url: if domain in url:
return source.capitalize() return source.capitalize()
return None return None
async def download_video(self, url, source): async def download_video(self, url, source):
"""Скачивает видео через yt-dlp"""
try: try:
# Создаем временную папку
with tempfile.TemporaryDirectory() as temp_dir: with tempfile.TemporaryDirectory() as temp_dir:
output_template = os.path.join(temp_dir, '%(title)s.%(ext)s') output_template = os.path.join(temp_dir, '%(title)s.%(ext)s')
# Формируем команду
cmd = [ cmd = [
'yt-dlp', 'yt-dlp',
'-f', 'best[filesize<50M]', # Лучшее качество до 50 МБ '-f', 'best[filesize<50M]',
'--max-filesize', f"{self.config['max_size']}M", '--max-filesize', f"{self.config['max_size']}M",
'-o', output_template, '-o', output_template,
url url
] ]
# Добавляем специфичные опции
if source.lower() == 'tiktok': if source.lower() == 'tiktok':
cmd.extend(['--cookies', 'tiktok_cookies.txt']) # Опционально cmd.extend(['--cookies', 'tiktok_cookies.txt'])
elif source.lower() == 'instagram': elif source.lower() == 'instagram':
cmd.extend(['--cookies', 'instagram_cookies.txt']) cmd.extend(['--cookies', 'instagram_cookies.txt'])
# Запускаем process = await asyncio.create_subprocess_exec(
result = subprocess.run(cmd, capture_output=True, text=True) *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if result.returncode != 0: if process.returncode != 0:
return None, {} return None, {}
# Ищем скачанный файл
for file in os.listdir(temp_dir): for file in os.listdir(temp_dir):
file_path = os.path.join(temp_dir, file) file_path = os.path.join(temp_dir, file)
if os.path.isfile(file_path): if os.path.isfile(file_path):
# Копируем в постоянное место
import shutil
final_path = os.path.join(tempfile.gettempdir(), file) final_path = os.path.join(tempfile.gettempdir(), file)
shutil.copy2(file_path, final_path) shutil.copy2(file_path, final_path)
# Получаем информацию
info = await self.get_video_info(url) info = await self.get_video_info(url)
return final_path, info return final_path, info
return None, {} return None, {}
@@ -153,19 +139,15 @@ class SocialDownloaderMod(loader.Module):
return None, {} return None, {}
async def get_video_info(self, url): async def get_video_info(self, url):
"""Получает информацию о видео"""
try: try:
cmd = [ cmd = ['yt-dlp', '--dump-json', '--no-playlist', url]
'yt-dlp', process = await asyncio.create_subprocess_exec(
'--dump-json', *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
'--no-playlist', )
url stdout, stderr = await process.communicate()
]
result = subprocess.run(cmd, capture_output=True, text=True) if process.returncode == 0:
data = json.loads(stdout.decode('utf-8'))
if result.returncode == 0:
data = json.loads(result.stdout)
duration = data.get('duration', 0) duration = data.get('duration', 0)
minutes = duration // 60 minutes = duration // 60
@@ -176,7 +158,6 @@ class SocialDownloaderMod(loader.Module):
'author': data.get('uploader', data.get('channel', '')), 'author': data.get('uploader', data.get('channel', '')),
'duration': f"{minutes}:{seconds:02d}" if duration else '' 'duration': f"{minutes}:{seconds:02d}" if duration else ''
} }
return {} return {}
except: except:
return {} return {}
@@ -187,7 +168,6 @@ class SocialDownloaderMod(loader.Module):
if not args: if not args:
await utils.answer(message, "❌ Укажи ссылку на TikTok") await utils.answer(message, "❌ Укажи ссылку на TikTok")
return return
await self.dlcmd(message) await self.dlcmd(message)
async def instacmd(self, message): async def instacmd(self, message):
@@ -196,7 +176,6 @@ class SocialDownloaderMod(loader.Module):
if not args: if not args:
await utils.answer(message, "❌ Укажи ссылку на Instagram") await utils.answer(message, "❌ Укажи ссылку на Instagram")
return return
await self.dlcmd(message) await self.dlcmd(message)
async def ytcmd(self, message): async def ytcmd(self, message):
@@ -206,7 +185,6 @@ class SocialDownloaderMod(loader.Module):
await utils.answer(message, "❌ Укажи ссылку на YouTube") await utils.answer(message, "❌ Укажи ссылку на YouTube")
return return
# Разбираем качество если указано
parts = args.split() parts = args.split()
url = parts[0] url = parts[0]
quality = parts[1] if len(parts) > 1 else 'best' quality = parts[1] if len(parts) > 1 else 'best'
@@ -225,16 +203,18 @@ class SocialDownloaderMod(loader.Module):
url url
] ]
result = subprocess.run(cmd, capture_output=True, text=True) process = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if result.returncode != 0: if process.returncode != 0:
await utils.answer(message, "❌ Ошибка скачивания") await utils.answer(message, "❌ Ошибка скачивания")
return return
for file in os.listdir(temp_dir): for file in os.listdir(temp_dir):
file_path = os.path.join(temp_dir, file) file_path = os.path.join(temp_dir, file)
if os.path.isfile(file_path): if os.path.isfile(file_path):
import shutil
final_path = os.path.join(tempfile.gettempdir(), file) final_path = os.path.join(tempfile.gettempdir(), file)
shutil.copy2(file_path, final_path) shutil.copy2(file_path, final_path)
@@ -267,5 +247,4 @@ class SocialDownloaderMod(loader.Module):
if not args: if not args:
await utils.answer(message, "❌ Укажи ссылку на VK видео") await utils.answer(message, "❌ Укажи ссылку на VK видео")
return return
await self.dlcmd(message) await self.dlcmd(message)