2022-05-24 01:53:05 +00:00
|
|
|
"""
|
|
|
|
functionality:
|
|
|
|
- base class to make all calls to yt-dlp
|
|
|
|
- handle yt-dlp errors
|
|
|
|
"""
|
|
|
|
|
2022-05-24 08:51:58 +00:00
|
|
|
import os
|
2022-06-15 03:39:48 +00:00
|
|
|
from http import cookiejar
|
2022-05-24 08:51:58 +00:00
|
|
|
from io import StringIO
|
|
|
|
|
2022-05-24 01:53:05 +00:00
|
|
|
import yt_dlp
|
2022-05-24 08:51:58 +00:00
|
|
|
from home.src.ta.ta_redis import RedisArchivist
|
2022-05-24 01:53:05 +00:00
|
|
|
|
|
|
|
|
|
|
|
class YtWrap:
|
|
|
|
"""wrap calls to yt"""
|
|
|
|
|
2022-05-24 03:00:40 +00:00
|
|
|
OBS_BASE = {
|
|
|
|
"default_search": "ytsearch",
|
|
|
|
"quiet": True,
|
|
|
|
"check_formats": "selected",
|
|
|
|
"socket_timeout": 2,
|
|
|
|
}
|
|
|
|
|
2022-05-24 08:51:58 +00:00
|
|
|
def __init__(self, obs_request, config=False):
|
2022-05-24 03:00:40 +00:00
|
|
|
self.obs_request = obs_request
|
2022-05-24 08:51:58 +00:00
|
|
|
self.config = config
|
2022-05-24 03:00:40 +00:00
|
|
|
self.build_obs()
|
|
|
|
|
|
|
|
def build_obs(self):
|
|
|
|
"""build yt-dlp obs"""
|
|
|
|
self.obs = self.OBS_BASE.copy()
|
|
|
|
self.obs.update(self.obs_request)
|
2022-05-30 11:22:17 +00:00
|
|
|
if self.config:
|
|
|
|
self.add_cookie()
|
2022-05-24 08:51:58 +00:00
|
|
|
|
|
|
|
def add_cookie(self):
|
|
|
|
"""add cookie if enabled"""
|
|
|
|
if self.config["downloads"]["cookie_import"]:
|
|
|
|
cookie_io = CookieHandler(self.config).get()
|
|
|
|
self.obs["cookiefile"] = cookie_io
|
2022-05-24 01:53:05 +00:00
|
|
|
|
|
|
|
def download(self, url):
|
|
|
|
"""make download request"""
|
|
|
|
with yt_dlp.YoutubeDL(self.obs) as ydl:
|
|
|
|
try:
|
|
|
|
ydl.download([url])
|
|
|
|
except yt_dlp.utils.DownloadError:
|
2022-05-24 03:18:33 +00:00
|
|
|
print(f"{url}: failed to download.")
|
|
|
|
return False
|
|
|
|
|
|
|
|
return True
|
2022-05-24 01:53:05 +00:00
|
|
|
|
|
|
|
def extract(self, url):
|
|
|
|
"""make extract request"""
|
2022-05-24 03:00:40 +00:00
|
|
|
try:
|
|
|
|
response = yt_dlp.YoutubeDL(self.obs).extract_info(url)
|
2022-06-15 03:39:48 +00:00
|
|
|
except cookiejar.LoadError:
|
|
|
|
print("cookie file is invalid")
|
|
|
|
return False
|
2022-05-24 03:00:40 +00:00
|
|
|
except (yt_dlp.utils.ExtractorError, yt_dlp.utils.DownloadError):
|
|
|
|
print(f"{url}: failed to get info from youtube")
|
2022-06-15 03:39:48 +00:00
|
|
|
return False
|
2022-05-24 01:53:05 +00:00
|
|
|
|
|
|
|
return response
|
2022-05-24 08:51:58 +00:00
|
|
|
|
|
|
|
|
|
|
|
class CookieHandler:
|
|
|
|
"""handle youtube cookie for yt-dlp"""
|
|
|
|
|
|
|
|
def __init__(self, config):
|
|
|
|
self.cookie_io = False
|
|
|
|
self.config = config
|
|
|
|
|
|
|
|
def get(self):
|
|
|
|
"""get cookie io stream"""
|
|
|
|
cookie = RedisArchivist().get_message("cookie")
|
|
|
|
self.cookie_io = StringIO(cookie)
|
|
|
|
return self.cookie_io
|
|
|
|
|
|
|
|
def import_cookie(self):
|
|
|
|
"""import cookie from file"""
|
|
|
|
cache_path = self.config["application"]["cache_dir"]
|
|
|
|
import_path = os.path.join(cache_path, "import", "cookies.google.txt")
|
|
|
|
with open(import_path, encoding="utf-8") as cookie_file:
|
|
|
|
cookie = cookie_file.read()
|
|
|
|
|
|
|
|
RedisArchivist().set_message("cookie", cookie, expire=False)
|
|
|
|
|
|
|
|
os.remove(import_path)
|
|
|
|
print("cookie: import successful")
|
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
def revoke():
|
|
|
|
"""revoke cookie"""
|
|
|
|
RedisArchivist().del_message("cookie")
|
2022-06-15 08:39:24 +00:00
|
|
|
RedisArchivist().set_message(
|
|
|
|
"config", False, path=".downloads.cookie_import"
|
|
|
|
)
|
2022-05-24 08:51:58 +00:00
|
|
|
print("cookie: revoked")
|
|
|
|
|
|
|
|
def validate(self):
|
|
|
|
"""validate cookie using the liked videos playlist"""
|
|
|
|
obs_request = {
|
|
|
|
"skip_download": True,
|
|
|
|
"extract_flat": True,
|
|
|
|
}
|
|
|
|
response = YtWrap(obs_request, self.config).extract("LL")
|
|
|
|
return bool(response)
|