mirror of
https://github.com/Dvorinka/swingmusic-extended.git
synced 2026-06-03 20:13:02 +00:00
Save complete tracks and albums to the db for faster startup
- refactor function locations - add logger - check for new tracks instead of re-processing all files
This commit is contained in:
@@ -13,13 +13,13 @@ from app.lib import folderslib
|
||||
from app.lib import playlistlib
|
||||
|
||||
|
||||
PRE_TRACKS = instances.songs_instance.get_all_songs()
|
||||
DB_TRACKS = instances.tracks_instance.get_all_tracks()
|
||||
VALID_FOLDERS: Set[str] = set()
|
||||
|
||||
ALBUMS: List[models.Album] = []
|
||||
TRACKS: List[models.Track] = []
|
||||
PLAYLISTS: List[models.Playlist] = []
|
||||
FOLDERS: List[models.Folder] = []
|
||||
FOLDERS: Set[models.Folder] = set()
|
||||
|
||||
|
||||
@helpers.background
|
||||
|
||||
@@ -23,7 +23,7 @@ def get_albums():
|
||||
"""returns all the albums"""
|
||||
albums = []
|
||||
|
||||
for song in api.PRE_TRACKS:
|
||||
for song in api.DB_TRACKS:
|
||||
al_obj = {"name": song["album"], "artist": song["artists"]}
|
||||
|
||||
if al_obj not in albums:
|
||||
@@ -41,7 +41,8 @@ def get_album_tracks():
|
||||
artist = data["artist"]
|
||||
|
||||
songs = trackslib.get_album_tracks(album, artist)
|
||||
album = albumslib.find_album(album, artist)
|
||||
index = albumslib.find_album(album, artist)
|
||||
album = api.ALBUMS[index]
|
||||
|
||||
return {"songs": songs, "info": album}
|
||||
|
||||
@@ -50,9 +51,8 @@ def get_album_tracks():
|
||||
def get_album_bio():
|
||||
"""Returns the album bio for the given album."""
|
||||
data = request.get_json()
|
||||
print(data)
|
||||
|
||||
bio = functions.get_album_bio(data["album"], data["albumartist"])
|
||||
bio = functions.fetch_album_bio(data["album"], data["albumartist"])
|
||||
|
||||
if bio is not None:
|
||||
return {"bio": bio}
|
||||
|
||||
@@ -1,25 +0,0 @@
|
||||
import os
|
||||
import urllib
|
||||
from typing import List
|
||||
from flask import request, send_file
|
||||
|
||||
from app import functions, instances, helpers, cache, db, prep
|
||||
from app import api
|
||||
|
||||
|
||||
home_dir = helpers.home_dir
|
||||
|
||||
# @api.bp.route("/populate")
|
||||
# def find_tracks():
|
||||
# """call the populate function"""
|
||||
# functions.populate()
|
||||
# return "🎸"
|
||||
|
||||
|
||||
# @api.bp.route("/populate/images")
|
||||
# def populate_images():
|
||||
# """
|
||||
# Populates the artist images.
|
||||
# """
|
||||
# functions.populate_images()
|
||||
# return "Done"
|
||||
@@ -21,7 +21,7 @@ def get_artist_data(artist: str):
|
||||
artist_obj = instances.artist_instance.get_artists_by_name(artist)
|
||||
|
||||
def get_artist_tracks():
|
||||
songs = instances.songs_instance.find_songs_by_artist(artist)
|
||||
songs = instances.tracks_instance.find_songs_by_artist(artist)
|
||||
|
||||
return songs
|
||||
|
||||
@@ -32,7 +32,7 @@ def get_artist_data(artist: str):
|
||||
artist_albums = []
|
||||
albums_with_count = []
|
||||
|
||||
albums = instances.songs_instance.find_songs_by_albumartist(artist)
|
||||
albums = instances.tracks_instance.find_songs_by_albumartist(artist)
|
||||
|
||||
for song in albums:
|
||||
if song["album"] not in artist_albums:
|
||||
|
||||
@@ -16,7 +16,7 @@ def send_track_file(trackid):
|
||||
"""
|
||||
try:
|
||||
filepath = [
|
||||
file["filepath"] for file in api.PRE_TRACKS
|
||||
file["filepath"] for file in api.DB_TRACKS
|
||||
if file["_id"]["$oid"] == trackid
|
||||
][0]
|
||||
except (FileNotFoundError, IndexError) as e:
|
||||
@@ -31,5 +31,5 @@ def get_sample_track():
|
||||
Returns a sample track object.
|
||||
"""
|
||||
|
||||
return instances.songs_instance.get_song_by_album("Legends Never Die",
|
||||
return instances.tracks_instance.get_song_by_album("Legends Never Die",
|
||||
"Juice WRLD")
|
||||
|
||||
@@ -5,6 +5,8 @@ album documents in MongoDB.
|
||||
from app import db
|
||||
from bson import ObjectId
|
||||
|
||||
from app.models import Album
|
||||
|
||||
convert_many = db.convert_many
|
||||
convert_one = db.convert_one
|
||||
|
||||
@@ -18,13 +20,13 @@ class Albums(db.Mongo):
|
||||
super(Albums, self).__init__("ALICE_ALBUMS")
|
||||
self.collection = self.db["ALL_ALBUMS"]
|
||||
|
||||
def insert_album(self, album: dict) -> None:
|
||||
def insert_album(self, album: Album) -> None:
|
||||
"""
|
||||
Inserts a new album object into the database.
|
||||
"""
|
||||
return self.collection.update_one(
|
||||
{
|
||||
"album": album["album"],
|
||||
"album": album["title"],
|
||||
"artist": album["artist"]
|
||||
},
|
||||
{
|
||||
|
||||
@@ -31,7 +31,7 @@ class AllSongs(db.Mongo):
|
||||
},
|
||||
upsert=True).upserted_id
|
||||
|
||||
def get_all_songs(self) -> list:
|
||||
def get_all_tracks(self) -> list:
|
||||
"""
|
||||
Returns all tracks in the database.
|
||||
"""
|
||||
|
||||
+79
-261
@@ -1,15 +1,14 @@
|
||||
"""
|
||||
This module contains functions for the server
|
||||
"""
|
||||
from dataclasses import asdict
|
||||
import datetime
|
||||
import os
|
||||
import random
|
||||
import time
|
||||
import urllib
|
||||
from io import BytesIO
|
||||
from typing import List
|
||||
|
||||
import mutagen
|
||||
import requests
|
||||
from app import api
|
||||
from app import helpers
|
||||
@@ -18,15 +17,12 @@ from app import models
|
||||
from app import settings
|
||||
from app.lib import albumslib
|
||||
from app.lib import folderslib
|
||||
from app.lib import playlistlib
|
||||
from app.lib import watchdoge
|
||||
from mutagen.flac import FLAC
|
||||
from mutagen.flac import MutagenError
|
||||
from mutagen.id3 import ID3
|
||||
from PIL import Image
|
||||
from progress.bar import Bar
|
||||
|
||||
# from pprint import pprint
|
||||
from app.logger import Log
|
||||
from app.lib.taglib import get_tags, return_album_art
|
||||
|
||||
|
||||
@helpers.background
|
||||
@@ -34,13 +30,12 @@ def reindex_tracks():
|
||||
"""
|
||||
Checks for new songs every 5 minutes.
|
||||
"""
|
||||
flag = False
|
||||
|
||||
while flag is False:
|
||||
while True:
|
||||
populate()
|
||||
populate_images()
|
||||
fetch_artist_images()
|
||||
|
||||
time.sleep(300)
|
||||
time.sleep(60)
|
||||
|
||||
|
||||
@helpers.background
|
||||
@@ -60,33 +55,84 @@ def populate():
|
||||
extract it.
|
||||
"""
|
||||
start = time.time()
|
||||
db_tracks = instances.tracks_instance.get_all_tracks()
|
||||
tagged_tracks = []
|
||||
albums = []
|
||||
folders = set()
|
||||
|
||||
s, files = helpers.run_fast_scandir(settings.HOME_DIR, [".flac", ".mp3"],
|
||||
full=True)
|
||||
files = helpers.run_fast_scandir(settings.HOME_DIR, [".flac", ".mp3"], full=True)[1]
|
||||
|
||||
_bar = Bar("Processing files", max=len(files))
|
||||
_bar = Bar("Checking files", max=len(files))
|
||||
for track in db_tracks:
|
||||
if track["filepath"] in files:
|
||||
files.remove(track["filepath"])
|
||||
_bar.next()
|
||||
|
||||
_bar.finish()
|
||||
|
||||
Log(f"Found {len(files)} untagged files")
|
||||
|
||||
_bar = Bar("Tagging files", max=len(files))
|
||||
for file in files:
|
||||
tags = get_tags(file)
|
||||
foldername = os.path.dirname(file)
|
||||
folders.add(foldername)
|
||||
|
||||
if tags is not None:
|
||||
upsert_id = instances.songs_instance.insert_song(tags)
|
||||
|
||||
if upsert_id is not None:
|
||||
tags["_id"] = {"$oid": str(upsert_id)}
|
||||
api.PRE_TRACKS.append(tags)
|
||||
tagged_tracks.append(tags)
|
||||
api.DB_TRACKS.append(tags)
|
||||
|
||||
_bar.next()
|
||||
_bar.finish()
|
||||
|
||||
albumslib.create_everything()
|
||||
folderslib.run_scandir()
|
||||
Log(f"Tagged {len(tagged_tracks)} tracks")
|
||||
|
||||
_bar = Bar("Creating stuff", max=len(tagged_tracks))
|
||||
for track in tagged_tracks:
|
||||
albumindex = albumslib.find_album(track["album"], track["albumartist"])
|
||||
album = None
|
||||
|
||||
if albumindex is None:
|
||||
album = albumslib.create_album(track)
|
||||
api.ALBUMS.append(album)
|
||||
albums.append(album)
|
||||
instances.album_instance.insert_album(asdict(album))
|
||||
else:
|
||||
album = api.ALBUMS[albumindex]
|
||||
|
||||
track["image"] = album.image
|
||||
upsert_id = instances.tracks_instance.insert_song(track)
|
||||
|
||||
track["_id"] = {"$oid": str(upsert_id)}
|
||||
api.TRACKS.append(models.Track(track))
|
||||
|
||||
_bar.next()
|
||||
|
||||
_bar.finish()
|
||||
|
||||
Log(f"Added {len(tagged_tracks)} new tracks and {len(albums)} new albums")
|
||||
|
||||
_bar = Bar("Creating folders", max=len(folders))
|
||||
for folder in folders:
|
||||
if folder not in api.VALID_FOLDERS:
|
||||
api.VALID_FOLDERS.add(folder)
|
||||
fff = folderslib.create_folder(folder)
|
||||
api.FOLDERS.add(fff)
|
||||
|
||||
_bar.next()
|
||||
|
||||
_bar.finish()
|
||||
|
||||
Log(f"Created {len(api.FOLDERS)} folders")
|
||||
|
||||
end = time.time()
|
||||
|
||||
print(
|
||||
str(datetime.timedelta(seconds=round(end - start))) + " elapsed for " +
|
||||
str(len(files)) + " files")
|
||||
str(datetime.timedelta(seconds=round(end - start)))
|
||||
+ " elapsed for "
|
||||
+ str(len(files))
|
||||
+ " files"
|
||||
)
|
||||
|
||||
|
||||
def fetch_image_path(artist: str) -> str or None:
|
||||
@@ -107,12 +153,12 @@ def fetch_image_path(artist: str) -> str or None:
|
||||
return None
|
||||
|
||||
|
||||
def populate_images():
|
||||
"""populates the artists images"""
|
||||
def fetch_artist_images():
|
||||
"""Downloads the artists images"""
|
||||
|
||||
artists = []
|
||||
|
||||
for song in api.PRE_TRACKS:
|
||||
for song in api.DB_TRACKS:
|
||||
this_artists = song["artists"].split(", ")
|
||||
|
||||
for artist in this_artists:
|
||||
@@ -121,8 +167,9 @@ def populate_images():
|
||||
|
||||
_bar = Bar("Processing images", max=len(artists))
|
||||
for artist in artists:
|
||||
file_path = (helpers.app_dir + "/images/artists/" +
|
||||
artist.replace("/", "::") + ".webp")
|
||||
file_path = (
|
||||
helpers.app_dir + "/images/artists/" + artist.replace("/", "::") + ".webp"
|
||||
)
|
||||
|
||||
if not os.path.exists(file_path):
|
||||
img_path = fetch_image_path(artist)
|
||||
@@ -139,221 +186,13 @@ def populate_images():
|
||||
_bar.finish()
|
||||
|
||||
|
||||
def use_defaults() -> str:
|
||||
"""
|
||||
Returns a path to a random image in the defaults directory.
|
||||
"""
|
||||
path = "defaults/" + str(random.randint(0, 20)) + ".webp"
|
||||
return path
|
||||
|
||||
|
||||
def save_track_colors(img, filepath) -> None:
|
||||
"""Saves the track colors to the database"""
|
||||
|
||||
track_colors = helpers.extract_colors(img)
|
||||
|
||||
tc_dict = {
|
||||
"filepath": filepath,
|
||||
"colors": track_colors,
|
||||
}
|
||||
|
||||
instances.track_color_instance.insert_track_color(tc_dict)
|
||||
|
||||
|
||||
def return_album_art(filepath):
|
||||
"""
|
||||
Returns the album art for a given audio file.
|
||||
"""
|
||||
|
||||
if filepath.endswith(".flac"):
|
||||
try:
|
||||
audio = FLAC(filepath)
|
||||
return audio.pictures[0].data
|
||||
except:
|
||||
return None
|
||||
elif filepath.endswith(".mp3"):
|
||||
try:
|
||||
audio = ID3(filepath)
|
||||
return audio.getall("APIC")[0].data
|
||||
except:
|
||||
return None
|
||||
|
||||
|
||||
def save_t_colors():
|
||||
_bar = Bar("Processing image colors", max=len(api.PRE_TRACKS))
|
||||
|
||||
for track in api.PRE_TRACKS:
|
||||
filepath = track["filepath"]
|
||||
album_art = return_album_art(filepath)
|
||||
|
||||
if album_art is not None:
|
||||
img = Image.open(BytesIO(album_art))
|
||||
save_track_colors(img, filepath)
|
||||
|
||||
_bar.next()
|
||||
|
||||
_bar.finish()
|
||||
|
||||
|
||||
def extract_thumb(audio_file_path: str, webp_path: str) -> str:
|
||||
"""
|
||||
Extracts the thumbnail from an audio file. Returns the path to the thumbnail.
|
||||
"""
|
||||
img_path = os.path.join(settings.THUMBS_PATH, webp_path)
|
||||
|
||||
if os.path.exists(img_path):
|
||||
return urllib.parse.quote(webp_path)
|
||||
|
||||
album_art = return_album_art(audio_file_path)
|
||||
|
||||
if album_art is not None:
|
||||
img = Image.open(BytesIO(album_art))
|
||||
|
||||
try:
|
||||
small_img = img.resize((250, 250), Image.ANTIALIAS)
|
||||
small_img.save(img_path, format="webp")
|
||||
except OSError:
|
||||
try:
|
||||
png = img.convert("RGB")
|
||||
small_img = png.resize((250, 250), Image.ANTIALIAS)
|
||||
small_img.save(webp_path, format="webp")
|
||||
except:
|
||||
return None
|
||||
|
||||
return urllib.parse.quote(webp_path)
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
def parse_artist_tag(audio):
|
||||
"""
|
||||
Parses the artist tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
artists = audio["artist"][0]
|
||||
except (KeyError, IndexError):
|
||||
artists = "Unknown"
|
||||
|
||||
return artists
|
||||
|
||||
|
||||
def parse_title_tag(audio, full_path: str):
|
||||
"""
|
||||
Parses the title tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
title = audio["title"][0]
|
||||
except (KeyError, IndexError):
|
||||
title = full_path.split("/")[-1]
|
||||
|
||||
return title
|
||||
|
||||
|
||||
def parse_album_artist_tag(audio):
|
||||
"""
|
||||
Parses the album artist tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
albumartist = audio["albumartist"][0]
|
||||
except (KeyError, IndexError):
|
||||
albumartist = "Unknown"
|
||||
|
||||
return albumartist
|
||||
|
||||
|
||||
def parse_album_tag(audio, full_path: str):
|
||||
"""
|
||||
Parses the album tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
album = audio["album"][0]
|
||||
except (KeyError, IndexError):
|
||||
album = full_path.split("/")[-1]
|
||||
|
||||
return album
|
||||
|
||||
|
||||
def parse_genre_tag(audio):
|
||||
"""
|
||||
Parses the genre tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
genre = audio["genre"][0]
|
||||
except (KeyError, IndexError):
|
||||
genre = "Unknown"
|
||||
|
||||
return genre
|
||||
|
||||
|
||||
def parse_date_tag(audio):
|
||||
"""
|
||||
Parses the date tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
date = audio["date"][0]
|
||||
except (KeyError, IndexError):
|
||||
date = "Unknown"
|
||||
|
||||
return date
|
||||
|
||||
|
||||
def parse_track_number(audio):
|
||||
"""
|
||||
Parses the track number from an audio file.
|
||||
"""
|
||||
try:
|
||||
track_number = audio["tracknumber"][0]
|
||||
except (KeyError, IndexError):
|
||||
track_number = "Unknown"
|
||||
|
||||
return track_number
|
||||
|
||||
|
||||
def parse_disk_number(audio):
|
||||
"""
|
||||
Parses the disk number from an audio file.
|
||||
"""
|
||||
try:
|
||||
disk_number = audio["discnumber"][0]
|
||||
except (KeyError, IndexError):
|
||||
disk_number = "Unknown"
|
||||
|
||||
return disk_number
|
||||
|
||||
|
||||
def get_tags(fullpath: str) -> dict:
|
||||
"""
|
||||
Returns a dictionary of tags for a given file.
|
||||
"""
|
||||
try:
|
||||
audio = mutagen.File(fullpath, easy=True)
|
||||
except MutagenError:
|
||||
return None
|
||||
|
||||
tags = {
|
||||
"artists": parse_artist_tag(audio),
|
||||
"title": parse_title_tag(audio, fullpath),
|
||||
"albumartist": parse_album_artist_tag(audio),
|
||||
"album": parse_album_tag(audio, fullpath),
|
||||
"genre": parse_genre_tag(audio),
|
||||
"date": parse_date_tag(audio)[:4],
|
||||
"tracknumber": parse_track_number(audio),
|
||||
"discnumber": parse_disk_number(audio),
|
||||
"length": round(audio.info.length),
|
||||
"bitrate": round(int(audio.info.bitrate) / 1000),
|
||||
"filepath": fullpath,
|
||||
"folder": os.path.dirname(fullpath),
|
||||
}
|
||||
|
||||
return tags
|
||||
|
||||
|
||||
def get_album_bio(title: str, albumartist: str):
|
||||
def fetch_album_bio(title: str, albumartist: str):
|
||||
"""
|
||||
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)
|
||||
settings.LAST_FM_API_KEY, albumartist, title
|
||||
)
|
||||
|
||||
try:
|
||||
response = requests.get(last_fm_url)
|
||||
@@ -362,29 +201,8 @@ def get_album_bio(title: str, albumartist: str):
|
||||
return None
|
||||
|
||||
try:
|
||||
bio = data["album"]["wiki"]["summary"].split(
|
||||
'<a href="https://www.last.fm/')[0]
|
||||
bio = data["album"]["wiki"]["summary"].split('<a href="https://www.last.fm/')[0]
|
||||
except KeyError:
|
||||
bio = None
|
||||
|
||||
return bio
|
||||
|
||||
|
||||
def get_all_albums() -> List[models.Album]:
|
||||
"""
|
||||
Returns a list of album objects for all albums in the database.
|
||||
"""
|
||||
|
||||
albums: List[models.Album] = []
|
||||
|
||||
_bar = Bar("Creating albums", max=len(api.PRE_TRACKS))
|
||||
for track in api.PRE_TRACKS:
|
||||
xx = albumslib.create_album(track)
|
||||
if xx not in albums:
|
||||
albums.append(xx)
|
||||
|
||||
_bar.next()
|
||||
|
||||
_bar.finish()
|
||||
|
||||
return albums
|
||||
|
||||
+3
-19
@@ -6,8 +6,8 @@ import datetime
|
||||
import os
|
||||
import random
|
||||
import threading
|
||||
from typing import List
|
||||
import colorgram, time
|
||||
from typing import Dict, List
|
||||
import time
|
||||
|
||||
from app import models, settings
|
||||
|
||||
@@ -26,7 +26,7 @@ def background(func):
|
||||
return background_func
|
||||
|
||||
|
||||
def run_fast_scandir(__dir: str, ext: list, full=False):
|
||||
def run_fast_scandir(__dir: str, ext: list, full=False) -> Dict[List[str], List[str]]:
|
||||
"""
|
||||
Scans a directory for files with a specific extension. Returns a list of files and folders in the directory.
|
||||
"""
|
||||
@@ -93,22 +93,6 @@ def is_valid_file(filename: str) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def extract_image_colors(image) -> list:
|
||||
"""Extracts 2 of the most dominant colors from an image."""
|
||||
try:
|
||||
colors = sorted(colorgram.extract(image, 2), key=lambda c: c.hsl.h)
|
||||
except OSError:
|
||||
return []
|
||||
|
||||
formatted_colors = []
|
||||
|
||||
for color in colors:
|
||||
color = f"rgb({color.rgb.r}, {color.rgb.g}, {color.rgb.b})"
|
||||
formatted_colors.append(color)
|
||||
|
||||
return formatted_colors
|
||||
|
||||
|
||||
def use_memoji():
|
||||
"""
|
||||
Returns a path to a random memoji image.
|
||||
|
||||
@@ -4,7 +4,7 @@ All the MongoDB instances are created here.
|
||||
|
||||
from app.db import artists, albums, trackcolors, tracks, playlists
|
||||
|
||||
songs_instance = tracks.AllSongs()
|
||||
tracks_instance = tracks.AllSongs()
|
||||
artist_instance = artists.Artists()
|
||||
track_color_instance = trackcolors.TrackColors()
|
||||
album_instance = albums.Albums()
|
||||
|
||||
+65
-15
@@ -1,14 +1,37 @@
|
||||
"""
|
||||
This library contains all the functions related to albums.
|
||||
"""
|
||||
import random
|
||||
import urllib
|
||||
from pprint import pprint
|
||||
from typing import List
|
||||
from progress.bar import Bar
|
||||
|
||||
from app import api
|
||||
from app import functions
|
||||
from app import models
|
||||
from app.lib import trackslib
|
||||
from app import instances
|
||||
|
||||
|
||||
def get_all_albums() -> List[models.Album]:
|
||||
"""
|
||||
Returns a list of album objects for all albums in the database.
|
||||
"""
|
||||
print("Getting all albums...")
|
||||
|
||||
albums: List[models.Album] = []
|
||||
db_albums = instances.album_instance.get_all_albums()
|
||||
|
||||
_bar = Bar("Creating albums", max=len(db_albums))
|
||||
for album in db_albums:
|
||||
aa = models.Album(album)
|
||||
albums.append(aa)
|
||||
_bar.next()
|
||||
|
||||
_bar.finish()
|
||||
|
||||
return albums
|
||||
|
||||
|
||||
def create_everything() -> List[models.Track]:
|
||||
@@ -16,15 +39,39 @@ def create_everything() -> List[models.Track]:
|
||||
Creates album objects for all albums and returns
|
||||
a list of track objects
|
||||
"""
|
||||
albums: list[models.Album] = functions.get_all_albums()
|
||||
albums: list[models.Album] = get_all_albums()
|
||||
|
||||
api.ALBUMS.clear()
|
||||
api.ALBUMS.extend(albums)
|
||||
api.ALBUMS = albums
|
||||
api.ALBUMS.sort(key=lambda x: x.title)
|
||||
|
||||
tracks = trackslib.create_all_tracks()
|
||||
|
||||
api.TRACKS.clear()
|
||||
api.TRACKS.extend(tracks)
|
||||
api.TRACKS.sort(key=lambda x: x.title)
|
||||
|
||||
|
||||
def find_album(albumtitle: str, artist: str) -> models.Album:
|
||||
"""
|
||||
Finds an album by album title and artist.
|
||||
"""
|
||||
left = 0
|
||||
right = len(api.ALBUMS) - 1
|
||||
iter = 0
|
||||
|
||||
while left <= right:
|
||||
iter += 1
|
||||
mid = (left + right) // 2
|
||||
|
||||
if api.ALBUMS[mid].title == albumtitle and api.ALBUMS[mid].artist == artist:
|
||||
return mid
|
||||
|
||||
if api.ALBUMS[mid].title < albumtitle:
|
||||
left = mid + 1
|
||||
else:
|
||||
right = mid - 1
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def get_album_duration(album: list) -> int:
|
||||
@@ -40,32 +87,40 @@ def get_album_duration(album: list) -> int:
|
||||
return album_duration
|
||||
|
||||
|
||||
def use_defaults() -> str:
|
||||
"""
|
||||
Returns a path to a random image in the defaults directory.
|
||||
"""
|
||||
path = "defaults/" + str(random.randint(0, 20)) + ".webp"
|
||||
return path
|
||||
|
||||
|
||||
def get_album_image(album: list) -> str:
|
||||
"""
|
||||
Gets the image of an album.
|
||||
"""
|
||||
|
||||
for track in album:
|
||||
img_p = (track["album"] + track["albumartist"] + ".webp").replace(
|
||||
"/", "::")
|
||||
img_p = (track["album"] + track["albumartist"] + ".webp").replace("/", "::")
|
||||
img = functions.extract_thumb(track["filepath"], webp_path=img_p)
|
||||
|
||||
if img is not None:
|
||||
return img
|
||||
|
||||
return functions.use_defaults()
|
||||
return use_defaults()
|
||||
|
||||
|
||||
def get_album_tracks(album: str, artist: str) -> List:
|
||||
tracks = []
|
||||
|
||||
for track in api.PRE_TRACKS:
|
||||
for track in api.DB_TRACKS:
|
||||
try:
|
||||
if track["album"] == album and track["albumartist"] == artist:
|
||||
tracks.append(track)
|
||||
except TypeError:
|
||||
pprint(track, indent=4)
|
||||
print(album, artist)
|
||||
|
||||
return tracks
|
||||
|
||||
|
||||
@@ -85,19 +140,14 @@ def create_album(track) -> models.Album:
|
||||
album["date"] = album_tracks[0]["date"]
|
||||
|
||||
album["artistimage"] = urllib.parse.quote_plus(
|
||||
album_tracks[0]["albumartist"] + ".webp")
|
||||
album_tracks[0]["albumartist"] + ".webp"
|
||||
)
|
||||
|
||||
album["image"] = get_album_image(album_tracks)
|
||||
|
||||
return models.Album(album)
|
||||
|
||||
|
||||
def find_album(albumtitle, artist):
|
||||
for album in api.ALBUMS:
|
||||
if album.album == albumtitle and album.artist == artist:
|
||||
return album
|
||||
|
||||
|
||||
def search_albums_by_name(query: str) -> List[models.Album]:
|
||||
"""
|
||||
Searches albums by album name.
|
||||
@@ -106,7 +156,7 @@ def search_albums_by_name(query: str) -> List[models.Album]:
|
||||
artist_albums: List[models.Album] = []
|
||||
|
||||
for album in api.ALBUMS:
|
||||
if query.lower() in album.album.lower():
|
||||
if query.lower() in album.title.lower():
|
||||
title_albums.append(album)
|
||||
|
||||
for album in api.ALBUMS:
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import colorgram
|
||||
from progress.bar import Bar
|
||||
from PIL import Image
|
||||
from io import BytesIO
|
||||
|
||||
from app import api, instances
|
||||
from app.lib.taglib import return_album_art
|
||||
|
||||
|
||||
def get_image_colors(image) -> list:
|
||||
"""Extracts 2 of the most dominant colors from an image."""
|
||||
try:
|
||||
colors = sorted(colorgram.extract(image, 2), key=lambda c: c.hsl.h)
|
||||
except OSError:
|
||||
return []
|
||||
|
||||
formatted_colors = []
|
||||
|
||||
for color in colors:
|
||||
color = f"rgb({color.rgb.r}, {color.rgb.g}, {color.rgb.b})"
|
||||
formatted_colors.append(color)
|
||||
|
||||
return formatted_colors
|
||||
|
||||
|
||||
def save_track_colors(img, filepath) -> None:
|
||||
"""Saves the track colors to the database"""
|
||||
|
||||
track_colors = get_image_colors(img)
|
||||
|
||||
tc_dict = {
|
||||
"filepath": filepath,
|
||||
"colors": track_colors,
|
||||
}
|
||||
|
||||
instances.track_color_instance.insert_track_color(tc_dict)
|
||||
|
||||
|
||||
def save_t_colors():
|
||||
_bar = Bar("Processing image colors", max=len(api.DB_TRACKS))
|
||||
|
||||
for track in api.DB_TRACKS:
|
||||
filepath = track["filepath"]
|
||||
album_art = return_album_art(filepath)
|
||||
|
||||
if album_art is not None:
|
||||
img = Image.open(BytesIO(album_art))
|
||||
save_track_colors(img, filepath)
|
||||
|
||||
_bar.next()
|
||||
|
||||
_bar.finish()
|
||||
@@ -1,5 +1,5 @@
|
||||
import time
|
||||
from typing import List
|
||||
from typing import List, Set
|
||||
|
||||
from app import api
|
||||
from app import helpers
|
||||
@@ -31,17 +31,18 @@ def create_folder(foldername: str) -> models.Folder:
|
||||
return models.Folder(folder)
|
||||
|
||||
|
||||
def create_all_folders() -> List[models.Folder]:
|
||||
folders_: List[models.Folder] = []
|
||||
def create_all_folders() -> Set[models.Folder]:
|
||||
folders: List[models.Folder] = []
|
||||
_bar = Bar("Creating folders", max=len(api.VALID_FOLDERS))
|
||||
|
||||
for foldername in api.VALID_FOLDERS:
|
||||
folder = create_folder(foldername)
|
||||
folders_.append(folder)
|
||||
folders.append(folder)
|
||||
|
||||
_bar.next()
|
||||
_bar.finish()
|
||||
|
||||
return folders_
|
||||
return folders
|
||||
|
||||
|
||||
def get_subdirs(foldername: str) -> List[models.Folder]:
|
||||
@@ -76,5 +77,4 @@ def run_scandir():
|
||||
folders_ = create_all_folders()
|
||||
"""Create all the folder objects before clearing api.FOLDERS"""
|
||||
|
||||
api.FOLDERS.clear()
|
||||
api.FOLDERS.extend(folders_)
|
||||
api.FOLDERS = folders_
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
import os
|
||||
from mutagen.flac import FLAC
|
||||
from mutagen.id3 import ID3
|
||||
from mutagen.flac import MutagenError
|
||||
import mutagen
|
||||
import urllib
|
||||
from PIL import Image
|
||||
from io import BytesIO
|
||||
|
||||
from app import settings
|
||||
|
||||
|
||||
def return_album_art(filepath: str):
|
||||
"""
|
||||
Returns the album art for a given audio file.
|
||||
"""
|
||||
|
||||
if filepath.endswith(".flac"):
|
||||
try:
|
||||
audio = FLAC(filepath)
|
||||
return audio.pictures[0].data
|
||||
except:
|
||||
return None
|
||||
elif filepath.endswith(".mp3"):
|
||||
try:
|
||||
audio = ID3(filepath)
|
||||
return audio.getall("APIC")[0].data
|
||||
except:
|
||||
return None
|
||||
|
||||
|
||||
def extract_thumb(audio_file_path: str, webp_path: str) -> str:
|
||||
"""
|
||||
Extracts the thumbnail from an audio file. Returns the path to the thumbnail.
|
||||
"""
|
||||
img_path = os.path.join(settings.THUMBS_PATH, webp_path)
|
||||
|
||||
if os.path.exists(img_path):
|
||||
return urllib.parse.quote(webp_path)
|
||||
|
||||
album_art = return_album_art(audio_file_path)
|
||||
|
||||
if album_art is not None:
|
||||
img = Image.open(BytesIO(album_art))
|
||||
|
||||
try:
|
||||
small_img = img.resize((250, 250), Image.ANTIALIAS)
|
||||
small_img.save(img_path, format="webp")
|
||||
except OSError:
|
||||
try:
|
||||
png = img.convert("RGB")
|
||||
small_img = png.resize((250, 250), Image.ANTIALIAS)
|
||||
small_img.save(webp_path, format="webp")
|
||||
except:
|
||||
return None
|
||||
|
||||
return urllib.parse.quote(webp_path)
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
def parse_artist_tag(audio):
|
||||
"""
|
||||
Parses the artist tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
artists = audio["artist"][0]
|
||||
except (KeyError, IndexError):
|
||||
artists = "Unknown"
|
||||
|
||||
return artists
|
||||
|
||||
|
||||
def parse_title_tag(audio, full_path: str):
|
||||
"""
|
||||
Parses the title tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
title = audio["title"][0]
|
||||
except (KeyError, IndexError):
|
||||
title = full_path.split("/")[-1]
|
||||
|
||||
return title
|
||||
|
||||
|
||||
def parse_album_artist_tag(audio):
|
||||
"""
|
||||
Parses the album artist tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
albumartist = audio["albumartist"][0]
|
||||
except (KeyError, IndexError):
|
||||
albumartist = "Unknown"
|
||||
|
||||
return albumartist
|
||||
|
||||
|
||||
def parse_album_tag(audio, full_path: str):
|
||||
"""
|
||||
Parses the album tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
album = audio["album"][0]
|
||||
except (KeyError, IndexError):
|
||||
album = full_path.split("/")[-1]
|
||||
|
||||
return album
|
||||
|
||||
|
||||
def parse_genre_tag(audio):
|
||||
"""
|
||||
Parses the genre tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
genre = audio["genre"][0]
|
||||
except (KeyError, IndexError):
|
||||
genre = "Unknown"
|
||||
|
||||
return genre
|
||||
|
||||
|
||||
def parse_date_tag(audio):
|
||||
"""
|
||||
Parses the date tag from an audio file.
|
||||
"""
|
||||
try:
|
||||
date = audio["date"][0]
|
||||
except (KeyError, IndexError):
|
||||
date = "Unknown"
|
||||
|
||||
return date
|
||||
|
||||
|
||||
def parse_track_number(audio):
|
||||
"""
|
||||
Parses the track number from an audio file.
|
||||
"""
|
||||
try:
|
||||
track_number = audio["tracknumber"][0]
|
||||
except (KeyError, IndexError):
|
||||
track_number = "Unknown"
|
||||
|
||||
return track_number
|
||||
|
||||
|
||||
def parse_disk_number(audio):
|
||||
"""
|
||||
Parses the disk number from an audio file.
|
||||
"""
|
||||
try:
|
||||
disk_number = audio["discnumber"][0]
|
||||
except (KeyError, IndexError):
|
||||
disk_number = "Unknown"
|
||||
|
||||
return disk_number
|
||||
|
||||
|
||||
def get_tags(fullpath: str) -> dict:
|
||||
"""
|
||||
Returns a dictionary of tags for a given file.
|
||||
"""
|
||||
try:
|
||||
audio = mutagen.File(fullpath, easy=True)
|
||||
except MutagenError:
|
||||
return None
|
||||
|
||||
tags = {
|
||||
"artists": parse_artist_tag(audio),
|
||||
"title": parse_title_tag(audio, fullpath),
|
||||
"albumartist": parse_album_artist_tag(audio),
|
||||
"album": parse_album_tag(audio, fullpath),
|
||||
"genre": parse_genre_tag(audio),
|
||||
"date": parse_date_tag(audio)[:4],
|
||||
"tracknumber": parse_track_number(audio),
|
||||
"discnumber": parse_disk_number(audio),
|
||||
"length": round(audio.info.length),
|
||||
"bitrate": round(int(audio.info.bitrate) / 1000),
|
||||
"filepath": fullpath,
|
||||
"folder": os.path.dirname(fullpath),
|
||||
}
|
||||
|
||||
return tags
|
||||
@@ -18,18 +18,14 @@ def create_all_tracks() -> List[models.Track]:
|
||||
"""
|
||||
tracks: list[models.Track] = []
|
||||
|
||||
_bar = Bar("Creating tracks", max=len(api.PRE_TRACKS))
|
||||
_bar = Bar("Creating tracks", max=len(api.DB_TRACKS))
|
||||
|
||||
for track in api.PRE_TRACKS:
|
||||
for track in api.DB_TRACKS:
|
||||
try:
|
||||
os.chmod(track["filepath"], 0o755)
|
||||
except FileNotFoundError:
|
||||
instances.songs_instance.remove_song_by_filepath(track["filepath"])
|
||||
api.PRE_TRACKS.remove(track)
|
||||
|
||||
album = albumslib.find_album(track["album"], track["albumartist"])
|
||||
|
||||
track["image"] = album.image
|
||||
instances.tracks_instance.remove_song_by_id(track["_id"]["$oid"])
|
||||
api.DB_TRACKS.remove(track)
|
||||
|
||||
tracks.append(models.Track(track))
|
||||
_bar.next()
|
||||
|
||||
+11
-10
@@ -8,11 +8,12 @@ import os
|
||||
from watchdog.observers import Observer
|
||||
from watchdog.events import PatternMatchingEventHandler
|
||||
|
||||
from app import instances, functions
|
||||
from app import instances
|
||||
from app import models
|
||||
from app.lib import albumslib
|
||||
from app import api
|
||||
from app.lib import folderslib
|
||||
from app.lib.taglib import get_tags
|
||||
from app.lib.albumslib import create_album
|
||||
|
||||
|
||||
class OnMyWatch:
|
||||
@@ -46,14 +47,14 @@ def add_track(filepath: str) -> None:
|
||||
|
||||
Then creates a folder object for the added track and adds it to api.FOLDERS
|
||||
"""
|
||||
tags = functions.get_tags(filepath)
|
||||
tags = get_tags(filepath)
|
||||
|
||||
if tags is not None:
|
||||
instances.songs_instance.insert_song(tags)
|
||||
tags = instances.songs_instance.get_song_by_path(tags["filepath"])
|
||||
instances.tracks_instance.insert_song(tags)
|
||||
tags = instances.tracks_instance.get_song_by_path(tags["filepath"])
|
||||
|
||||
api.PRE_TRACKS.append(tags)
|
||||
album = albumslib.create_album(tags)
|
||||
api.DB_TRACKS.append(tags)
|
||||
album = create_album(tags)
|
||||
api.ALBUMS.append(album)
|
||||
|
||||
tags["image"] = album.image
|
||||
@@ -64,7 +65,7 @@ def add_track(filepath: str) -> None:
|
||||
if folder not in api.VALID_FOLDERS:
|
||||
api.VALID_FOLDERS.add(folder)
|
||||
f = folderslib.create_folder(folder)
|
||||
api.FOLDERS.append(f)
|
||||
api.FOLDERS.add(f)
|
||||
|
||||
|
||||
def remove_track(filepath: str) -> None:
|
||||
@@ -75,12 +76,12 @@ def remove_track(filepath: str) -> None:
|
||||
fpath = filepath.replace(fname, "")
|
||||
|
||||
try:
|
||||
trackid = instances.songs_instance.get_song_by_path(filepath)["_id"]["$oid"]
|
||||
trackid = instances.tracks_instance.get_song_by_path(filepath)["_id"]["$oid"]
|
||||
except TypeError:
|
||||
print(f"💙 Watchdog Error: Error removing track {filepath} TypeError")
|
||||
return
|
||||
|
||||
instances.songs_instance.remove_song_by_id(trackid)
|
||||
instances.tracks_instance.remove_song_by_id(trackid)
|
||||
|
||||
for track in api.TRACKS:
|
||||
if track.trackid == trackid:
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
from app.settings import logger
|
||||
|
||||
|
||||
class Log:
|
||||
def __init__(self, msg):
|
||||
if logger.enable:
|
||||
print("\n🦋 " + msg + "\n")
|
||||
@@ -53,7 +53,7 @@ class Album:
|
||||
Album class
|
||||
"""
|
||||
|
||||
album: str
|
||||
title: str
|
||||
artist: str
|
||||
count: int
|
||||
duration: int
|
||||
@@ -62,20 +62,22 @@ class Album:
|
||||
image: str
|
||||
|
||||
def __init__(self, tags):
|
||||
self.album = tags["album"]
|
||||
self.title = tags["album"]
|
||||
self.artist = tags["artist"]
|
||||
self.count = tags["count"]
|
||||
self.duration = tags["duration"]
|
||||
self.date = tags["date"]
|
||||
self.artistimage = settings.IMG_ARTIST_URI + tags["artistimage"]
|
||||
self.image = settings.IMG_THUMB_URI + tags["image"]
|
||||
self.artistimage = tags["artistimage"]
|
||||
self.image = tags["image"]
|
||||
|
||||
|
||||
def get_p_track(ptrack):
|
||||
for track in api.TRACKS:
|
||||
if (track.title == ptrack["title"]
|
||||
and track.artists == ptrack["artists"]
|
||||
and ptrack["album"] == track.album):
|
||||
if (
|
||||
track.title == ptrack["title"]
|
||||
and track.artists == ptrack["artists"]
|
||||
and ptrack["album"] == track.album
|
||||
):
|
||||
return track
|
||||
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""
|
||||
Contains default configs
|
||||
"""
|
||||
from dataclasses import dataclass
|
||||
import os
|
||||
|
||||
# paths
|
||||
@@ -32,3 +33,7 @@ P_COLORS = [
|
||||
"rgb(141, 132, 2)",
|
||||
"rgb(141, 11, 2)",
|
||||
]
|
||||
|
||||
|
||||
class logger:
|
||||
enable = True
|
||||
+6
-12
@@ -25,11 +25,8 @@
|
||||
</template>
|
||||
|
||||
<script setup>
|
||||
import { ref } from "vue";
|
||||
import Navigation from "./components/LeftSidebar/Navigation.vue";
|
||||
|
||||
import perks from "@/composables/perks.js";
|
||||
|
||||
import Main from "./components/RightSideBar/Main.vue";
|
||||
import nowPlaying from "./components/LeftSidebar/nowPlaying.vue";
|
||||
import NavBar from "./components/nav/NavBar.vue";
|
||||
@@ -40,19 +37,16 @@ import ContextMenu from "./components/contextMenu.vue";
|
||||
import Modal from "./components/modal.vue";
|
||||
import Notification from "./components/Notification.vue";
|
||||
import useQStore from "./stores/queue";
|
||||
import shortcuts from "./composables/keyboard";
|
||||
|
||||
const context_store = useContextStore();
|
||||
const queue = useQStore();
|
||||
|
||||
queue.readQueueFromLocalStorage();
|
||||
import listenForKeyboardEvents from "./composables/keyboard";
|
||||
|
||||
const RightSideBar = Main;
|
||||
|
||||
shortcuts(queue);
|
||||
|
||||
const context_store = useContextStore();
|
||||
const queue = useQStore();
|
||||
const app_dom = document.getElementById("app");
|
||||
|
||||
queue.readQueueFromLocalStorage();
|
||||
listenForKeyboardEvents(queue);
|
||||
|
||||
app_dom.addEventListener("click", (e) => {
|
||||
if (context_store.visible) {
|
||||
context_store.hideContextMenu();
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
<div class="info">
|
||||
<div class="top">
|
||||
<div class="h">Album</div>
|
||||
<div class="title ellip">{{ props.album.album }}</div>
|
||||
<div class="title ellip">{{ props.album.title }}</div>
|
||||
</div>
|
||||
<div class="bottom">
|
||||
<div class="stats">
|
||||
|
||||
+1
-1
@@ -25,7 +25,7 @@ interface Folder {
|
||||
}
|
||||
|
||||
interface AlbumInfo {
|
||||
album: string;
|
||||
title: string;
|
||||
artist: string;
|
||||
count: number;
|
||||
duration: number;
|
||||
|
||||
Reference in New Issue
Block a user