Add cover downloader, add cfg.save_{lyrics,cover} (mainly for testing)

backend/config.py

@@ -22,6 +22,9 @@ class Config:
         # Proxy URL for yt_proxied downloader (can be used for geo-restricted content)
         self.yt_proxy = os.getenv('YT_PROXY') or 'http://127.0.0.1:1080'
 
+        self.save_lyrics = _parse_bool(os.getenv('SAVE_LYRICS'), True)
+        self.save_cover = _parse_bool(os.getenv('SAVE_COVER'), True)
+
 
 _config: Config | None = None
 
@@ -31,3 +34,11 @@ def get() -> Config:
     if _config is None:
         _config = Config()
     return _config
+
+
+def _parse_bool(val: str | None, default: bool) -> bool:
+    if val is None:
+        return default
+    if val in {'false', 'off', 'no', '0'}:
+        return False
+    return bool(val)

backend/cover.py

@@ -0,0 +1,44 @@
+import mimetypes
+import subprocess
+from pathlib import Path
+
+import http_pool
+
+
+def download_from_yt(url: str, album_path: Path) -> None:
+    '''YouTube-specific cover art downloader.
+    See https://git.dc09.ru/DarkCat09/musicdlp/issues/1#issuecomment-202'''
+
+    path = album_path / 'cover.webp'
+    if path.exists():
+        return
+
+    ret = subprocess.call(
+        [
+            'ffmpeg',
+            '-nostdin',
+            '-i', url,
+            '-vf', 'crop=ih:ih',
+            str(album_path / 'cover.webp'),
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+    if ret != 0:
+        raise RuntimeError(f'FFmpeg returned {ret} exit code')
+
+
+def download(url: str, album_path: Path) -> None:
+    '''General cover art downloader'''
+
+    resp = http_pool.get().request('GET', url)
+    ext = mimetypes.guess_extension(resp.headers['content-type']) or '.jpg'
+
+    path = album_path / ('cover' + ext)
+    if path.exists():
+        return
+
+    with (album_path / ('cover' + ext)).open('wb') as f:
+        for chunk in resp.read_chunked():
+            f.write(chunk)

backend/id3pp.py

@@ -1,6 +1,9 @@
+from pathlib import Path
 from mutagen import mp3, id3
 from yt_dlp.postprocessor import PostProcessor
 
+import config
+import cover
 import genius
 
 
@@ -44,6 +47,10 @@ class ID3TagsPP(PostProcessor):
     '''Inserts ID3 tags after all PPs (for YT: InfoYouTubePP and FFmpegExtractAudioPP),
     triggers searching and parsing lyrics from Genius'''
 
+    def __init__(self) -> None:
+        self.cfg = config.get()
+        super().__init__()
+
     def run(self, information):
 
         file = mp3.MP3(information['filepath'])
@@ -62,6 +69,7 @@ class ID3TagsPP(PostProcessor):
         if 'genre' in information:
             file['TCON'] = id3.TCON(encoding=ENC_UTF8, text=information['genre'])
 
+        if self.cfg.save_lyrics:
             try:
                 lyr_title, lyr_url = genius.search(title, artists[0])
                 genius.raise_on_irrelevant_result(lyr_title, title, artists[0])
@@ -71,4 +79,14 @@ class ID3TagsPP(PostProcessor):
 
         file.save()
 
+        if self.cfg.save_cover:
+            try:
+                album_path = Path(information['filepath']).parent
+                if 'youtube' in information['extractor']:
+                    cover.download_from_yt(information['thumbnail'], album_path)
+                else:
+                    cover.download(information['thumbnail'], album_path)
+            except:
+                pass
+
         return [], information

backend/test_pp.py

@@ -5,8 +5,9 @@ import subprocess
 
 from mutagen import mp3
 
-import id3pp
+import config
 import http_pool
+import id3pp
 
 
 TEST_MP3 = 'music/test.mp3'
@@ -35,6 +36,10 @@ class TestPostProcessorsOnFakeData(TestCase):
 
     def setUp(self) -> None:
 
+        cfg = config.get()
+        cfg.save_lyrics = False
+        cfg.save_cover = False
+
         http_pool.get()
 
     def test_infoytpp(self) -> None: