""" This module contains functions for the server """ import os import time from concurrent.futures import ThreadPoolExecutor from io import BytesIO import requests from app import helpers from app import settings from app.lib import trackslib from app.lib import watchdoge from app.lib.albumslib import ValidateAlbumThumbs from app.lib.colorlib import ProcessAlbumColors from app.lib.playlistlib import ValidatePlaylistThumbs from app.lib.populate import CreateAlbums from app.lib.populate import Populate from app.logger import get_logger from PIL import Image log = get_logger() @helpers.background def run_checks(): """ Checks for new songs every 5 minutes. """ ValidateAlbumThumbs() while True: trackslib.validate_tracks() Populate() CreateAlbums() if helpers.Ping()(): CheckArtistImages()() @helpers.background def process_album_colors(): ProcessAlbumColors() ValidatePlaylistThumbs() process_album_colors() time.sleep(300) @helpers.background def start_watchdog(): """ Starts the file watcher. """ watchdoge.watch.run() class getArtistImage: """ Returns an artist image url. """ def __init__(self, artist: str): self.artist = artist def __call__(self): try: url = f"https://api.deezer.com/search/artist?q={self.artist}" response = requests.get(url) data = response.json() return data["data"][0]["picture_medium"] except requests.exceptions.ConnectionError: time.sleep(5) return None except (IndexError, KeyError): return None class useImageDownloader: def __init__(self, url: str, dest: str) -> None: self.url = url self.dest = dest def __call__(self) -> None: try: img = Image.open(BytesIO(requests.get(self.url).content)) img.save(self.dest, format="webp") img.close() return "fetched image" except requests.exceptions.ConnectionError: time.sleep(5) return "connection error" class CheckArtistImages: def __init__(self): self.artists: list[str] = [] print("Checking for artist images") log.info("Checking artist images") @staticmethod def check_if_exists(img_path: str): """ Checks if an image exists on disk. """ if os.path.exists(img_path): return True else: return False @classmethod def download_image(cls, artistname: str): """ Checks if an artist image exists and downloads it if not. :param artistname: The artist name """ img_path = ( settings.APP_DIR + "/images/artists/" + helpers.create_safe_name(artistname) + ".webp" ) if cls.check_if_exists(img_path): return "exists" url = getArtistImage(artistname)() if url is None: return "url is none" return useImageDownloader(url, img_path)() def __call__(self): self.artists = helpers.Get.get_all_artists() with ThreadPoolExecutor() as pool: iter = pool.map(self.download_image, self.artists) [i for i in iter] print("Done fetching images") 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.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('