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
|