mirror of
https://github.com/clinton-hall/nzbToMedia.git
synced 2024-11-14 17:40:24 -08:00
56c6773c6b
Updates colorama to 0.4.6 Adds confuse version 1.7.0 Updates jellyfish to 0.9.0 Adds mediafile 0.10.1 Updates munkres to 1.1.4 Updates musicbrainzngs to 0.7.1 Updates mutagen to 1.46.0 Updates pyyaml to 6.0 Updates unidecode to 1.3.6
985 lines
33 KiB
Python
985 lines
33 KiB
Python
# This file is part of beets.
|
|
# Copyright 2020, Callum Brown.
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining
|
|
# a copy of this software and associated documentation files (the
|
|
# "Software"), to deal in the Software without restriction, including
|
|
# without limitation the rights to use, copy, modify, merge, publish,
|
|
# distribute, sublicense, and/or sell copies of the Software, and to
|
|
# permit persons to whom the Software is furnished to do so, subject to
|
|
# the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be
|
|
# included in all copies or substantial portions of the Software.
|
|
|
|
"""An AURA server using Flask."""
|
|
|
|
|
|
from mimetypes import guess_type
|
|
import re
|
|
import os.path
|
|
from os.path import isfile, getsize
|
|
|
|
from beets.plugins import BeetsPlugin
|
|
from beets.ui import Subcommand, _open_library
|
|
from beets import config
|
|
from beets.util import py3_path
|
|
from beets.library import Item, Album
|
|
from beets.dbcore.query import (
|
|
MatchQuery,
|
|
NotQuery,
|
|
RegexpQuery,
|
|
AndQuery,
|
|
FixedFieldSort,
|
|
SlowFieldSort,
|
|
MultipleSort,
|
|
)
|
|
|
|
from flask import (
|
|
Blueprint,
|
|
Flask,
|
|
current_app,
|
|
send_file,
|
|
make_response,
|
|
request,
|
|
)
|
|
|
|
|
|
# Constants
|
|
|
|
# AURA server information
|
|
# TODO: Add version information
|
|
SERVER_INFO = {
|
|
"aura-version": "0",
|
|
"server": "beets-aura",
|
|
"server-version": "0.1",
|
|
"auth-required": False,
|
|
"features": ["albums", "artists", "images"],
|
|
}
|
|
|
|
# Maps AURA Track attribute to beets Item attribute
|
|
TRACK_ATTR_MAP = {
|
|
# Required
|
|
"title": "title",
|
|
"artist": "artist",
|
|
# Optional
|
|
"album": "album",
|
|
"track": "track", # Track number on album
|
|
"tracktotal": "tracktotal",
|
|
"disc": "disc",
|
|
"disctotal": "disctotal",
|
|
"year": "year",
|
|
"month": "month",
|
|
"day": "day",
|
|
"bpm": "bpm",
|
|
"genre": "genre",
|
|
"recording-mbid": "mb_trackid", # beets trackid is MB recording
|
|
"track-mbid": "mb_releasetrackid",
|
|
"composer": "composer",
|
|
"albumartist": "albumartist",
|
|
"comments": "comments",
|
|
# Optional for Audio Metadata
|
|
# TODO: Support the mimetype attribute, format != mime type
|
|
# "mimetype": track.format,
|
|
"duration": "length",
|
|
"framerate": "samplerate",
|
|
# I don't think beets has a framecount field
|
|
# "framecount": ???,
|
|
"channels": "channels",
|
|
"bitrate": "bitrate",
|
|
"bitdepth": "bitdepth",
|
|
"size": "filesize",
|
|
}
|
|
|
|
# Maps AURA Album attribute to beets Album attribute
|
|
ALBUM_ATTR_MAP = {
|
|
# Required
|
|
"title": "album",
|
|
"artist": "albumartist",
|
|
# Optional
|
|
"tracktotal": "albumtotal",
|
|
"disctotal": "disctotal",
|
|
"year": "year",
|
|
"month": "month",
|
|
"day": "day",
|
|
"genre": "genre",
|
|
"release-mbid": "mb_albumid",
|
|
"release-group-mbid": "mb_releasegroupid",
|
|
}
|
|
|
|
# Maps AURA Artist attribute to beets Item field
|
|
# Artists are not first-class in beets, so information is extracted from
|
|
# beets Items.
|
|
ARTIST_ATTR_MAP = {
|
|
# Required
|
|
"name": "artist",
|
|
# Optional
|
|
"artist-mbid": "mb_artistid",
|
|
}
|
|
|
|
|
|
class AURADocument:
|
|
"""Base class for building AURA documents."""
|
|
|
|
@staticmethod
|
|
def error(status, title, detail):
|
|
"""Make a response for an error following the JSON:API spec.
|
|
|
|
Args:
|
|
status: An HTTP status code string, e.g. "404 Not Found".
|
|
title: A short, human-readable summary of the problem.
|
|
detail: A human-readable explanation specific to this
|
|
occurrence of the problem.
|
|
"""
|
|
document = {
|
|
"errors": [{"status": status, "title": title, "detail": detail}]
|
|
}
|
|
return make_response(document, status)
|
|
|
|
def translate_filters(self):
|
|
"""Translate filters from request arguments to a beets Query."""
|
|
# The format of each filter key in the request parameter is:
|
|
# filter[<attribute>]. This regex extracts <attribute>.
|
|
pattern = re.compile(r"filter\[(?P<attribute>[a-zA-Z0-9_-]+)\]")
|
|
queries = []
|
|
for key, value in request.args.items():
|
|
match = pattern.match(key)
|
|
if match:
|
|
# Extract attribute name from key
|
|
aura_attr = match.group("attribute")
|
|
# Get the beets version of the attribute name
|
|
beets_attr = self.attribute_map.get(aura_attr, aura_attr)
|
|
converter = self.get_attribute_converter(beets_attr)
|
|
value = converter(value)
|
|
# Add exact match query to list
|
|
# Use a slow query so it works with all fields
|
|
queries.append(MatchQuery(beets_attr, value, fast=False))
|
|
# NOTE: AURA doesn't officially support multiple queries
|
|
return AndQuery(queries)
|
|
|
|
def translate_sorts(self, sort_arg):
|
|
"""Translate an AURA sort parameter into a beets Sort.
|
|
|
|
Args:
|
|
sort_arg: The value of the 'sort' query parameter; a comma
|
|
separated list of fields to sort by, in order.
|
|
E.g. "-year,title".
|
|
"""
|
|
# Change HTTP query parameter to a list
|
|
aura_sorts = sort_arg.strip(",").split(",")
|
|
sorts = []
|
|
for aura_attr in aura_sorts:
|
|
if aura_attr[0] == "-":
|
|
ascending = False
|
|
# Remove leading "-"
|
|
aura_attr = aura_attr[1:]
|
|
else:
|
|
# JSON:API default
|
|
ascending = True
|
|
# Get the beets version of the attribute name
|
|
beets_attr = self.attribute_map.get(aura_attr, aura_attr)
|
|
# Use slow sort so it works with all fields (inc. computed)
|
|
sorts.append(SlowFieldSort(beets_attr, ascending=ascending))
|
|
return MultipleSort(sorts)
|
|
|
|
def paginate(self, collection):
|
|
"""Get a page of the collection and the URL to the next page.
|
|
|
|
Args:
|
|
collection: The raw data from which resource objects can be
|
|
built. Could be an sqlite3.Cursor object (tracks and
|
|
albums) or a list of strings (artists).
|
|
"""
|
|
# Pages start from zero
|
|
page = request.args.get("page", 0, int)
|
|
# Use page limit defined in config by default.
|
|
default_limit = config["aura"]["page_limit"].get(int)
|
|
limit = request.args.get("limit", default_limit, int)
|
|
# start = offset of first item to return
|
|
start = page * limit
|
|
# end = offset of last item + 1
|
|
end = start + limit
|
|
if end > len(collection):
|
|
end = len(collection)
|
|
next_url = None
|
|
else:
|
|
# Not the last page so work out links.next url
|
|
if not request.args:
|
|
# No existing arguments, so current page is 0
|
|
next_url = request.url + "?page=1"
|
|
elif not request.args.get("page", None):
|
|
# No existing page argument, so add one to the end
|
|
next_url = request.url + "&page=1"
|
|
else:
|
|
# Increment page token by 1
|
|
next_url = request.url.replace(
|
|
f"page={page}", "page={}".format(page + 1)
|
|
)
|
|
# Get only the items in the page range
|
|
data = [self.resource_object(collection[i]) for i in range(start, end)]
|
|
return data, next_url
|
|
|
|
def get_included(self, data, include_str):
|
|
"""Build a list of resource objects for inclusion.
|
|
|
|
Args:
|
|
data: An array of dicts in the form of resource objects.
|
|
include_str: A comma separated list of resource types to
|
|
include. E.g. "tracks,images".
|
|
"""
|
|
# Change HTTP query parameter to a list
|
|
to_include = include_str.strip(",").split(",")
|
|
# Build a list of unique type and id combinations
|
|
# For each resource object in the primary data, iterate over it's
|
|
# relationships. If a relationship matches one of the types
|
|
# requested for inclusion (e.g. "albums") then add each type-id pair
|
|
# under the "data" key to unique_identifiers, checking first that
|
|
# it has not already been added. This ensures that no resources are
|
|
# included more than once.
|
|
unique_identifiers = []
|
|
for res_obj in data:
|
|
for rel_name, rel_obj in res_obj["relationships"].items():
|
|
if rel_name in to_include:
|
|
# NOTE: Assumes relationship is to-many
|
|
for identifier in rel_obj["data"]:
|
|
if identifier not in unique_identifiers:
|
|
unique_identifiers.append(identifier)
|
|
# TODO: I think this could be improved
|
|
included = []
|
|
for identifier in unique_identifiers:
|
|
res_type = identifier["type"]
|
|
if res_type == "track":
|
|
track_id = int(identifier["id"])
|
|
track = current_app.config["lib"].get_item(track_id)
|
|
included.append(TrackDocument.resource_object(track))
|
|
elif res_type == "album":
|
|
album_id = int(identifier["id"])
|
|
album = current_app.config["lib"].get_album(album_id)
|
|
included.append(AlbumDocument.resource_object(album))
|
|
elif res_type == "artist":
|
|
artist_id = identifier["id"]
|
|
included.append(ArtistDocument.resource_object(artist_id))
|
|
elif res_type == "image":
|
|
image_id = identifier["id"]
|
|
included.append(ImageDocument.resource_object(image_id))
|
|
else:
|
|
raise ValueError(f"Invalid resource type: {res_type}")
|
|
return included
|
|
|
|
def all_resources(self):
|
|
"""Build document for /tracks, /albums or /artists."""
|
|
query = self.translate_filters()
|
|
sort_arg = request.args.get("sort", None)
|
|
if sort_arg:
|
|
sort = self.translate_sorts(sort_arg)
|
|
# For each sort field add a query which ensures all results
|
|
# have a non-empty, non-zero value for that field.
|
|
for s in sort.sorts:
|
|
query.subqueries.append(
|
|
NotQuery(
|
|
# Match empty fields (^$) or zero fields, (^0$)
|
|
RegexpQuery(s.field, "(^$|^0$)", fast=False)
|
|
)
|
|
)
|
|
else:
|
|
sort = None
|
|
# Get information from the library
|
|
collection = self.get_collection(query=query, sort=sort)
|
|
# Convert info to AURA form and paginate it
|
|
data, next_url = self.paginate(collection)
|
|
document = {"data": data}
|
|
# If there are more pages then provide a way to access them
|
|
if next_url:
|
|
document["links"] = {"next": next_url}
|
|
# Include related resources for each element in "data"
|
|
include_str = request.args.get("include", None)
|
|
if include_str:
|
|
document["included"] = self.get_included(data, include_str)
|
|
return document
|
|
|
|
def single_resource_document(self, resource_object):
|
|
"""Build document for a specific requested resource.
|
|
|
|
Args:
|
|
resource_object: A dictionary in the form of a JSON:API
|
|
resource object.
|
|
"""
|
|
document = {"data": resource_object}
|
|
include_str = request.args.get("include", None)
|
|
if include_str:
|
|
# [document["data"]] is because arg needs to be list
|
|
document["included"] = self.get_included(
|
|
[document["data"]], include_str
|
|
)
|
|
return document
|
|
|
|
|
|
class TrackDocument(AURADocument):
|
|
"""Class for building documents for /tracks endpoints."""
|
|
|
|
attribute_map = TRACK_ATTR_MAP
|
|
|
|
def get_collection(self, query=None, sort=None):
|
|
"""Get Item objects from the library.
|
|
|
|
Args:
|
|
query: A beets Query object or a beets query string.
|
|
sort: A beets Sort object.
|
|
"""
|
|
return current_app.config["lib"].items(query, sort)
|
|
|
|
def get_attribute_converter(self, beets_attr):
|
|
"""Work out what data type an attribute should be for beets.
|
|
|
|
Args:
|
|
beets_attr: The name of the beets attribute, e.g. "title".
|
|
"""
|
|
# filesize is a special field (read from disk not db?)
|
|
if beets_attr == "filesize":
|
|
converter = int
|
|
else:
|
|
try:
|
|
# Look for field in list of Item fields
|
|
# and get python type of database type.
|
|
# See beets.library.Item and beets.dbcore.types
|
|
converter = Item._fields[beets_attr].model_type
|
|
except KeyError:
|
|
# Fall back to string (NOTE: probably not good)
|
|
converter = str
|
|
return converter
|
|
|
|
@staticmethod
|
|
def resource_object(track):
|
|
"""Construct a JSON:API resource object from a beets Item.
|
|
|
|
Args:
|
|
track: A beets Item object.
|
|
"""
|
|
attributes = {}
|
|
# Use aura => beets attribute map, e.g. size => filesize
|
|
for aura_attr, beets_attr in TRACK_ATTR_MAP.items():
|
|
a = getattr(track, beets_attr)
|
|
# Only set attribute if it's not None, 0, "", etc.
|
|
# NOTE: This could result in required attributes not being set
|
|
if a:
|
|
attributes[aura_attr] = a
|
|
|
|
# JSON:API one-to-many relationship to parent album
|
|
relationships = {
|
|
"artists": {"data": [{"type": "artist", "id": track.artist}]}
|
|
}
|
|
# Only add album relationship if not singleton
|
|
if not track.singleton:
|
|
relationships["albums"] = {
|
|
"data": [{"type": "album", "id": str(track.album_id)}]
|
|
}
|
|
|
|
return {
|
|
"type": "track",
|
|
"id": str(track.id),
|
|
"attributes": attributes,
|
|
"relationships": relationships,
|
|
}
|
|
|
|
def single_resource(self, track_id):
|
|
"""Get track from the library and build a document.
|
|
|
|
Args:
|
|
track_id: The beets id of the track (integer).
|
|
"""
|
|
track = current_app.config["lib"].get_item(track_id)
|
|
if not track:
|
|
return self.error(
|
|
"404 Not Found",
|
|
"No track with the requested id.",
|
|
"There is no track with an id of {} in the library.".format(
|
|
track_id
|
|
),
|
|
)
|
|
return self.single_resource_document(self.resource_object(track))
|
|
|
|
|
|
class AlbumDocument(AURADocument):
|
|
"""Class for building documents for /albums endpoints."""
|
|
|
|
attribute_map = ALBUM_ATTR_MAP
|
|
|
|
def get_collection(self, query=None, sort=None):
|
|
"""Get Album objects from the library.
|
|
|
|
Args:
|
|
query: A beets Query object or a beets query string.
|
|
sort: A beets Sort object.
|
|
"""
|
|
return current_app.config["lib"].albums(query, sort)
|
|
|
|
def get_attribute_converter(self, beets_attr):
|
|
"""Work out what data type an attribute should be for beets.
|
|
|
|
Args:
|
|
beets_attr: The name of the beets attribute, e.g. "title".
|
|
"""
|
|
try:
|
|
# Look for field in list of Album fields
|
|
# and get python type of database type.
|
|
# See beets.library.Album and beets.dbcore.types
|
|
converter = Album._fields[beets_attr].model_type
|
|
except KeyError:
|
|
# Fall back to string (NOTE: probably not good)
|
|
converter = str
|
|
return converter
|
|
|
|
@staticmethod
|
|
def resource_object(album):
|
|
"""Construct a JSON:API resource object from a beets Album.
|
|
|
|
Args:
|
|
album: A beets Album object.
|
|
"""
|
|
attributes = {}
|
|
# Use aura => beets attribute name map
|
|
for aura_attr, beets_attr in ALBUM_ATTR_MAP.items():
|
|
a = getattr(album, beets_attr)
|
|
# Only set attribute if it's not None, 0, "", etc.
|
|
# NOTE: This could mean required attributes are not set
|
|
if a:
|
|
attributes[aura_attr] = a
|
|
|
|
# Get beets Item objects for all tracks in the album sorted by
|
|
# track number. Sorting is not required but it's nice.
|
|
query = MatchQuery("album_id", album.id)
|
|
sort = FixedFieldSort("track", ascending=True)
|
|
tracks = current_app.config["lib"].items(query, sort)
|
|
# JSON:API one-to-many relationship to tracks on the album
|
|
relationships = {
|
|
"tracks": {
|
|
"data": [{"type": "track", "id": str(t.id)} for t in tracks]
|
|
}
|
|
}
|
|
# Add images relationship if album has associated images
|
|
if album.artpath:
|
|
path = py3_path(album.artpath)
|
|
filename = path.split("/")[-1]
|
|
image_id = f"album-{album.id}-{filename}"
|
|
relationships["images"] = {
|
|
"data": [{"type": "image", "id": image_id}]
|
|
}
|
|
# Add artist relationship if artist name is same on tracks
|
|
# Tracks are used to define artists so don't albumartist
|
|
# Check for all tracks in case some have featured artists
|
|
if album.albumartist in [t.artist for t in tracks]:
|
|
relationships["artists"] = {
|
|
"data": [{"type": "artist", "id": album.albumartist}]
|
|
}
|
|
|
|
return {
|
|
"type": "album",
|
|
"id": str(album.id),
|
|
"attributes": attributes,
|
|
"relationships": relationships,
|
|
}
|
|
|
|
def single_resource(self, album_id):
|
|
"""Get album from the library and build a document.
|
|
|
|
Args:
|
|
album_id: The beets id of the album (integer).
|
|
"""
|
|
album = current_app.config["lib"].get_album(album_id)
|
|
if not album:
|
|
return self.error(
|
|
"404 Not Found",
|
|
"No album with the requested id.",
|
|
"There is no album with an id of {} in the library.".format(
|
|
album_id
|
|
),
|
|
)
|
|
return self.single_resource_document(self.resource_object(album))
|
|
|
|
|
|
class ArtistDocument(AURADocument):
|
|
"""Class for building documents for /artists endpoints."""
|
|
|
|
attribute_map = ARTIST_ATTR_MAP
|
|
|
|
def get_collection(self, query=None, sort=None):
|
|
"""Get a list of artist names from the library.
|
|
|
|
Args:
|
|
query: A beets Query object or a beets query string.
|
|
sort: A beets Sort object.
|
|
"""
|
|
# Gets only tracks with matching artist information
|
|
tracks = current_app.config["lib"].items(query, sort)
|
|
collection = []
|
|
for track in tracks:
|
|
# Do not add duplicates
|
|
if track.artist not in collection:
|
|
collection.append(track.artist)
|
|
return collection
|
|
|
|
def get_attribute_converter(self, beets_attr):
|
|
"""Work out what data type an attribute should be for beets.
|
|
|
|
Args:
|
|
beets_attr: The name of the beets attribute, e.g. "artist".
|
|
"""
|
|
try:
|
|
# Look for field in list of Item fields
|
|
# and get python type of database type.
|
|
# See beets.library.Item and beets.dbcore.types
|
|
converter = Item._fields[beets_attr].model_type
|
|
except KeyError:
|
|
# Fall back to string (NOTE: probably not good)
|
|
converter = str
|
|
return converter
|
|
|
|
@staticmethod
|
|
def resource_object(artist_id):
|
|
"""Construct a JSON:API resource object for the given artist.
|
|
|
|
Args:
|
|
artist_id: A string which is the artist's name.
|
|
"""
|
|
# Get tracks where artist field exactly matches artist_id
|
|
query = MatchQuery("artist", artist_id)
|
|
tracks = current_app.config["lib"].items(query)
|
|
if not tracks:
|
|
return None
|
|
|
|
# Get artist information from the first track
|
|
# NOTE: It could be that the first track doesn't have a
|
|
# MusicBrainz id but later tracks do, which isn't ideal.
|
|
attributes = {}
|
|
# Use aura => beets attribute map, e.g. artist => name
|
|
for aura_attr, beets_attr in ARTIST_ATTR_MAP.items():
|
|
a = getattr(tracks[0], beets_attr)
|
|
# Only set attribute if it's not None, 0, "", etc.
|
|
# NOTE: This could mean required attributes are not set
|
|
if a:
|
|
attributes[aura_attr] = a
|
|
|
|
relationships = {
|
|
"tracks": {
|
|
"data": [{"type": "track", "id": str(t.id)} for t in tracks]
|
|
}
|
|
}
|
|
album_query = MatchQuery("albumartist", artist_id)
|
|
albums = current_app.config["lib"].albums(query=album_query)
|
|
if len(albums) != 0:
|
|
relationships["albums"] = {
|
|
"data": [{"type": "album", "id": str(a.id)} for a in albums]
|
|
}
|
|
|
|
return {
|
|
"type": "artist",
|
|
"id": artist_id,
|
|
"attributes": attributes,
|
|
"relationships": relationships,
|
|
}
|
|
|
|
def single_resource(self, artist_id):
|
|
"""Get info for the requested artist and build a document.
|
|
|
|
Args:
|
|
artist_id: A string which is the artist's name.
|
|
"""
|
|
artist_resource = self.resource_object(artist_id)
|
|
if not artist_resource:
|
|
return self.error(
|
|
"404 Not Found",
|
|
"No artist with the requested id.",
|
|
"There is no artist with an id of {} in the library.".format(
|
|
artist_id
|
|
),
|
|
)
|
|
return self.single_resource_document(artist_resource)
|
|
|
|
|
|
def safe_filename(fn):
|
|
"""Check whether a string is a simple (non-path) filename.
|
|
|
|
For example, `foo.txt` is safe because it is a "plain" filename. But
|
|
`foo/bar.txt` and `../foo.txt` and `.` are all non-safe because they
|
|
can traverse to other directories other than the current one.
|
|
"""
|
|
# Rule out any directories.
|
|
if os.path.basename(fn) != fn:
|
|
return False
|
|
|
|
# In single names, rule out Unix directory traversal names.
|
|
if fn in ('.', '..'):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
class ImageDocument(AURADocument):
|
|
"""Class for building documents for /images/(id) endpoints."""
|
|
|
|
@staticmethod
|
|
def get_image_path(image_id):
|
|
"""Works out the full path to the image with the given id.
|
|
|
|
Returns None if there is no such image.
|
|
|
|
Args:
|
|
image_id: A string in the form
|
|
"<parent_type>-<parent_id>-<img_filename>".
|
|
"""
|
|
# Split image_id into its constituent parts
|
|
id_split = image_id.split("-")
|
|
if len(id_split) < 3:
|
|
# image_id is not in the required format
|
|
return None
|
|
parent_type = id_split[0]
|
|
parent_id = id_split[1]
|
|
img_filename = "-".join(id_split[2:])
|
|
if not safe_filename(img_filename):
|
|
return None
|
|
|
|
# Get the path to the directory parent's images are in
|
|
if parent_type == "album":
|
|
album = current_app.config["lib"].get_album(int(parent_id))
|
|
if not album or not album.artpath:
|
|
return None
|
|
# Cut the filename off of artpath
|
|
# This is in preparation for supporting images in the same
|
|
# directory that are not tracked by beets.
|
|
artpath = py3_path(album.artpath)
|
|
dir_path = "/".join(artpath.split("/")[:-1])
|
|
else:
|
|
# Images for other resource types are not supported
|
|
return None
|
|
|
|
img_path = os.path.join(dir_path, img_filename)
|
|
# Check the image actually exists
|
|
if isfile(img_path):
|
|
return img_path
|
|
else:
|
|
return None
|
|
|
|
@staticmethod
|
|
def resource_object(image_id):
|
|
"""Construct a JSON:API resource object for the given image.
|
|
|
|
Args:
|
|
image_id: A string in the form
|
|
"<parent_type>-<parent_id>-<img_filename>".
|
|
"""
|
|
# Could be called as a static method, so can't use
|
|
# self.get_image_path()
|
|
image_path = ImageDocument.get_image_path(image_id)
|
|
if not image_path:
|
|
return None
|
|
|
|
attributes = {
|
|
"role": "cover",
|
|
"mimetype": guess_type(image_path)[0],
|
|
"size": getsize(image_path),
|
|
}
|
|
try:
|
|
from PIL import Image
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
im = Image.open(image_path)
|
|
attributes["width"] = im.width
|
|
attributes["height"] = im.height
|
|
|
|
relationships = {}
|
|
# Split id into [parent_type, parent_id, filename]
|
|
id_split = image_id.split("-")
|
|
relationships[id_split[0] + "s"] = {
|
|
"data": [{"type": id_split[0], "id": id_split[1]}]
|
|
}
|
|
|
|
return {
|
|
"id": image_id,
|
|
"type": "image",
|
|
# Remove attributes that are None, 0, "", etc.
|
|
"attributes": {k: v for k, v in attributes.items() if v},
|
|
"relationships": relationships,
|
|
}
|
|
|
|
def single_resource(self, image_id):
|
|
"""Get info for the requested image and build a document.
|
|
|
|
Args:
|
|
image_id: A string in the form
|
|
"<parent_type>-<parent_id>-<img_filename>".
|
|
"""
|
|
image_resource = self.resource_object(image_id)
|
|
if not image_resource:
|
|
return self.error(
|
|
"404 Not Found",
|
|
"No image with the requested id.",
|
|
"There is no image with an id of {} in the library.".format(
|
|
image_id
|
|
),
|
|
)
|
|
return self.single_resource_document(image_resource)
|
|
|
|
|
|
# Initialise flask blueprint
|
|
aura_bp = Blueprint("aura_bp", __name__)
|
|
|
|
|
|
@aura_bp.route("/server")
|
|
def server_info():
|
|
"""Respond with info about the server."""
|
|
return {"data": {"type": "server", "id": "0", "attributes": SERVER_INFO}}
|
|
|
|
|
|
# Track endpoints
|
|
|
|
|
|
@aura_bp.route("/tracks")
|
|
def all_tracks():
|
|
"""Respond with a list of all tracks and related information."""
|
|
doc = TrackDocument()
|
|
return doc.all_resources()
|
|
|
|
|
|
@aura_bp.route("/tracks/<int:track_id>")
|
|
def single_track(track_id):
|
|
"""Respond with info about the specified track.
|
|
|
|
Args:
|
|
track_id: The id of the track provided in the URL (integer).
|
|
"""
|
|
doc = TrackDocument()
|
|
return doc.single_resource(track_id)
|
|
|
|
|
|
@aura_bp.route("/tracks/<int:track_id>/audio")
|
|
def audio_file(track_id):
|
|
"""Supply an audio file for the specified track.
|
|
|
|
Args:
|
|
track_id: The id of the track provided in the URL (integer).
|
|
"""
|
|
track = current_app.config["lib"].get_item(track_id)
|
|
if not track:
|
|
return AURADocument.error(
|
|
"404 Not Found",
|
|
"No track with the requested id.",
|
|
"There is no track with an id of {} in the library.".format(
|
|
track_id
|
|
),
|
|
)
|
|
|
|
path = py3_path(track.path)
|
|
if not isfile(path):
|
|
return AURADocument.error(
|
|
"404 Not Found",
|
|
"No audio file for the requested track.",
|
|
(
|
|
"There is no audio file for track {} at the expected location"
|
|
).format(track_id),
|
|
)
|
|
|
|
file_mimetype = guess_type(path)[0]
|
|
if not file_mimetype:
|
|
return AURADocument.error(
|
|
"500 Internal Server Error",
|
|
"Requested audio file has an unknown mimetype.",
|
|
(
|
|
"The audio file for track {} has an unknown mimetype. "
|
|
"Its file extension is {}."
|
|
).format(track_id, path.split(".")[-1]),
|
|
)
|
|
|
|
# Check that the Accept header contains the file's mimetype
|
|
# Takes into account */* and audio/*
|
|
# Adding support for the bitrate parameter would require some effort so I
|
|
# left it out. This means the client could be sent an error even if the
|
|
# audio doesn't need transcoding.
|
|
if not request.accept_mimetypes.best_match([file_mimetype]):
|
|
return AURADocument.error(
|
|
"406 Not Acceptable",
|
|
"Unsupported MIME type or bitrate parameter in Accept header.",
|
|
(
|
|
"The audio file for track {} is only available as {} and "
|
|
"bitrate parameters are not supported."
|
|
).format(track_id, file_mimetype),
|
|
)
|
|
|
|
return send_file(
|
|
path,
|
|
mimetype=file_mimetype,
|
|
# Handles filename in Content-Disposition header
|
|
as_attachment=True,
|
|
# Tries to upgrade the stream to support range requests
|
|
conditional=True,
|
|
)
|
|
|
|
|
|
# Album endpoints
|
|
|
|
|
|
@aura_bp.route("/albums")
|
|
def all_albums():
|
|
"""Respond with a list of all albums and related information."""
|
|
doc = AlbumDocument()
|
|
return doc.all_resources()
|
|
|
|
|
|
@aura_bp.route("/albums/<int:album_id>")
|
|
def single_album(album_id):
|
|
"""Respond with info about the specified album.
|
|
|
|
Args:
|
|
album_id: The id of the album provided in the URL (integer).
|
|
"""
|
|
doc = AlbumDocument()
|
|
return doc.single_resource(album_id)
|
|
|
|
|
|
# Artist endpoints
|
|
# Artist ids are their names
|
|
|
|
|
|
@aura_bp.route("/artists")
|
|
def all_artists():
|
|
"""Respond with a list of all artists and related information."""
|
|
doc = ArtistDocument()
|
|
return doc.all_resources()
|
|
|
|
|
|
# Using the path converter allows slashes in artist_id
|
|
@aura_bp.route("/artists/<path:artist_id>")
|
|
def single_artist(artist_id):
|
|
"""Respond with info about the specified artist.
|
|
|
|
Args:
|
|
artist_id: The id of the artist provided in the URL. A string
|
|
which is the artist's name.
|
|
"""
|
|
doc = ArtistDocument()
|
|
return doc.single_resource(artist_id)
|
|
|
|
|
|
# Image endpoints
|
|
# Image ids are in the form <parent_type>-<parent_id>-<img_filename>
|
|
# For example: album-13-cover.jpg
|
|
|
|
|
|
@aura_bp.route("/images/<string:image_id>")
|
|
def single_image(image_id):
|
|
"""Respond with info about the specified image.
|
|
|
|
Args:
|
|
image_id: The id of the image provided in the URL. A string in
|
|
the form "<parent_type>-<parent_id>-<img_filename>".
|
|
"""
|
|
doc = ImageDocument()
|
|
return doc.single_resource(image_id)
|
|
|
|
|
|
@aura_bp.route("/images/<string:image_id>/file")
|
|
def image_file(image_id):
|
|
"""Supply an image file for the specified image.
|
|
|
|
Args:
|
|
image_id: The id of the image provided in the URL. A string in
|
|
the form "<parent_type>-<parent_id>-<img_filename>".
|
|
"""
|
|
img_path = ImageDocument.get_image_path(image_id)
|
|
if not img_path:
|
|
return AURADocument.error(
|
|
"404 Not Found",
|
|
"No image with the requested id.",
|
|
"There is no image with an id of {} in the library".format(
|
|
image_id
|
|
),
|
|
)
|
|
return send_file(img_path)
|
|
|
|
|
|
# WSGI app
|
|
|
|
|
|
def create_app():
|
|
"""An application factory for use by a WSGI server."""
|
|
config["aura"].add(
|
|
{
|
|
"host": "127.0.0.1",
|
|
"port": 8337,
|
|
"cors": [],
|
|
"cors_supports_credentials": False,
|
|
"page_limit": 500,
|
|
}
|
|
)
|
|
|
|
app = Flask(__name__)
|
|
# Register AURA blueprint view functions under a URL prefix
|
|
app.register_blueprint(aura_bp, url_prefix="/aura")
|
|
# AURA specifies mimetype MUST be this
|
|
app.config["JSONIFY_MIMETYPE"] = "application/vnd.api+json"
|
|
# Disable auto-sorting of JSON keys
|
|
app.config["JSON_SORT_KEYS"] = False
|
|
# Provide a way to access the beets library
|
|
# The normal method of using the Library and config provided in the
|
|
# command function is not used because create_app() could be called
|
|
# by an external WSGI server.
|
|
# NOTE: this uses a 'private' function from beets.ui.__init__
|
|
app.config["lib"] = _open_library(config)
|
|
|
|
# Enable CORS if required
|
|
cors = config["aura"]["cors"].as_str_seq(list)
|
|
if cors:
|
|
from flask_cors import CORS
|
|
|
|
# "Accept" is the only header clients use
|
|
app.config["CORS_ALLOW_HEADERS"] = "Accept"
|
|
app.config["CORS_RESOURCES"] = {r"/aura/*": {"origins": cors}}
|
|
app.config["CORS_SUPPORTS_CREDENTIALS"] = config["aura"][
|
|
"cors_supports_credentials"
|
|
].get(bool)
|
|
CORS(app)
|
|
|
|
return app
|
|
|
|
|
|
# Beets Plugin Hook
|
|
|
|
|
|
class AURAPlugin(BeetsPlugin):
|
|
"""The BeetsPlugin subclass for the AURA server plugin."""
|
|
|
|
def __init__(self):
|
|
"""Add configuration options for the AURA plugin."""
|
|
super().__init__()
|
|
|
|
def commands(self):
|
|
"""Add subcommand used to run the AURA server."""
|
|
|
|
def run_aura(lib, opts, args):
|
|
"""Run the application using Flask's built in-server.
|
|
|
|
Args:
|
|
lib: A beets Library object (not used).
|
|
opts: Command line options. An optparse.Values object.
|
|
args: The list of arguments to process (not used).
|
|
"""
|
|
app = create_app()
|
|
# Start the built-in server (not intended for production)
|
|
app.run(
|
|
host=self.config["host"].get(str),
|
|
port=self.config["port"].get(int),
|
|
debug=opts.debug,
|
|
threaded=True,
|
|
)
|
|
|
|
run_aura_cmd = Subcommand("aura", help="run an AURA server")
|
|
run_aura_cmd.parser.add_option(
|
|
"-d",
|
|
"--debug",
|
|
action="store_true",
|
|
default=False,
|
|
help="use Flask debug mode",
|
|
)
|
|
run_aura_cmd.func = run_aura
|
|
return [run_aura_cmd]
|