import os import urllib from concurrent.futures import ThreadPoolExecutor from io import BytesIO from pathlib import Path import requests from PIL import Image, PngImagePlugin, UnidentifiedImageError from requests.exceptions import ConnectionError as RequestConnectionError from requests.exceptions import ReadTimeout from app import settings from app.db.libdata import ArtistTable # from app.store import artists as artist_store # from app.store.tracks import TrackStore from app.utils.hashing import create_hash from app.utils.progressbar import tqdm CHECK_ARTIST_IMAGES_KEY = "" LARGE_ENOUGH_NUMBER = 100 PngImagePlugin.MAX_TEXT_CHUNK = LARGE_ENOUGH_NUMBER * (1024**2) # https://stackoverflow.com/a/61466412 def get_artist_image_link(artist: str): """ Returns an artist image url. """ try: query = urllib.parse.quote(artist) # type: ignore url = f"https://api.deezer.com/search/artist?q={query}" response = requests.get(url, timeout=30) try: data = response.json() except requests.exceptions.JSONDecodeError: return None for res in data["data"]: res_hash = create_hash(res["name"], decode=True) artist_hash = create_hash(artist, decode=True) if res_hash == artist_hash: return str(res["picture_big"]) return None except (RequestConnectionError, ReadTimeout, IndexError, KeyError): return None # TODO: Move network calls to utils/network.py class DownloadImage: def __init__(self, url: str, name: str) -> None: img = self.download(url) if img is None: return sm_path = Path(settings.Paths.get_sm_artist_img_path()) / name lg_path = Path(settings.Paths.get_lg_artist_img_path()) / name md_path = Path(settings.Paths.get_md_artist_img_path()) / name entries = [ (lg_path, None), # save in the original size (sm_path, settings.Defaults.SM_ARTIST_IMG_SIZE), (md_path, settings.Defaults.MD_ARTIST_IMG_SIZE), ] self.save_img(img, entries) @staticmethod def download(url: str) -> Image.Image | None: """ Downloads the image from the url. """ try: return Image.open(BytesIO(requests.get(url, timeout=10).content)) except UnidentifiedImageError: return None @staticmethod def save_img(img: Image.Image, entries: list[tuple[Path, int | None]]): """ Saves the image to the destinations. """ ratio = img.width / img.height for entry in entries: path, size = entry if size is None: img.save(path, format="webp") continue img.resize((size, int(size / ratio)), Image.ANTIALIAS).save( path, format="webp" ) class CheckArtistImages: def __init__(self, instance_key: str): global CHECK_ARTIST_IMAGES_KEY CHECK_ARTIST_IMAGES_KEY = instance_key # read all files in the artist image folder path = settings.Paths.get_sm_artist_img_path() processed = [path.replace(".webp", "") for path in os.listdir(path)] unprocessed = ArtistTable.get_artisthashes_not_in(processed) key_artist_map = ((instance_key, artist) for artist in unprocessed) with ThreadPoolExecutor(max_workers=14) as executor: res = list( tqdm( executor.map(self.download_image, key_artist_map), total=len(unprocessed), desc="Downloading missing artist images", ) ) list(res) @staticmethod def download_image(_map: tuple[str, dict[str, str]]): """ Checks if an artist image exists and downloads it if not. :param artist: The artist name """ instance_key, artist = _map if CHECK_ARTIST_IMAGES_KEY != instance_key: return img_path = ( Path(settings.Paths.get_sm_artist_img_path()) / f"{artist['artisthash']}.webp" ) if img_path.exists(): return url = get_artist_image_link(artist["name"]) if url is not None: return DownloadImage(url, name=f"{artist['artisthash']}.webp") # def fetch_album_bio(title: str, albumartist: str) -> str | None: """ Returns the album bio for a given album. """ # last_fm_url = "http://ws.audioscrobbler.com/2.0/?method=album.getinfo&api_key={}&artist={}&album={ # }&format=json".format( settings.Paths.LAST_FM_API_KEY, albumartist, title ) # try: # response = requests.get(last_fm_url) # data = response.json() # except: # return None # try: # bio = data["album"]["wiki"]["summary"].split('