75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from django.conf import settings
|
|
|
|
UCAST_DIRNAME = "_ucast"
|
|
|
|
|
|
class ChannelFolder:
|
|
def __init__(self, dir_root: Path):
|
|
self.dir_root = dir_root
|
|
dir_ucast = self.dir_root / UCAST_DIRNAME
|
|
|
|
self.file_avatar = dir_ucast / "avatar.jpg"
|
|
self.file_avatar_sm = dir_ucast / "avatar_sm.webp"
|
|
|
|
self.dir_covers = dir_ucast / "covers"
|
|
self.dir_thumbnails = dir_ucast / "thumbnails"
|
|
|
|
@staticmethod
|
|
def _glob_file(parent_dir: Path, glob: str, default_filename: str = None) -> Path:
|
|
try:
|
|
return parent_dir.glob(glob).__next__()
|
|
except StopIteration:
|
|
if default_filename:
|
|
return parent_dir / default_filename
|
|
raise FileNotFoundError(f"file {str(parent_dir)}/{glob} not found")
|
|
|
|
def does_exist(self) -> bool:
|
|
return os.path.isdir(self.dir_covers)
|
|
|
|
def create(self):
|
|
os.makedirs(self.dir_covers, exist_ok=True)
|
|
os.makedirs(self.dir_thumbnails, exist_ok=True)
|
|
|
|
def get_cover(self, title_slug: str) -> Path:
|
|
return self.dir_covers / f"{title_slug}.png"
|
|
|
|
def get_thumbnail(self, title_slug: str, sm=False) -> Path:
|
|
filename = title_slug
|
|
if sm:
|
|
filename += "_sm"
|
|
|
|
return self._glob_file(self.dir_thumbnails, f"{filename}.*", f"{filename}.webp")
|
|
|
|
def get_audio(self, title_slug: str) -> Path:
|
|
return self.dir_root / f"{title_slug}.mp3"
|
|
|
|
|
|
class Storage:
|
|
def __init__(self):
|
|
self.dir_data = settings.DOWNLOAD_ROOT
|
|
|
|
def get_channel_folder(self, channel_slug: str) -> ChannelFolder:
|
|
cf = ChannelFolder(self.dir_data / channel_slug)
|
|
if not cf.does_exist():
|
|
raise FileNotFoundError
|
|
return cf
|
|
|
|
def get_or_create_channel_folder(self, channel_slug: str) -> ChannelFolder:
|
|
cf = ChannelFolder(self.dir_data / channel_slug)
|
|
if not cf.does_exist():
|
|
cf.create()
|
|
return cf
|
|
|
|
|
|
class Cache:
|
|
def __init__(self):
|
|
self.dir_cache = settings.CACHE_ROOT
|
|
self.dir_ytdlp_cache = self.dir_cache / "yt_dlp"
|
|
os.makedirs(self.dir_ytdlp_cache, exist_ok=True)
|
|
|
|
def create_tmpdir(self, prefix="dld") -> tempfile.TemporaryDirectory:
|
|
return tempfile.TemporaryDirectory(prefix=prefix + "_", dir=self.dir_cache)
|