74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
import asyncio
|
|
from functools import partial
|
|
from pathlib import Path
|
|
import os
|
|
import yt_dlp
|
|
|
|
def build_opts(for_video: bool, outdir: Path, cookiefile: Path | None, browser: str | None, message=None, out_for_playlist: bool = True) -> dict:
|
|
postprocessors = []
|
|
if for_video:
|
|
# На этапе постпроцессора делаем простую конвертацию в mp4 (дальше будет наша принудительная перекодировка)
|
|
postprocessors.append({'key': 'FFmpegVideoConvertor', 'preferedformat': 'mp4'})
|
|
|
|
# Для плейлистов (stories/галереи) — уникальные имена, чтобы не затирать
|
|
outtmpl = '%(uploader)s_%(id)s.%(ext)s' if out_for_playlist else '%(title)s.%(ext)s'
|
|
|
|
opts: dict = {
|
|
'outtmpl': str(outdir / outtmpl),
|
|
'quiet': True,
|
|
'merge_output_format': 'mp4',
|
|
'postprocessors': postprocessors,
|
|
'retries': 5,
|
|
'fragment_retries': 5,
|
|
'concurrent_fragment_downloads': 5,
|
|
'noprogress': True,
|
|
'source_address': '0.0.0.0', # форс IPv4
|
|
}
|
|
|
|
proxy = os.getenv('HTTPS_PROXY') or os.getenv('HTTP_PROXY') or ''
|
|
if proxy:
|
|
opts['proxy'] = proxy
|
|
|
|
if cookiefile and cookiefile.exists():
|
|
opts['cookiefile'] = str(cookiefile)
|
|
elif browser:
|
|
opts['cookiesfrombrowser'] = (browser,)
|
|
|
|
if message:
|
|
opts['progress_hooks'] = [partial(progress_hook_async_bridge, message=message)]
|
|
|
|
return opts
|
|
|
|
async def progress_hook_async_bridge(d, message):
|
|
try:
|
|
if d.get('status') == 'downloading':
|
|
percent = d.get('_percent_str', '').strip()
|
|
speed = d.get('_speed_str', '').strip()
|
|
eta = d.get('_eta_str', '').strip()
|
|
text = f"⬇️ Загрузка: {percent or '?'}\n🚀 Скорость: {speed or '?'}\n⏱ Осталось: {eta or '?'}"
|
|
await message.edit_text(text)
|
|
except Exception:
|
|
pass
|
|
|
|
async def run(url: str, opts: dict) -> tuple[Path | None, dict | None]:
|
|
loop = asyncio.get_running_loop()
|
|
info, filepath = None, None
|
|
|
|
def _do():
|
|
nonlocal info, filepath
|
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
|
info = ydl.extract_info(url, download=True)
|
|
out = None
|
|
if isinstance(info, dict):
|
|
out = info.get('_filename')
|
|
if not out:
|
|
rd = info.get('requested_downloads') or []
|
|
for it in rd[::-1]:
|
|
out = it.get('filepath') or it.get('filename')
|
|
if out:
|
|
break
|
|
filepath = Path(out) if out else None
|
|
|
|
await loop.run_in_executor(None, _do)
|
|
return filepath, info
|