merge refactors pr #364 from @michilyy

* Save to DB only unique trackhashes

* Add check if track already exists in playlist

* replace all paths with `pathlib.Path`

* `architecture.md`:
* add config folder layout

`config.py`:
* fix bug where `pathlib.Path` cannot be serialized

`files.py`:
* remove unused imports
* update path concatenation to `pathlib.Path`
* add config-folder creation

`imgserver.py`:
* fix serialisation bug

`playlistlib.py`:
* update path concatenation to `pathlib.Path`

* update all `settings.Paths` usages to new singleton `Paths` class.

* update all usages of `settings.Paths`

* `files.py`:
* rework assets copy function.
* remove unused loop and unused `shutil.copy2` function

`settings.py`
* fix recursion exception in `Paths`

* `settings.py`:
* remove Singleton and `@property` todos from `Paths`

* `__init__.py`:
* remove now unused function `create_config_dir()`

`setup.files`:
* remove because merged into `settings.Paths()`
  for more central and clear flow how the base path gets decided

`settings.py`:
* add `copy_assets` function

`start_swingmusic.py`:
* add configurable settings.Paths class

`__main__.py`:
* update click to used correct default path

* remove wrong commited egg files

* remove change in the wrong branch

* add forgotten `property` decorator
update `get_files_and_dirs` to use pathlib where possible

`config.py`:
* update type annotation

`folders.py`:
* convert `pathlib` to posix path where needed for sub-functions

`folderlib.py`:
* rework `get_files_and_dirs` to use `pathlib` where possible

`settings.py`:
* add forgotten `@property`

`start_swingmusic.py`:
* remove second `log_startup_info()`

* `artistlib.py`:
* fix calling property
`tagger.py`:
* fix comparing elements in `pathlib.Path`

* add support for repeating lyrics.

* rework lyrics api and lib

* update most path functions.
add type-hint pathlib where needed

* for serialization paths are converted to posix path

* use `open` instead of `os.open`
update `metaclass` with constant

* fix initial config exception if empty file existed

* update `userConfig` with `InitVar` to be excluded from `asdict`

* remove `is_windows_slash()`
rework path function to use pathlib

* convert `pathlib.Path` to `str` for serialization

* fixing bug with str + pathlib

* `__main__.py`:
* update click to use package version
* remove now unused function `print_version`

`filesystem.py`:
* rework `CWD` to use importlib

`pyproject.toml`:
* disable namespace for `importlib.resources` to work correctly

* update `lyrics.py`:
* remove unused functions
* simplify functions

* fix bug where assets get created on root

* remove unused code

* update lyrics for clearer structure.

* add support for unsynced lyrics

* fix wrong return type in unsynced lyrics

* update `/check` to use `send_lyrics`

* prefer tags to duplicates

* `lyrics.py`:
* add docs to a function group

* `logger.py`:
* add logging config dict.
* combine Logging into one file
* add socket logger
* add debug mode to logger
* add JSONL formater

* `logger.py`:
* update config to directly use the formater.
resolves circular import exception

`__main__.py`:
* add logger setup to main

`start_swingmusic.py`:
* add debug option to cli

* `lyrics.py`:
* add offset support

* add `setuptools-scm` to get version from git

* add support for docker build with scm

* add support for docker build with scm
need someone who can test the changes workflow

* update all usage of `version.txt` to `metadata.version()`

* 2x update all usage of `version.txt` to `metadata.version()`

* update to no local_scheme version

* provide fix for #331.
convert `sql.Row` and `TrackTable` to dict before converting to dataclass.

* fix `__main__.py`:
* wrong import and uncommited changes
* add debug and base_path parameter

* fix logger pathlib

* add client build workflow

* set name

* split client from build

* try fixing builds

* try another fix

* try also another fix

* try again something new

* try again something new

* change runner

* fix failed run because of malformed runner

* add wheel builds

* remove systems from pure python build

* add isolated pyinstaller build

* artifacts with names

* wrong wheel path

* try fetch-depth for tag fetch

* disable fail-fast.
add wheel installation

* add install system packages

* add debug

* fix wheel install
fix pyinstaller spec file

* try fix for pyinstaller

* try another fix

* build on release

* add concrete release types

* only run on released or pre-released

* try release upload

* reformat upload

* fix needs tag

* identifiable pyinstaller builds

* compress client folder before uploading

* update to src build

* remove no more needed aarch64 build script
rename pyinstaller assets to lowercase

* remove unneeded code

* fix: save to DB only unique track hashes

* replace click with argparse

* set concrete types in argparse

* replace manuall path usages with pathlib

* remove unused `configs.py` file

* reformat `start_swingmusic.py`

* fix empty set startup exception

* optimizing static files serve function

* fixing bug in optimisation of static files serve function

* fix folder view bug

* colorlib.py:
* fix wrong type exception
* remove singe use Index_everything class
* update logging of populate.py

* cleanup files

* fix settings.py Paths copy function.
Created folder on file.

* add exist check to folder

* remove unused `INFO` class

* fix multiprocessing bug on windows

* potential icon fix for pyinstaller
fix multiple logging bug

* fix argparse config path bug
add jobs file

* cleanup code fragments
fix logging issue
add notes to function

* note that concurrent creates own sys.modules

* refactor some lyrics plugin condition
remove unused import from hashing

* refactor taglib.py

* update import statements to be static

* playlistlib.py:
* refactoring and more doc strings
populate.py:
* add poc bugfix
settings.py:
* add typehint

* possible bugfix for multitreading globals

* folder.py:
* add check if provided path is absolute
populate.py:
* add bug note
settings.py:
* add possible error from Singleton implementation
start_swingmusic.py:
* correct spelling
* pass resolved path to Paths
tagger.py:
* add logging

* trying out fixes for multithreading

* only upload results not metadata

* fix build action again

* folder.py:
* strictly use pathlib where possible

folderlib.py:
* add missing docstring to function, who really need it.

track.py:
* refactor some code

folder.py:
* refactor some more code

* Merge DBPath class and Paths class.
Update all usages of DBPath

folderslib.py:
* fix bug with logging

taglib.py:
* add missing docstring

settings.py:
* merge classes
* refactor

* network.py:
* add more docstring

config.py:
* update pathlib usage

tools.py:
* refactor
* add docstrings

* colorlib.py:
* add docstring

Refactor App builder into grouped config settings.

* update assets access for migration

* Update FUNDING.yml

* Update FUNDING.yml

* upgrade tinytag in requirements.txt

* update readme

* update license

* update readme

* Update README.md

* Update README.md

* cleanup requirements.txt
remove unused import in audio_segment.py
add entrypoint.sh for appimage support
update pyproject.toml for optional dependencies
add appimage to github workflow

* fix invalid workflow file

* AppImage build needs more research.
Commenting for now

* testing a new build workflow

* add libev installation

* update workflow to new optional dependencies

* trying again another fix

* finally fix all optional deps installation correctly

* remove AppImage poc

* albumslib.py:
* add docstring

folder.py:
* add unix path fix

update logger name to `__name__`

* update build with docker
update Dockerfile with git
fix typo in lyrics.py
add dynamic deps back

* add log for static folder

* add missing import

* add some more todos

* add support for AppImages even when it's not perfect.

* quick bugfix for wrong appimage config path

* fix uploading not finding AppImages builds aka wrong pattern

* optimise docker build by using artifacts.
Add client path option.
change docstring to sphinx format

* add todos

* Now support AppImages for real:
manually build AppImage as we are building a complex project.

* fix missing dep in AppImage build

* add full AppImage metadata

* add missing image file.

* only update swingmusic appimage not tool

* add todo and fix AppImage build again.

* Try fixing some path mixup in AppImage build

* add debug tag to action

* correct path to appimage folder

* do not download tool before checkout

* Another fix for path in appimage build

* extend config files with more information

* default client dir is now inside the config dir.
TODOs updated.

* default client dir is now inside the config dir.
TODOs updated.
Add priority todos.

* Auto download client when client not found.
Respects user provided dir.

* rename `requests` submodule to `request`

* poc for arm AppImage builds

* try out another fix

* fix typo in build.yml

* add missing arch tag

* fix uploading double names

* unique naming

* enable fallback version for project.

* do not download client into readonly dir.

* fix relative client download path. Client was resolved into parent of config.

* remove client backup path as client is now downloadable

* `Paths` checks if config folder exists and creates it if necessary.
logger no more creates the config folder.
`app_builder.py`: static route no more with '/client'

* path are only created in MainProcess.
fix gz file not found.

* move assets into src and update usages accordingly

* remove solved todos

* Only upload artefacts if not draft/master aka only on tag

* wrong type in assets copy

* update log with correct priority

* add debug statements and logging to Paths

* remove debugging statement

* remove double version tag from docker build

* fork save release protection

* fix typo

* add fallback client dir for static builds.

* update argparse to new param

* add missing import pathlib

* add sparse checkout as we do not need everything downloaded

* add assets copy check

* init logger bevor Paths

* remove unused import

* check if logdir exists and create if not

* only add exec info to file

* remove exception log from cli

* move logging into main.
Allows tools support again.

* UserConfig now correctly uses _finished key.
Bug where _finished was never written

* double save serverId.
update root_dir to trow no exception on init.
remove debug param

* clean up TODOs

---------

Co-authored-by: skilletfun <skilletfun.laptew.sergey@yandex.ru>
Co-authored-by: Mungai Njoroge <geoffreymungai45@gmail.com>
This commit is contained in:
michily
2025-08-28 09:28:11 +00:00
committed by GitHub
parent b4b0a6e11f
commit e770606567
197 changed files with 2961 additions and 2150 deletions
+35
View File
@@ -0,0 +1,35 @@
"""
This module combines all API blueprints into a single Flask app instance.
"""
from swingmusic.api import (
album,
artist,
collections,
colors,
favorites,
folder,
imgserver,
playlist,
search,
settings,
lyrics,
plugins,
scrobble,
home,
getall,
auth,
stream,
backup_and_restore,
)
from swingmusic.api.plugins import lyrics as lyrics_plugin
from swingmusic.api.plugins import mixes as mixes_plugin
__all__ = [
"album", "artist", "collections", "colors", "favorites", "folder", "imgserver", "playlist", "search", "settings",
"lyrics", "plugins", "scrobble", "home", "getall", "auth", "stream", "backup_and_restore",
"lyrics_plugin",
"mixes_plugin"
]
+222
View File
@@ -0,0 +1,222 @@
"""
Contains all the album routes.
"""
from dataclasses import asdict
from pprint import pprint
import random
from pydantic import BaseModel, Field
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from swingmusic.api.apischemas import AlbumHashSchema, AlbumLimitSchema, ArtistHashSchema
from swingmusic.config import UserConfig
from swingmusic.db.userdata import SimilarArtistTable
from swingmusic.models.album import Album
from swingmusic.settings import Defaults
from swingmusic.store.albums import AlbumStore
from swingmusic.store.artists import ArtistStore
from swingmusic.store.tracks import TrackStore
from swingmusic.utils.hashing import create_hash
from swingmusic.lib.albumslib import sort_by_track_no
from swingmusic.serializers.album import serialize_for_card_many
from swingmusic.serializers.track import serialize_tracks
from swingmusic.utils.stats import get_track_group_stats
bp_tag = Tag(name="Album", description="Single album")
api = APIBlueprint("album", __name__, url_prefix="/album", abp_tags=[bp_tag])
class GetAlbumVersionsBody(BaseModel):
og_album_title: str = Field(
description="The original album title (album.og_title)",
)
albumhash: str = Field(
description="The album hash of the album to exclude from the results.",
)
class GetMoreFromArtistsBody(AlbumLimitSchema):
albumartists: list = Field(
description="The artist hashes to get more albums from",
)
base_title: str = Field(
description="The base title of the album to exclude from the results.",
)
class GetAlbumInfoBody(AlbumHashSchema, AlbumLimitSchema):
pass
# NOTE: Don't use "/" as it will cause redirects (failure)
@api.post("")
def get_album_tracks_and_info(body: GetAlbumInfoBody):
"""
Get album and tracks
Returns album info and tracks for the given albumhash.
"""
albumhash = body.albumhash
albumentry = AlbumStore.albummap.get(albumhash)
if albumentry is None:
return {"error": "Album not found"}, 404
album = albumentry.album
tracks = TrackStore.get_tracks_by_trackhashes(albumentry.trackhashes)
album.trackcount = len(tracks)
album.duration = sum(t.duration for t in tracks)
album.check_type(
tracks=tracks, singleTrackAsSingle=UserConfig().showAlbumsAsSingles
)
track_total = sum({int(t.extra.get("track_total", 1) or 1) for t in tracks})
avg_bitrate = sum(t.bitrate for t in tracks) // (len(tracks) or 1)
more_from_data = GetMoreFromArtistsBody(
albumartists=[a["artisthash"] for a in album.albumartists],
albumlimit=body.limit,
base_title=album.base_title,
)
other_versions_data = GetAlbumVersionsBody(
albumhash=albumhash,
og_album_title=album.og_title,
)
more_from_albums = get_more_from_artist(more_from_data)
other_versions = get_album_versions(other_versions_data)
return {
"stats": get_track_group_stats(tracks, is_album=True),
"info": {
**asdict(album),
"is_favorite": album.is_favorite,
},
"extra": {
# INFO: track_total is the sum of a set of track_total values from each track
# ASSUMPTIONS
# 1. All the tracks have the correct track totals
# 2. Tracks with the same track total are from the same disc
"track_total": track_total,
"avg_bitrate": avg_bitrate,
},
"copyright": tracks[0].copyright,
"tracks": serialize_tracks(tracks, remove_disc=False),
"more_from": more_from_albums,
"other_versions": other_versions,
}
@api.get("/<albumhash>/tracks")
def get_album_tracks(path: AlbumHashSchema):
"""
Get album tracks
Returns all the tracks in the given album, sorted by disc and track number.
NOTE: No album info is returned.
"""
tracks = AlbumStore.get_album_tracks(path.albumhash)
tracks = sort_by_track_no(tracks)
return serialize_tracks(tracks)
@api.post("/from-artist")
def get_more_from_artist(body: GetMoreFromArtistsBody):
"""
Get more from artist
Returns more albums from the given artist hashes.
"""
albumartists = body.albumartists
limit = body.limit
base_title = body.base_title
all_albums: dict[str, list[Album]] = {}
for artisthash in albumartists:
all_albums[artisthash] = AlbumStore.get_albums_by_artisthash(artisthash)
seen_hashes = set()
for artisthash, albums in all_albums.items():
albums = [
a
for a in albums
# INFO: filter out albums added to other artists
if a.albumhash not in seen_hashes and artisthash in a.artisthashes
# INFO: filter out albums with the same base title
and create_hash(a.base_title) != create_hash(base_title)
]
all_albums[artisthash] = serialize_for_card_many(
[a for a in albums if create_hash(a.base_title) != create_hash(base_title)][
:limit
]
)
# INFO: record albums added to other artists
seen_hashes.update([a.albumhash for a in albums][:limit])
return all_albums
@api.post("/other-versions")
def get_album_versions(body: GetAlbumVersionsBody):
"""
Get other versions
Returns other versions of the given album.
"""
albumhash = body.albumhash
album = AlbumStore.albummap.get(albumhash)
if not album:
return []
artisthash = album.album.artisthashes[0]
albums = AlbumStore.get_albums_by_artisthash(artisthash)
basetitle = album.basetitle
albums = [
a
for a in albums
if a.og_title != album.album.og_title
if a.base_title == basetitle
and artisthash in {a["artisthash"] for a in a.albumartists}
]
return serialize_for_card_many(albums)
class GetSimilarAlbumsQuery(ArtistHashSchema, AlbumLimitSchema):
pass
@api.get("/similar")
def get_similar_albums(query: GetSimilarAlbumsQuery):
"""
Get similar albums
Returns similar albums to the given album.
"""
artisthash = query.artisthash
limit = query.limit
similar_artists = SimilarArtistTable.get_by_hash(artisthash)
if similar_artists is None:
return []
artisthashes = similar_artists.get_artist_hash_set()
del similar_artists
artists = ArtistStore.get_artists_by_hashes(artisthashes)
albums = AlbumStore.get_albums_by_artisthashes([a.artisthash for a in artists])
sample = random.sample(albums, min(len(albums), limit))
return serialize_for_card_many(sample[:limit])
+111
View File
@@ -0,0 +1,111 @@
"""
Reusable Pydantic basic schemas for the API
"""
from pydantic import BaseModel, Field
from swingmusic.settings import Defaults
class AlbumHashSchema(BaseModel):
"""
Extending this class will give you a model with the `albumhash` field
"""
albumhash: str = Field(
description="The album hash",
json_schema_extra={
"example": Defaults.API_ALBUMHASH,
},
min_length=Defaults.HASH_LENGTH,
max_length=Defaults.HASH_LENGTH,
)
class ArtistHashSchema(BaseModel):
"""
Extending this class will give you a model with the `artisthash` field
"""
artisthash: str = Field(
description="The artist hash",
json_schema_extra={
"example": Defaults.API_ARTISTHASH,
},
min_length=Defaults.HASH_LENGTH,
max_length=Defaults.HASH_LENGTH,
)
class TrackHashSchema(BaseModel):
"""
Extending this class will give you a model with the `trackhash` field
"""
trackhash: str = Field(
description="The track hash",
json_schema_extra={
"example": Defaults.API_TRACKHASH,
},
min_length=Defaults.HASH_LENGTH,
max_length=Defaults.HASH_LENGTH,
)
class GenericLimitSchema(BaseModel):
"""
Extending this class will give you a model with the `limit` field
"""
limit: int = Field(
description="The number of items to return",
json_schema_extra={
"example": Defaults.API_CARD_LIMIT,
},
default=Defaults.API_CARD_LIMIT,
)
# INFO: The following 3 classes are duplicated to specify the type of items
class TrackLimitSchema(BaseModel):
"""
Extending this class will give you a model with the `limit` field
"""
limit: int = Field(
description="The number of tracks to return",
json_schema_extra={
"example": Defaults.API_CARD_LIMIT,
},
default=5,
alias="tracklimit",
)
class AlbumLimitSchema(BaseModel):
"""
Extending this class will give you a model with the `limit` field
"""
limit: int = Field(
description="The number of albums to return",
json_schema_extra={
"example": Defaults.API_CARD_LIMIT,
},
default=Defaults.API_CARD_LIMIT,
alias="albumlimit",
)
class ArtistLimitSchema(BaseModel):
"""
Extending this class will give you a model with the `limit` field
"""
limit: int = Field(
description="The number of artists to return",
json_schema_extra={
"example": Defaults.API_CARD_LIMIT,
},
default=Defaults.API_CARD_LIMIT,
alias="artistlimit",
)
+226
View File
@@ -0,0 +1,226 @@
"""
Contains all the artist(s) routes.
"""
import math
from pprint import pprint
import random
from datetime import datetime
from itertools import groupby
from typing import Any
from flask_openapi3 import APIBlueprint, Tag
from pydantic import Field
from swingmusic.api.apischemas import (
AlbumLimitSchema,
ArtistHashSchema,
ArtistLimitSchema,
TrackLimitSchema,
)
from swingmusic.config import UserConfig
from swingmusic.db.userdata import SimilarArtistTable
from swingmusic.lib.sortlib import sort_tracks
from swingmusic.serializers.album import serialize_for_card_many
from swingmusic.serializers.artist import serialize_for_cards, serialize_for_card
from swingmusic.serializers.track import serialize_track
from swingmusic.store.albums import AlbumStore
from swingmusic.store.artists import ArtistStore
from swingmusic.store.tracks import TrackStore
from swingmusic.utils.stats import get_track_group_stats
bp_tag = Tag(name="Artist", description="Single artist")
api = APIBlueprint("artist", __name__, url_prefix="/artist", abp_tags=[bp_tag])
class GetArtistAlbumsQuery(AlbumLimitSchema):
all: bool = Field(
description="Whether to ignore albumlimit and return all albums", default=False
)
class GetArtistQuery(TrackLimitSchema, GetArtistAlbumsQuery):
albumlimit: int = Field(7, description="The number of albums to return")
@api.get("/<string:artisthash>")
def get_artist(path: ArtistHashSchema, query: GetArtistQuery):
"""
Get artist
Returns artist data, tracks and genres for the given artisthash.
"""
artisthash = path.artisthash
limit = query.limit
entry = ArtistStore.artistmap.get(artisthash)
if entry is None:
return {"error": "Artist not found"}, 404
tracks = TrackStore.get_tracks_by_trackhashes(entry.trackhashes)
tracks = sort_tracks(tracks, key="playcount", reverse=True)
tcount = len(tracks)
artist = entry.artist
if artist.albumcount == 0 and tcount < 10:
limit = tcount
try:
year = datetime.fromtimestamp(artist.date).year
except ValueError:
year = 0
genres = [*artist.genres]
decade = None
if year:
decade = math.floor(year / 10) * 10
decade = str(decade)[2:] + "s"
if decade:
genres.insert(0, {"name": decade, "genrehash": decade})
stats = get_track_group_stats(tracks)
duration = sum(t.duration for t in tracks) if tracks else 0
tracks = tracks[:limit] if (limit and limit != -1) else tracks
tracks = [
{
**serialize_track(t),
"help_text": (
"unplayed"
if t.playcount == 0
else f"{t.playcount} play{'' if t.playcount == 1 else 's'}"
),
}
for t in tracks
]
query.limit = query.albumlimit
albums = get_artist_albums(path, query)
return {
"artist": {
**serialize_for_card(artist),
"duration": duration,
"trackcount": tcount,
"albumcount": artist.albumcount,
"genres": genres,
"is_favorite": artist.is_favorite,
},
"tracks": tracks,
"albums": albums,
"stats": stats,
}
@api.get("/<artisthash>/albums")
def get_artist_albums(path: ArtistHashSchema, query: GetArtistAlbumsQuery):
"""
Get artist albums.
"""
return_all = query.all
artisthash = path.artisthash
limit = query.limit
entry = ArtistStore.artistmap.get(artisthash)
if entry is None:
return {"error": "Artist not found"}, 404
albums = AlbumStore.get_albums_by_hashes(entry.albumhashes)
tracks = TrackStore.get_tracks_by_trackhashes(entry.trackhashes)
missing_albumhashes = {
t.albumhash for t in tracks if t.albumhash not in {a.albumhash for a in albums}
}
albums.extend(AlbumStore.get_albums_by_hashes(missing_albumhashes))
albumdict = {a.albumhash: a for a in albums}
config = UserConfig()
albumgroups = groupby(tracks, key=lambda t: t.albumhash)
for albumhash, tracks in albumgroups:
album = albumdict.get(albumhash)
if album:
album.check_type(list(tracks), config.showAlbumsAsSingles)
albums = [a for a in albumdict.values()]
all_albums = sorted(albums, key=lambda a: a.date, reverse=True)
res: dict[str, Any] = {
"albums": [],
"appearances": [],
"compilations": [],
"singles_and_eps": [],
}
for album in all_albums:
if album.type == "single" or album.type == "ep":
res["singles_and_eps"].append(album)
elif album.type == "compilation":
res["compilations"].append(album)
elif (
album.albumhash in missing_albumhashes
or artisthash not in album.artisthashes
):
res["appearances"].append(album)
else:
res["albums"].append(album)
if return_all:
limit = len(all_albums)
# loop through the res dict and serialize the albums
for key, value in res.items():
res[key] = serialize_for_card_many(value[:limit])
res["artistname"] = entry.artist.name
return res
@api.get("/<artisthash>/tracks")
def get_all_artist_tracks(path: ArtistHashSchema):
"""
Get artist tracks
Returns all artists by a given artist.
"""
tracks = ArtistStore.get_artist_tracks(path.artisthash)
tracks = sort_tracks(tracks, key="playcount", reverse=True)
tracks = [
{
**serialize_track(t),
"help_text": (
"unplayed"
if t.playcount == 0
else f"{t.playcount} play{'' if t.playcount == 1 else 's'}"
),
}
for t in tracks
]
return tracks
@api.get("/<artisthash>/similar")
def get_similar_artists(path: ArtistHashSchema, query: ArtistLimitSchema):
"""
Get similar artists.
"""
limit = query.limit
result = SimilarArtistTable.get_by_hash(path.artisthash)
if result is None:
return []
similar = ArtistStore.get_artists_by_hashes(result.get_artist_hash_set())
if len(similar) > limit:
similar = random.sample(similar, min(limit, len(similar)))
return serialize_for_cards(similar[:limit])
+379
View File
@@ -0,0 +1,379 @@
import json
from functools import wraps
import sqlite3
from flask import current_app, jsonify
from flask_jwt_extended import (
create_access_token,
create_refresh_token,
current_user,
get_jwt_identity,
jwt_required,
set_access_cookies,
)
from pydantic import BaseModel, Field
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from swingmusic.db.userdata import UserTable
from swingmusic.store.homepage import HomepageStore
from swingmusic.utils.auth import check_password, hash_password
from swingmusic.config import UserConfig
bp_tag = Tag(name="Auth", description="Authentication stuff")
api = APIBlueprint("auth", __name__, url_prefix="/auth", abp_tags=[bp_tag])
def admin_required():
"""
Decorator to require admin role
"""
def wrapper(fn):
@wraps(fn)
def decorator(*args, **kwargs):
if "admin" not in current_user["roles"]:
return {"msg": "Only admins can do that!"}, 403
return fn(*args, **kwargs)
return decorator
return wrapper
def create_new_token(user: dict):
"""
Create a new token response
"""
access_token = create_access_token(identity=user)
max_age: int = current_app.config.get("JWT_ACCESS_TOKEN_EXPIRES")
return {
"msg": f"Logged in as {user['username']}",
"accesstoken": access_token,
"refreshtoken": create_refresh_token(identity=user),
"maxage": max_age,
}
class LoginBody(BaseModel):
username: str = Field(description="The username", example="user0")
password: str = Field(description="The password", example="password0")
@api.post("/login")
def login(body: LoginBody):
"""
Authenticate using username and password
"""
user = UserTable.get_by_username(body.username)
if user is None:
return {"msg": "User not found"}, 404
password_ok = check_password(body.password, user.password)
if not password_ok:
return {"msg": "Hehe! invalid password"}, 401
res = create_new_token(user.todict())
token = res["accesstoken"]
age = res["maxage"]
res = jsonify(res)
set_access_cookies(res, token, max_age=age)
return res
pair_token = dict()
@api.get("/getpaircode")
def get_pair():
"""
Get a new pair code to log in to thee Swing Music mobile app
"""
# INFO: if user is already logged in, create a new pair code
token = create_new_token(get_jwt_identity())
key = token["accesstoken"][-6:]
global pair_token
pair_token = {
key: token,
}
return {"code": key}
class PairDeviceQuery(BaseModel):
code: str = Field("", description="The code")
@api.get("/pair")
@jwt_required(optional=True)
def pair_with_code(query: PairDeviceQuery):
"""
Get an access token by sending a pair code. NOTE: A code can only be used once!
"""
global pair_token
token = pair_token.get(query.code, None)
if token:
pair_token = {}
return token
return {"msg": "Invalid code"}, 400
@api.post("/refresh")
@jwt_required(refresh=True)
def refresh():
"""
Refresh an access token by sending a refresh token in the Authorization header
>>> Headers:
>>> Authorization: Bearer <refresh_token>
Won't work with cookies!!!
"""
user = get_jwt_identity()
return create_new_token(user)
class UpdateProfileBody(BaseModel):
id: int = Field(0, description="The user id")
email: str = Field("", description="The email")
username: str = Field("", description="The username", example="user0")
password: str = Field("", description="The password", example="password0")
roles: list[str] = Field(None, description="The roles")
@api.put("/profile/update")
def update_profile(body: UpdateProfileBody):
"""
Update user profile
"""
user = {
"id": body.id,
"username": body.username,
"password": body.password,
"roles": body.roles,
}
# prevent updating guest
if current_user["username"] == "guest" or user["username"] == "guest":
return {"msg": "Cannot update guest user"}, 400
# if not id, update self
if not user["id"]:
user["id"] = current_user["id"]
if body.roles is not None:
# only admins can update roles
if "admin" not in current_user["roles"]:
return {"msg": "Only admins can update roles"}, 403
all_users = UserTable.get_all()
if "admin" not in body.roles:
# check if we're removing the last admin
admins = [user for user in all_users if "admin" in user.roles]
if len(admins) == 1 and admins[0].id == user["id"]:
return {"msg": "Cannot remove the only admin"}, 400
# guest roles cannot be updated
_user = [u for u in all_users if u.id == user["id"]][0]
if "guest" in _user.roles:
return {"msg": "Cannot update guest user"}, 400
# finally, convert roles to json string
user["roles"] = body.roles
if user["password"]:
user["password"] = hash_password(user["password"])
# remove empty values
clean_user = {k: v for k, v in user.items() if v}
try:
# return authdb.update_user(clean_user)
UserTable.update_one(clean_user)
return UserTable.get_by_id(user["id"]).todict()
except sqlite3.IntegrityError:
return {"msg": "Username already exists"}, 400
@api.post("/profile/create")
@admin_required()
def create_user(body: UpdateProfileBody):
"""
Create a new user
"""
if not body.username or not body.password:
return {"msg": "Username and password are required"}, 400
user = {
"username": body.username,
"password": hash_password(body.password),
"roles": [],
}
# check if user already exists
if UserTable.get_by_username(user["username"]):
return {"msg": "Username already exists"}, 400
UserTable.insert_one(user)
user = UserTable.get_by_username(user["username"])
if user:
HomepageStore.entries["recently_played"].add_new_user(user.id)
return user.todict()
return {
"msg": "Failed to create user",
}, 500
@api.post("/profile/guest/create")
@admin_required()
def create_guest_user():
"""
Create a guest user
"""
# check if guest user already exists
guest_user = UserTable.get_by_username("guest")
if guest_user:
return {
"msg": "Guest user already exists",
}, 400
UserTable.insert_guest_user()
user = UserTable.get_by_username("guest")
if user:
HomepageStore.entries["recently_played"].add_new_user(user.id)
return {
"msg": "Guest user created",
}
return {
"msg": "Failed to create guest user",
}, 500
class DeleteUseBody(BaseModel):
username: str = Field("", description="The username")
@api.delete("/profile/delete")
@admin_required()
def delete_user(body: DeleteUseBody):
"""
Delete a user by username
"""
# prevent admin from deleting themselves
if body.username == current_user["username"]:
return {"msg": "Sorry! you cannot delete yourselfu"}, 400
# prevent deleting the only admin
users = UserTable.get_all()
admins = [user for user in users if "admin" in user.roles]
if len(admins) == 1 and admins[0].username == body.username:
return {"msg": "Cannot delete the only admin"}, 400
UserTable.remove_by_username(body.username)
return {"msg": f"User {body.username} deleted"}
@api.get("/logout")
def logout():
"""
Log out and clear the access token cookie
"""
res = jsonify({"msg": "Logged out"})
res.delete_cookie("access_token_cookie")
return res
class GetAllUsersQuery(BaseModel):
simplified: bool = Field(
False, description="Whether to return simplified user data"
)
@api.get("/users")
@jwt_required(optional=True)
def get_all_users(query: GetAllUsersQuery):
"""
Get all users (if you're an admin, you will also receive accounts settings)
"""
config = UserConfig()
settings = {
"enableGuest": False,
"usersOnLogin": config.usersOnLogin,
}
res = {
"settings": {},
"users": [],
}
users = [u for u in UserTable.get_all()]
is_admin = current_user and "admin" in current_user["roles"]
settings["enableGuest"] = [
user for user in users if user.username == "guest"
].__len__() > 0
# if user is admin, also return settings
if is_admin:
res = {
"settings": settings,
}
# if is normal user, return empty response
elif current_user:
return res
# if not logged in and showing users on login is disabled, return empty response
elif (
not current_user
and not settings["usersOnLogin"]
and not settings["enableGuest"]
):
return res
# remove guest user
# if not settings["enableGuest"]:
# users = [user for user in users if user.username != "guest"]
if not settings["usersOnLogin"]:
users = [user for user in users if user.username == "guest"]
# reverse list to show latest users first
users = reversed(users)
# bring admins to the front
users = sorted(users, key=lambda x: "admin" in x.roles, reverse=True)
# bring current user to index 0
if current_user:
users = sorted(
users,
key=lambda x: x.username == current_user["username"],
reverse=True,
)
if query.simplified:
res["users"] = [user.todict_simplified() for user in users]
else:
res["users"] = [user.todict() for user in users]
return res
@api.get("/user")
def get_logged_in_user():
"""
Get logged in user
"""
return dict(current_user)
+314
View File
@@ -0,0 +1,314 @@
from dataclasses import asdict
import json
import os
from pathlib import Path
from pprint import pprint
import shutil
from time import time
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
import sqlalchemy.exc
from swingmusic.api.auth import admin_required
from swingmusic.db.userdata import FavoritesTable, PlaylistTable, ScrobbleTable, CollectionTable
from swingmusic.lib.index import index_everything
from swingmusic.settings import Paths
from datetime import datetime
from swingmusic.utils.dates import timestamp_to_time_passed
from pydantic import BaseModel, Field
from typing import Optional
bp_tag = Tag(name="Backup and Restore", description="Backup and Restore")
api = APIBlueprint(
"backup_and_restore", __name__, url_prefix="/backup", abp_tags=[bp_tag]
)
@api.post("/create")
@admin_required()
def backup():
"""
Create a backup file of your favorites, playlists, scrobble data, and collections.
"""
backup_name = f"backup.{int(time())}"
backup_dir = Path("~").expanduser() / "swingmusic.backup" / backup_name
backup_dir.mkdir(parents=True, exist_ok=True)
backup_file = backup_dir / "data.json"
img_folder = backup_dir / "images"
img_folder_created = img_folder.exists()
favorites = FavoritesTable.get_all()
favorites = [asdict(entry) for entry in favorites]
scrobbles = ScrobbleTable.get_all(start=0)
scrobbles = [asdict(entry) for entry in scrobbles]
for scrobble in scrobbles:
del scrobble["id"]
# SECTION: Playlists
playlists = PlaylistTable.get_all()
playlist_dicts = []
for entry in playlists:
playlist = asdict(entry)
for key in [
"id",
"_last_updated",
"has_image",
"images",
"duration",
"count",
"pinned",
"thumb",
]:
del playlist[key]
playlist_dicts.append(playlist)
# copy images
img_path = Path(Paths().playlist_img_path) / str(playlist["image"])
if img_path.exists():
if not img_folder_created:
img_folder.mkdir(parents=True)
img_folder_created = True
shutil.copy(img_path, img_folder / playlist["image"])
# !SECTION
# SECTION: Collections
collections_list = list(CollectionTable.get_all())
collections_dicts = []
for collection in collections_list:
# Remove auto-generated id field
collection_copy = collection.copy()
if "id" in collection_copy:
del collection_copy["id"]
collections_dicts.append(collection_copy)
# !SECTION
data = {
"favorites": favorites,
"scrobbles": scrobbles,
"playlists": playlist_dicts,
"collections": collections_dicts,
}
with open(backup_file, "w") as f:
json.dump(data, f, indent=4)
return {
"name": backup_name,
"date": timestamp_to_time_passed(int(backup_name.split(".")[1])),
"scrobbles": len(scrobbles),
"favorites": len(favorites),
"playlists": len(playlist_dicts),
"collections": len(collections_dicts),
}, 200
class RestoreBackup:
# TODO: BACKUP AND RESTORE MIXES!
# TODO: IMPROVE UX WHEN WAITING FOR RESTORE TO COMPLETE!
def __init__(self, backup_dir: Path):
self.backup_dir = backup_dir
self.backup_file = backup_dir / "data.json"
with open(self.backup_file, "r") as f:
self.data = json.load(f)
self.restore_favorites(self.data["favorites"])
self.restore_playlists(self.data["playlists"])
self.restore_scrobbles(self.data["scrobbles"])
self.restore_collections(self.data.get("collections", []))
def restore(self):
pass
def restore_favorites(self, favorites: list[dict]):
existing_favorites = FavoritesTable.get_all()
existing_hashes = set(fav.hash for fav in existing_favorites)
new_favorites = [fav for fav in favorites if fav["hash"] not in existing_hashes]
for fav in new_favorites:
try:
FavoritesTable.insert_item(fav)
except sqlalchemy.exc.IntegrityError:
print("Integrity error, skipping favorite")
print(fav)
def restore_playlists(self, playlists: list[dict]):
existing_playlists = PlaylistTable.get_all()
existing_names = set(playlist.name for playlist in existing_playlists)
new_playlists = [
playlist for playlist in playlists if playlist["name"] not in existing_names
]
for playlist in new_playlists:
try:
if playlist.get("_score") is not None:
del playlist["_score"]
PlaylistTable.add_one(playlist)
except sqlalchemy.exc.IntegrityError:
print("Integrity error, skipping playlist:")
print(playlist)
def restore_scrobbles(self, scrobbles: list[dict]):
existing_scrobbles = ScrobbleTable.get_all(0)
existing_hashes = set(
f"{scrobble.trackhash}.{scrobble.timestamp}"
for scrobble in existing_scrobbles
)
new_scrobbles = [
scrobble
for scrobble in scrobbles
if f"{scrobble['trackhash']}.{scrobble['timestamp']}" not in existing_hashes
]
for scrobble in new_scrobbles:
try:
ScrobbleTable.add(scrobble)
except sqlalchemy.exc.IntegrityError:
print("Integrity error, skipping scrobble:")
print(scrobble)
def restore_collections(self, collections: list[dict]):
existing_collections = list(CollectionTable.get_all())
existing_names = set(collection["name"] for collection in existing_collections)
new_collections = [
collection for collection in collections if collection["name"] not in existing_names
]
for collection in new_collections:
try:
# Ensure userid is set for the collection
if collection.get("userid") is None:
from swingmusic.utils.auth import get_current_userid
collection["userid"] = get_current_userid()
CollectionTable.insert_one(collection)
except sqlalchemy.exc.IntegrityError:
print("Integrity error, skipping collection:")
print(collection)
class RestoreBackupBody(BaseModel):
backup_dir: Optional[str] = Field(
default=None,
description="The name of the backup directory to restore from. If not provided, all backups will be restored.",
example="backup.1234567890",
)
@api.post("/restore")
@admin_required()
def restore(body: RestoreBackupBody):
"""
Restore your favorites, playlists, scrobble data, and collections from a specified backup or all backups.
"""
backup_base_dir = Path("~").expanduser() / "swingmusic.backup"
backups = []
if body.backup_dir:
# Restore from a specific backup
specified_backup_dir = backup_base_dir / body.backup_dir
if not specified_backup_dir.exists() or not specified_backup_dir.is_dir():
return {"msg": f"Backup '{body.backup_dir}' not found"}, 404
restore_backup = RestoreBackup(specified_backup_dir)
restore_backup.restore()
backups.append(body.backup_dir)
else:
# Restore from all backups
try:
backup_dirs = [d for d in backup_base_dir.iterdir() if d.is_dir()]
except FileNotFoundError:
backup_dirs = []
if not backup_dirs:
return {"msg": "No backups found"}, 404
for backup_dir in sorted(backup_dirs, key=lambda x: x.name, reverse=True):
restore_backup = RestoreBackup(backup_dir)
restore_backup.restore()
backups.append(backup_dir.name)
index_everything()
return {"msg": f"Restored successfully", "backups": backups}, 200
@api.get("/list")
@admin_required()
def list_backups():
"""
List all backups with detailed information.
"""
backup_dir = Path("~").expanduser() / "swingmusic.backup"
backups = []
entries = []
try:
paths = [p for p in backup_dir.iterdir() if p.is_dir()]
except FileNotFoundError:
paths = []
for path in paths:
try:
entries.append(
{"path": path, "timestamp": int(path.name.split(".")[1])}
)
except (IndexError, ValueError):
pass
entries = sorted(entries, key=lambda x: x["timestamp"], reverse=True)
for entry in entries:
backup_info = {
"name": entry["path"].name,
"date": timestamp_to_time_passed(entry["timestamp"]),
}
# Read the JSON file and count items
json_file: Path = entry["path"] / "data.json"
if json_file.exists():
with json_file.open("r") as f:
data = json.load(f)
backup_info["scrobbles"] = len(data.get("scrobbles", []))
backup_info["favorites"] = len(data.get("favorites", []))
backup_info["playlists"] = len(data.get("playlists", []))
backup_info["collections"] = len(data.get("collections", []))
else:
backup_info["scrobbles"] = 0
backup_info["favorites"] = 0
backup_info["playlists"] = 0
backup_info["collections"] = 0
backups.append(backup_info)
return {"backups": backups}, 200
class DeleteBackupBody(BaseModel):
backup_dir: str = Field(
..., description="The name of the backup directory to delete."
)
@api.delete("/delete")
@admin_required()
def delete_backup(body: DeleteBackupBody):
"""
Delete a backup.
"""
backup_dir = Path("~").expanduser() / "swingmusic.backup"
backup_dir = backup_dir / body.backup_dir
if not backup_dir.exists() or not backup_dir.is_dir():
return {"msg": f"Backup '{body.backup_dir}' not found"}, 404
shutil.rmtree(backup_dir)
return {"msg": f"Backup '{body.backup_dir}' deleted"}, 200
+182
View File
@@ -0,0 +1,182 @@
"""
Contains all the collection routes.
"""
from typing import Any
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import BaseModel, Field
from swingmusic.db.userdata import CollectionTable
from swingmusic.lib.pagelib import recover_page_items, remove_page_items, validate_page_items
from swingmusic.utils.auth import get_current_userid
bp_tag = Tag(name="Collections", description="Collections")
api = APIBlueprint(
"collections", __name__, url_prefix="/collections", abp_tags=[bp_tag]
)
class CreateCollectionBody(BaseModel):
name: str = Field(description="The name of the collection")
description: str = Field(description="The description of the collection")
items: list[dict[str, Any]] = Field(
description="The items to add to the collection",
json_schema_extra={"example": [{"type": "album", "hash": "1234567890"}]},
)
@api.post("")
def create_collection(body: CreateCollectionBody):
"""
Create a new collection.
"""
items = validate_page_items(body.items, existing=[])
if len(items) == 0:
return {"error": "No items to add"}, 400
payload = {
"name": body.name,
"items": items,
"userid": get_current_userid(),
"extra": {
"description": body.description,
},
}
CollectionTable.insert_one(payload)
return {"message": "collection created"}, 201
@api.get("")
def get_collections():
"""
Get all collections.
"""
return [collection for collection in CollectionTable.get_all()]
class AddCollectionItemBody(BaseModel):
item: dict[str, Any] = Field(
description="The item to add to the collection",
json_schema_extra={"example": {"type": "album", "hash": "1234567890"}},
)
class AddCollectionItemPath(BaseModel):
collection_id: int = Field(
description="The ID of the collection to add items to",
json_schema_extra={"example": 1},
)
@api.post("/<int:collection_id>/items")
def add_collection_item(path: AddCollectionItemPath, body: AddCollectionItemBody):
"""
Add an item to a collection.
"""
collection = CollectionTable.get_by_id(path.collection_id)
if collection is None:
return {"error": "Collection not found"}, 404
new_items = validate_page_items([body.item], existing=collection["items"])
if len(new_items) == 0:
return {"error": "items already in collection"}, 400
collection["items"].extend(new_items)
CollectionTable.update_items(collection["id"], collection["items"])
return {"message": "Items added to collection"}
class RemoveCollectionItemBody(BaseModel):
item: dict[str, Any] = Field(
description="The item to remove from the collection",
json_schema_extra={"example": {"type": "album", "hash": "1234567890"}},
)
class RemoveCollectionItemPath(BaseModel):
collection_id: int = Field(
description="The ID of the collection to remove items from"
)
@api.delete("/<int:collection_id>/items")
def remove_collection_item(
path: RemoveCollectionItemPath, body: RemoveCollectionItemBody
):
"""
Remove an item from a collection.
"""
collection = CollectionTable.get_by_id(path.collection_id)
if collection is None:
return {"error": "Collection not found"}, 404
remaining = remove_page_items(collection["items"], body.item)
CollectionTable.update_items(collection["id"], remaining)
return {"message": "Item removed from collection"}
class GetCollectionBody(BaseModel):
collection_id: int = Field(description="The ID of the collection to get")
@api.get("/<int:collection_id>")
def get_collection(path: GetCollectionBody):
"""
Get a collection.
"""
collection = CollectionTable.get_by_id(path.collection_id)
if not collection:
return {"error": "Collection not found"}, 404
items = recover_page_items(collection["items"])
return {
"id": collection["id"],
"name": collection["name"],
"items": items,
"extra": collection["extra"],
}
class UpdateCollectionBody(BaseModel):
name: str = Field(description="The name of the collection")
description: str = Field(
description="The description of the collection", default=""
)
@api.put("/<int:collection_id>")
def update_collection(path: GetCollectionBody, body: UpdateCollectionBody):
"""
Update a collection.
"""
payload = {
"id": path.collection_id,
"name": body.name,
"extra": {"description": body.description},
}
CollectionTable.update_one(payload)
return payload
class DeleteCollectionPath(BaseModel):
collection_id: int = Field(description="The ID of the collection to delete")
@api.delete("/<int:collection_id>")
def delete_collection(path: DeleteCollectionPath):
"""
Delete a collection.
"""
CollectionTable.delete_by_id(path.collection_id)
return {"message": "Collection deleted"}
+22
View File
@@ -0,0 +1,22 @@
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from swingmusic.api.apischemas import AlbumHashSchema
from swingmusic.store.albums import AlbumStore as Store
bp_tag = Tag(name="Colors", description="Get item colors")
api = APIBlueprint("colors", __name__, url_prefix="/colors", abp_tags=[bp_tag])
@api.get("/album/<albumhash>")
def get_album_color(path: AlbumHashSchema):
"""
Get album color
"""
album = Store.get_album_by_hash(path.albumhash)
msg = {"color": ""}
if album is None or len(album.colors) == 0:
return msg, 404
return {"color": album.colors[0]}
+297
View File
@@ -0,0 +1,297 @@
from typing import List, TypeVar
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import BaseModel, Field
from swingmusic.api.apischemas import GenericLimitSchema
from swingmusic.db.userdata import FavoritesTable
from swingmusic.lib.extras import get_extra_info
from swingmusic.models import FavType
from swingmusic.settings import Defaults
from swingmusic.store.albums import AlbumStore
from swingmusic.store.artists import ArtistStore
from swingmusic.store.tracks import TrackStore
from swingmusic.serializers.track import serialize_track, serialize_tracks
from swingmusic.serializers.artist import (
serialize_for_card as serialize_artist,
serialize_for_cards,
)
from swingmusic.utils.dates import timestamp_to_time_passed
from swingmusic.serializers.album import serialize_for_card, serialize_for_card_many
bp_tag = Tag(name="Favorites", description="Your favorite items")
api = APIBlueprint("favorites", __name__, url_prefix="/favorites", abp_tags=[bp_tag])
T = TypeVar("T")
def remove_none(items: List[T]) -> List[T]:
return [i for i in items if i is not None]
class FavoritesAddBody(BaseModel):
hash: str = Field(
description="The hash of the item",
min_length=Defaults.HASH_LENGTH,
max_length=Defaults.HASH_LENGTH,
)
type: str = Field(description="The type of the item")
def toggle_fav(type: str, hash: str):
"""
Toggles a favorite item.
"""
if type == FavType.track:
entry = TrackStore.trackhashmap.get(hash)
if entry is not None:
entry.toggle_favorite_user()
elif type == FavType.album:
entry = AlbumStore.albummap.get(hash)
if entry is not None:
entry.toggle_favorite_user()
elif type == FavType.artist:
entry = ArtistStore.artistmap.get(hash)
if entry is not None:
entry.toggle_favorite_user()
return {"msg": "Added to favorites"}
@api.post("/add")
def toggle_favorite(body: FavoritesAddBody):
"""
Adds a favorite to the database.
"""
extra = get_extra_info(body.hash, body.type)
try:
FavoritesTable.insert_item(
{"hash": body.hash, "type": body.type, "extra": extra}
)
except Exception as e:
print(e)
return {"msg": "Failed! An error occured"}, 500
toggle_fav(body.type, body.hash)
return {"msg": "Added to favorites"}
@api.post("/remove")
def remove_favorite(body: FavoritesAddBody):
"""
Removes a favorite from the database.
"""
try:
FavoritesTable.remove_item({"hash": body.hash, "type": body.type})
except Exception as e:
print(e)
return {"msg": "Failed! An error occured"}, 500
toggle_fav(body.type, body.hash)
return {"msg": "Removed from favorites"}
class GetAllOfTypeQuery(GenericLimitSchema):
"""
Extending this class will give you a model with the `limit` field
"""
start: int = Field(
description="Where to start from",
default=Defaults.API_CARD_LIMIT,
)
@api.get("/albums")
def get_favorite_albums(query: GetAllOfTypeQuery):
"""
Get favorite albums
Note: Only the first request will return the total number of favorites.
Others will return -1
"""
fav_albums, total = FavoritesTable.get_fav_albums(query.start, query.limit)
albums = AlbumStore.get_albums_by_hashes(a.hash for a in fav_albums)
return {"albums": serialize_for_card_many(albums), "total": total}
@api.get("/tracks")
def get_favorite_tracks(query: GetAllOfTypeQuery):
"""
Get favorite tracks
Note: Only the first request will return the total number of favorites.
Others will return -1
"""
tracks, total = FavoritesTable.get_fav_tracks(query.start, query.limit)
tracks = TrackStore.get_tracks_by_trackhashes([t.hash for t in tracks])
return {"tracks": serialize_tracks(tracks), "total": total}
@api.get("/artists")
def get_favorite_artists(query: GetAllOfTypeQuery):
"""
Get favorite artists
Note: Only the first request will return the total number of favorites.
Others will return -1
"""
artists, total = FavoritesTable.get_fav_artists(
start=query.start,
limit=query.limit,
)
artists = ArtistStore.get_artists_by_hashes(a.hash for a in artists)
return {"artists": [serialize_artist(a) for a in artists], "total": total}
class GetAllFavoritesQuery(BaseModel):
"""
Extending this class will give you a model with the `limit` field
"""
track_limit: int = Field(
description="The number of tracks to return",
default=Defaults.API_CARD_LIMIT,
)
album_limit: int = Field(
description="The number of albums to return",
default=Defaults.API_CARD_LIMIT,
)
artist_limit: int = Field(
description="The number of artists to return",
default=Defaults.API_CARD_LIMIT,
)
@api.get("")
def get_all_favorites(query: GetAllFavoritesQuery):
"""
Returns all the favorites in the database.
"""
track_limit = query.track_limit
album_limit = query.album_limit
artist_limit = query.artist_limit
# largest is x2 to accound for broken hashes if any
largest = max(track_limit, album_limit, artist_limit)
favs = FavoritesTable.get_all(with_user=True)
favs = sorted(favs, key=lambda x: x.timestamp, reverse=True)
tracks = []
albums = []
artists = []
track_master_hash = TrackStore.trackhashmap.keys()
album_master_hash = AlbumStore.albummap.keys()
artist_master_hash = ArtistStore.artistmap.keys()
# INFO: Filter out invalid hashes (file not found or tags edited)
for fav in favs:
hash = fav.hash
type = fav.type
if type == FavType.track:
tracks.append(hash) if hash in track_master_hash else None
if type == FavType.artist:
artists.append(hash) if hash in artist_master_hash else None
if type == FavType.album:
albums.append(hash) if hash in album_master_hash else None
count = {
"tracks": len(tracks),
"albums": len(albums),
"artists": len(artists),
}
tracks = TrackStore.get_tracks_by_trackhashes(tracks[:track_limit])
albums = AlbumStore.get_albums_by_hashes(albums[:album_limit])
artists = ArtistStore.get_artists_by_hashes(artists[:artist_limit])
recents = []
for fav in favs:
if len(recents) >= largest:
break
if fav.type == FavType.album:
album = next((a for a in albums if a.albumhash == fav.hash), None)
if album is None:
continue
album = serialize_for_card(album)
album["help_text"] = "album"
album["time"] = timestamp_to_time_passed(fav.timestamp)
recents.append(
{
"type": "album",
"item": album,
}
)
if fav.type == FavType.artist:
artist = next((a for a in artists if a.artisthash == fav.hash), None)
if artist is None:
continue
artist = serialize_artist(artist)
artist["help_text"] = "artist"
artist["time"] = timestamp_to_time_passed(fav.timestamp)
recents.append(
{
"type": "artist",
"item": artist,
}
)
if fav.type == FavType.track:
track = next((t for t in tracks if t.trackhash == fav.hash), None)
if track is None:
continue
track = serialize_track(track)
track["help_text"] = "track"
track["time"] = timestamp_to_time_passed(fav.timestamp)
recents.append({"type": "track", "item": track})
return {
"recents": recents[:album_limit],
"tracks": serialize_tracks(tracks[:track_limit]),
"albums": serialize_for_card_many(albums[:album_limit]),
"artists": serialize_for_cards(artists[:artist_limit]),
"count": count,
}
@api.get("/check")
def check_favorite(query: FavoritesAddBody):
"""
Checks if a favorite exists in the database.
"""
itemhash = query.hash
itemtype = query.type
return {"is_favorite": FavoritesTable.check_exists(itemhash, itemtype)}
+303
View File
@@ -0,0 +1,303 @@
"""
Contains all the folder routes.
"""
import pathlib
from datetime import datetime
import os
from pathlib import Path
import psutil
from pydantic import BaseModel, Field
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from showinfm import show_in_file_manager
from swingmusic import settings
from swingmusic.config import UserConfig
from swingmusic.db.libdata import TrackTable
from swingmusic.db.userdata import FavoritesTable, PlaylistTable
from swingmusic.lib.folderslib import get_files_and_dirs, get_folders
from swingmusic.serializers.track import serialize_track, serialize_tracks
from swingmusic.store.tracks import TrackStore
from swingmusic.utils.wintools import is_windows
tag = Tag(name="Folders", description="Get folders and tracks in a directory")
api = APIBlueprint("folder", __name__, url_prefix="/folder", abp_tags=[tag])
class FolderTree(BaseModel):
folder: str = Field("$home", description="The folder to things from")
sorttracksby: str = Field(
"default",
description="""The field to sort tracks by. Options: [
"default",
"album",
"albumartists",
"artists",
"bitrate",
"date",
"disc",
"duration",
"last_mod",
"lastplayed",
"playduration",
"playcount",
"title",
]""",
)
tracksort_reverse: bool = Field(
False,
description="Whether to reverse the sort order of the tracks",
)
sortfoldersby: str = Field(
"lastmod",
description="""The field to sort folders by.
Options: [
"default",
"name",
"lastmod",
"trackcount",
]
""",
)
foldersort_reverse: bool = Field(
False,
description="Whether to reverse the sort order of the folders",
)
start: int = Field(0, description="The start index")
limit: int = Field(50, description="The max number of items to return")
tracks_only: bool = Field(False, description="Whether to only get tracks")
@api.post("")
def get_folder_tree(body: FolderTree):
"""
Get folder
Returns a list of all the folders and tracks in the given folder.
"""
og_req_dir = body.folder
req_dir = body.folder
tracks_only = body.tracks_only
config = UserConfig()
root_dirs = config.rootDirs
if req_dir == "$home" and "$home" in root_dirs:
req_dir = settings.Paths().USER_HOME_DIR.as_posix()
if req_dir == "$home":
folders = get_folders(root_dirs)
return {
"folders": folders,
"tracks": [],
}
if req_dir.startswith("$playlist"):
splits = req_dir.split("/")
if len(splits) == 2:
pid = splits[1]
playlist = PlaylistTable.get_by_id(int(pid))
tracks = TrackStore.get_tracks_by_trackhashes(
playlist.trackhashes[
body.start : body.start + body.limit if body.limit != -1 else None
]
)
return {
"path": req_dir,
"folders": [],
"tracks": serialize_tracks(tracks),
}
playlists = PlaylistTable.get_all()
playlists = sorted(
playlists,
key=lambda p: datetime.strptime(p.last_updated, "%Y-%m-%d %H:%M:%S"),
reverse=True,
)
return {
"path": req_dir,
"folders": [
{
"name": p.name,
"path": f"$playlist/{p.id}",
"trackcount": p.count,
}
for p in playlists
],
"tracks": [],
}
if req_dir == "$favorites":
tracks, total = FavoritesTable.get_fav_tracks(body.start, body.limit)
tracks = TrackStore.get_tracks_by_trackhashes([t.hash for t in tracks])
return {
"tracks": serialize_tracks(tracks),
"folders": [],
"path": req_dir,
}
# TODO: currently only fixed on unix. Windows/Mac still pending.
# note
if not pathlib.Path(req_dir).exists():
req_dir = "/" + req_dir
results = get_files_and_dirs(
pathlib.Path(req_dir),
start=body.start,
limit=body.limit,
tracks_only=tracks_only,
tracksortby=body.sorttracksby,
foldersortby=body.sortfoldersby,
tracksort_reverse=body.tracksort_reverse,
foldersort_reverse=body.foldersort_reverse,
)
if og_req_dir == "$home" and config.showPlaylistsInFolderView:
# Get all playlists and return them as a list of folders
playlists_item = {
"name": "Playlists",
"path": "$playlists",
"trackcount": sum(p.count for p in PlaylistTable.get_all()),
}
favorites_item = {
"name": "Favorites",
"path": "$favorites",
"trackcount": FavoritesTable.get_fav_tracks(0, -1)[1],
}
results["folders"].insert(0, playlists_item)
results["folders"].insert(0, favorites_item)
return results
def get_all_drives(is_win: bool = False):
"""
Returns a list of all the drives on a Windows machine.
"""
drives_ = psutil.disk_partitions(all=True)
drives = [Path(d.mountpoint).as_posix() for d in drives_]
if is_win:
return drives
else:
remove = (
"/boot",
"/tmp",
"/snap",
"/var",
"/sys",
"/proc",
"/etc",
"/run",
"/dev",
)
drives = [d for d in drives if not d.startswith(remove)]
return drives
class DirBrowserBody(BaseModel):
folder: str = Field(
"$root",
description="The folder to list directories from",
)
@api.post("/dir-browser")
def list_folders(body: DirBrowserBody):
"""
List folders
Returns a list of all the folders in the given folder.
Used when selecting root dirs.
"""
req_dir = body.folder
is_win = is_windows()
if req_dir == "$root":
return {
"folders": [{"name": d, "path": d} for d in get_all_drives(is_win=is_win)]
}
req_dir = pathlib.Path(req_dir)
if not req_dir.exists():
req_dir = "/" / req_dir
try:
entries = os.scandir(req_dir)
except PermissionError:
return {"folders": []}
# only get dirs and remove hidden dirs
dirs = []
for entry in entries:
entry = pathlib.Path(entry)
name = entry.name
if name.startswith("$"): # ignore windows system folder
continue
if name.startswith("."): # ignore unix hidden folder
continue
if entry.is_dir(): # lastly, check if is dir
dirs.append({
"name": name,
"path": entry.as_posix()
})
return {
"folders": sorted(dirs, key=lambda i: i["name"]),
}
class FolderOpenInFileManagerQuery(BaseModel):
path: str = Field(
description="The path to open in the file manager",
)
@api.get("/show-in-files")
def open_in_file_manager(query: FolderOpenInFileManagerQuery):
"""
Open in file manager
Opens the given path in the file manager on the host machine.
"""
show_in_file_manager(query.path)
return {"success": True}
class GetTracksInPathQuery(BaseModel):
path: str = Field(
description="The path to get tracks from",
)
@api.get("/tracks/all")
def get_tracks_in_path(query: GetTracksInPathQuery):
"""
Get tracks in path
Gets all (or a max of 300) tracks from the given path and its subdirectories.
Used when adding tracks to the queue.
"""
tracks = TrackTable.get_tracks_in_path(query.path)
tracks = (serialize_track(t) for t in tracks if Path(t.filepath).exists())
return {
"tracks": list(tracks)[:300],
}
+152
View File
@@ -0,0 +1,152 @@
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import BaseModel, Field
from datetime import datetime
from swingmusic.api.apischemas import GenericLimitSchema
from swingmusic.store.albums import AlbumStore
from swingmusic.store.artists import ArtistStore
from swingmusic.serializers.album import serialize_for_card as serialize_album
from swingmusic.serializers.artist import serialize_for_card as serialize_artist
from swingmusic.utils import format_number
from swingmusic.utils.dates import (
create_new_date,
date_string_to_time_passed,
seconds_to_time_string,
timestamp_to_time_passed,
)
bp_tag = Tag(name="Get all", description="List all items")
api = APIBlueprint("getall", __name__, url_prefix="/getall", abp_tags=[bp_tag])
class GetAllItemsQuery(GenericLimitSchema):
start: int = Field(
description="The start index of the items to return",
example=0,
default=0,
)
sortby: str = Field(
description="The key to sort items by",
example="created_date",
default="created_date",
)
reverse: str = Field(
description="Reverse the sort",
example=1,
default="1",
)
class GetAllItemsPath(BaseModel):
itemtype: str = Field(
description="The type of items to return (albums | artists)",
example="albums",
default="albums",
)
@api.get("/<itemtype>")
def get_all_items(path: GetAllItemsPath, query: GetAllItemsQuery):
"""
Get all items
Used to show all albums or artists in the library
Sort keys:
-
Both albums and artists: `duration`, `created_date`, `playcount`, `playduration`, `lastplayed`, `trackcount`
Albums only: `title`, `albumartists`, `date`
Artists only: `name`, `albumcount`
"""
is_albums = path.itemtype == "albums"
is_artists = path.itemtype == "artists"
if is_albums:
items = AlbumStore.get_flat_list()
elif is_artists:
items = ArtistStore.get_flat_list()
total = len(items)
start = query.start
limit = query.limit
sort = query.sortby
reverse = query.reverse == "1"
sort_is_count = sort == "trackcount"
sort_is_duration = sort == "duration"
sort_is_create_date = sort == "created_date"
sort_is_playcount = sort == "playcount"
sort_is_playduration = sort == "playduration"
sort_is_lastplayed = sort == "lastplayed"
sort_is_date = is_albums and sort == "date"
sort_is_artist = is_albums and sort == "albumartists"
sort_is_artist_trackcount = is_artists and sort == "trackcount"
sort_is_artist_albumcount = is_artists and sort == "albumcount"
lambda_sort = lambda x: getattr(x, sort)
lambda_sort_casefold = lambda x: getattr(x, sort).casefold()
if sort_is_artist:
lambda_sort = lambda x: getattr(x, sort)[0]["name"].casefold()
try:
sorted_items = sorted(items, key=lambda_sort_casefold, reverse=reverse)
except AttributeError:
sorted_items = sorted(items, key=lambda_sort, reverse=reverse)
items = sorted_items[start : start + limit]
album_list = []
for item in items:
item_dict = serialize_album(item) if is_albums else serialize_artist(item)
if sort_is_date:
item_dict["help_text"] = datetime.fromtimestamp(item.date).year
if sort_is_create_date:
date = create_new_date(datetime.fromtimestamp(item.created_date))
timeago = date_string_to_time_passed(date)
item_dict["help_text"] = timeago
if sort_is_count:
item_dict["help_text"] = (
f"{format_number(item.trackcount)} track{'' if item.trackcount == 1 else 's'}"
)
if sort_is_duration:
item_dict["help_text"] = seconds_to_time_string(item.duration)
if sort_is_artist_trackcount:
item_dict["help_text"] = (
f"{format_number(item.trackcount)} track{'' if item.trackcount == 1 else 's'}"
)
if sort_is_artist_albumcount:
item_dict["help_text"] = (
f"{format_number(item.albumcount)} album{'' if item.albumcount == 1 else 's'}"
)
if sort_is_playcount:
item_dict["help_text"] = (
f"{format_number(item.playcount)} play{'' if item.playcount == 1 else 's'}"
)
if sort_is_lastplayed:
if item.playduration == 0:
item_dict["help_text"] = "Never played"
else:
item_dict["help_text"] = timestamp_to_time_passed(item.lastplayed)
if sort_is_playduration:
item_dict["help_text"] = seconds_to_time_string(item.playduration)
album_list.append(item_dict)
return {"items": album_list, "total": total}
+38
View File
@@ -0,0 +1,38 @@
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import BaseModel, Field
from swingmusic.api.apischemas import GenericLimitSchema
from swingmusic.lib.home.recentlyadded import get_recently_added_items
from swingmusic.lib.home.get_recently_played import get_recently_played
from swingmusic.store.homepage import HomepageStore
bp_tag = Tag(name="Home", description="Homepage items")
api = APIBlueprint("home", __name__, url_prefix="/nothome", abp_tags=[bp_tag])
@api.get("/recents/added")
def get_recently_added(query: GenericLimitSchema):
"""
Get recently added
"""
return {"items": get_recently_added_items(query.limit)}
@api.get("/recents/played")
def get_recent_plays(query: GenericLimitSchema):
"""
Get recently played
"""
return {"items": get_recently_played(query.limit)}
class HomepageItem(BaseModel):
limit: int = Field(
default=9, description="The max number of items per group to return"
)
@api.get("/")
def homepage_items(query: HomepageItem):
return HomepageStore.get_homepage_items(limit=query.limit)
+259
View File
@@ -0,0 +1,259 @@
from fileinput import filename
from pathlib import Path
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import BaseModel, Field
from flask import send_from_directory
from swingmusic.settings import Defaults, Paths
from swingmusic.store.albums import AlbumStore
from swingmusic.store.tracks import TrackStore
from swingmusic.utils.threading import background
from PIL import Image
bp_tag = Tag(
name="Images", description="Image filenames are constructured as '{itemhash}.webp'"
)
api = APIBlueprint("imgserver", __name__, url_prefix="/img", abp_tags=[bp_tag])
@background
def cache_thumbnails(filepath: Path, trackhash: str):
"""
Resizes the image and stores it in the cache directory.
"""
image = Image.open(filepath)
path = Path(Paths().image_cache_path)
aspect_ratio = image.width / image.height
sizes = {
"xsmall": 64,
"small": 96,
"medium": 256,
"large": 512,
}
for size, width in sizes.items():
width = min(width, image.width)
height = int(width / aspect_ratio)
resized_path = path / size / (trackhash + ".webp")
resized_path.parent.mkdir(parents=True, exist_ok=True)
image.resize((width, height)).save(resized_path, format="webp")
def find_thumbnail(albumhash: str, pathhash: str):
# entry = TrackStore.trackhashmap.get(albumhash)
entry = AlbumStore.albummap.get(albumhash)
if entry is None:
return None, None, ""
track_file = None
tracks = TrackStore.get_tracks_by_trackhashes(entry.trackhashes)
for track in tracks:
if track.pathhash == pathhash:
track_file = track
break
if track_file is None:
return None, None, ""
folder = Path(track_file.folder)
# INFO: Check if the folder has image files
extensions = [".jpg", ".jpeg", ".png", ".webp"]
hierarchy = ["cover", "front", "back", "folder", "album", "artwork"]
images: list[Path] = []
for item in folder.iterdir():
if item.suffix in extensions:
images.append(item)
if len(images) == 0:
return None, None, ""
# INFO: Check if the folder has image files in the hierarchy
for item in hierarchy:
for image in images:
if image.name.lower().startswith(item.lower()):
return image.parent, image.name, track_file.albumhash
# INFO: If no image falls in the hierarchy, return the first image
first_image = images[0]
return first_image.parent, first_image.name, track_file.albumhash
def send_fallback_img(filename: str = "default.webp"):
"""
Returns the fallback image from the assets folder.
"""
folder = Paths().assets_path
img = Path(folder) / filename
if not img.exists():
return "", 404
return send_from_directory(folder, filename)
def send_file_or_fallback(
folder: str, filename: str, fallback: str = "default.webp", pathhash: str = ""
):
"""
Returns the file from the folder or the fallback image.
"""
fpath = Path(folder) / filename
if fpath.exists():
return send_from_directory(folder, filename)
if pathhash != "":
# INFO: Check if the image is in the cache
cache_path = Paths().image_cache_path / fpath.parent.name / filename
if cache_path.exists():
return send_from_directory(cache_path.parent, cache_path.name)
# INFO: Find the thumbnail
parent, file, albumhash = find_thumbnail(
filename.replace(".webp", ""), pathhash
)
# INFO: Cache and send the thumbnail
if file is not None and parent is not None:
cache_thumbnails(parent / file, albumhash)
return send_from_directory(parent, file)
return send_fallback_img(fallback)
class ImagePath(BaseModel):
imgpath: str = Field(
description="The image filename",
example=Defaults.API_ALBUMHASH + ".webp",
)
class ImageQuery(BaseModel):
pathhash: str = Field(
description="The path hash used to find the thumbnail",
default="",
)
# @api.get("/t/o/<imgpath>")
# def send_original_thumbnail(path: ImagePath):
# """
# Get original thumbnail
# """
# folder = Paths.get_original_thumb_path()
# fpath = Path(folder) / path.imgpath
# if fpath.exists():
# return send_from_directory(folder, path.imgpath)
# return send_fallback_img()
# TRACK THUMBNAILS
@api.get("/thumbnail/<imgpath>")
def send_lg_thumbnail(path: ImagePath, query: ImageQuery):
"""
Get large thumbnail (500 x 500)
"""
folder = Paths().lg_thumb_path
return send_file_or_fallback(folder, path.imgpath, pathhash=query.pathhash)
@api.get("/thumbnail/xsmall/<imgpath>")
def send_xsm_thumbnail(path: ImagePath, query: ImageQuery):
"""
Get extra small thumbnail (64px)
"""
folder = Paths().xsm_thumb_path
return send_file_or_fallback(folder, path.imgpath, pathhash=query.pathhash)
@api.get("/thumbnail/small/<imgpath>")
def send_sm_thumbnail(path: ImagePath, query: ImageQuery):
"""
Get small thumbnail (96px)
"""
folder = Paths().sm_thumb_path
return send_file_or_fallback(folder, path.imgpath, pathhash=query.pathhash)
@api.get("/thumbnail/medium/<imgpath>")
def send_md_thumbnail(path: ImagePath, query: ImageQuery):
"""
Get medium thumbnail (256px)
"""
folder = Paths().md_thumb_path
return send_file_or_fallback(folder, path.imgpath, pathhash=query.pathhash)
# ARTISTS
@api.get("/artist/<imgpath>")
def send_lg_artist_image(path: ImagePath):
"""
Get large artist image (500 x 500)
"""
folder = Paths().lg_artist_img_path
return send_file_or_fallback(str(folder), path.imgpath, "artist.webp")
@api.get("/artist/small/<imgpath>")
def send_sm_artist_image(path: ImagePath):
"""
Get small artist image (128)
"""
folder = Paths().sm_artist_img_path
return send_file_or_fallback(str(folder), path.imgpath, "artist.webp")
@api.get("/artist/medium/<imgpath>")
def send_md_artist_image(path: ImagePath):
"""
Get medium artist image (256px)
"""
folder = Paths().md_artist_img_path
return send_file_or_fallback(folder, path.imgpath, "artist.webp")
# PLAYLISTS
class PlaylistImagePath(BaseModel):
imgpath: str = Field(
description="The image path",
example="1.webp",
)
@api.get("/playlist/<imgpath>")
def send_playlist_image(path: PlaylistImagePath):
"""
Get playlist image
Images are constructed as '{playlist_id}.webp'
"""
folder = Paths().playlist_img_path
return send_file_or_fallback(folder, path.imgpath, "playlist.svg")
# MIXES
@api.get("/mix/medium/<imgpath>")
def send_md_mix_image(path: ImagePath):
"""
Get medium mix image
"""
folder = Paths().md_mixes_img_path
return send_file_or_fallback(folder, path.imgpath, "playlist.svg")
@api.get("/mix/small/<imgpath>")
def send_sm_mix_image(path: ImagePath):
"""
Get small mix image
"""
folder = Paths().sm_mixes_img_path
return send_file_or_fallback(folder, path.imgpath, "playlist.svg")
+77
View File
@@ -0,0 +1,77 @@
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import Field
from swingmusic.store.tracks import TrackStore
from swingmusic.api.apischemas import TrackHashSchema
from swingmusic.lib.lyrics import (
get_lyrics_file,
get_lyrics_from_duplicates,
get_lyrics_from_tags,
)
bp_tag = Tag(name="Lyrics", description="Get lyrics")
api = APIBlueprint("lyrics", __name__, url_prefix="/lyrics", abp_tags=[bp_tag])
class SendLyricsBody(TrackHashSchema):
filepath: str = Field(description="The path to the file")
@api.post("")
def send_lyrics(body: SendLyricsBody):
"""
Returns the lyrics for a track
"""
# 1. try to get lyrics by .lrc / .elrc file
# 2. try to get lyrics by extra key
# 3. try to get by duplicates
# 4. iter plugins
filepath = body.filepath
trackhash = body.trackhash
# get copyright first
copyright = ""
if entry:=TrackStore.trackhashmap.get(trackhash, None):
for track in entry.tracks:
copyright = track.copyright
if copyright:
break
lyrics = get_lyrics_file(filepath)
if not lyrics:
lyrics = get_lyrics_from_tags(trackhash) # type: ignore
if not lyrics:
lyrics = get_lyrics_from_duplicates(filepath, trackhash)
# check lyrics plugins
if not lyrics:
return {"error": "No lyrics found"}
if lyrics.is_synced:
text = lyrics.format_synced_lyrics()
else:
text = lyrics.format_unsynced_lyrics()
return {"lyrics": text, "synced": lyrics.is_synced, "copyright": copyright}, 200
@api.post("/check")
def check_lyrics(body: SendLyricsBody):
"""
Checks if lyrics file or tag exists for a track
"""
result = send_lyrics(body)
if "error" in result:
return {"exists": False}
else:
return {"exists": True}, 200
+484
View File
@@ -0,0 +1,484 @@
"""
All playlist-related routes.
"""
import json
from datetime import datetime
import pathlib
from typing import Any
from PIL import UnidentifiedImageError, Image
from pydantic_core import core_schema
from pydantic import BaseModel, Field, GetCoreSchemaHandler
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint, FileStorage as _FileStorage
from swingmusic import models
from swingmusic.api.apischemas import GenericLimitSchema
from swingmusic.db.userdata import PlaylistTable
from swingmusic.lib import playlistlib
from swingmusic.lib.albumslib import sort_by_track_no
from swingmusic.lib.home.recentlyadded import get_recently_added_playlist
from swingmusic.lib.home.recentlyplayed import get_recently_played_playlist
from swingmusic.lib.sortlib import sort_tracks
from swingmusic.models.playlist import Playlist
from swingmusic.serializers.playlist import serialize_for_card
from swingmusic.serializers.track import serialize_tracks
from swingmusic.store.tracks import TrackStore
from swingmusic.utils.dates import create_new_date, date_string_to_time_passed
from swingmusic.settings import Paths
tag = Tag(name="Playlists", description="Get and manage playlists")
api = APIBlueprint("playlists", __name__, url_prefix="/playlists", abp_tags=[tag])
def insert_playlist(name: str, image: str = None):
playlist = {
"image": image,
"last_updated": create_new_date(),
"name": name,
"trackhashes": [],
"settings": {
"has_gif": False,
"banner_pos": 50,
"square_img": True if image else False,
"pinned": False,
},
}
rowid = PlaylistTable.add_one(playlist)
if rowid:
playlist["id"] = rowid
return Playlist(**playlist)
return None
def get_path_trackhashes(path: str, tracksortby: str, reverse: bool):
"""
Returns a list of trackhashes in a folder.
"""
tracks = TrackStore.get_tracks_in_path(path)
tracks = sort_tracks(tracks, key=tracksortby, reverse=reverse)
return [t.trackhash for t in tracks]
def get_album_trackhashes(albumhash: str):
"""
Returns a list of trackhashes in an album.
"""
tracks = TrackStore.get_tracks_by_albumhash(albumhash)
tracks = sort_by_track_no(tracks)
return [t.trackhash for t in tracks]
def get_artist_trackhashes(artisthash: str):
"""
Returns a list of trackhashes for an artist.
"""
tracks = TrackStore.get_tracks_by_artisthash(artisthash)
tracks = sort_tracks(tracks, key="playcount", reverse=True)
return [t.trackhash for t in tracks]
def format_custom_playlist(playlist: models.Playlist, tracks: list[models.Track]):
playlist.duration = sum(t.duration for t in tracks)
playlist.count = len(tracks)
return {
"info": serialize_for_card(playlist),
"tracks": serialize_tracks(tracks),
}
class SendAllPlaylistsQuery(BaseModel):
no_images: bool = Field(False, description="Whether to include images")
@api.get("")
def send_all_playlists(query: SendAllPlaylistsQuery):
"""
Gets all the playlists.
"""
playlists = PlaylistTable.get_all()
playlists = sorted(
playlists,
key=lambda p: datetime.strptime(p.last_updated, "%Y-%m-%d %H:%M:%S"),
reverse=True,
)
for playlist in playlists:
if not playlist.has_image:
playlist.images = playlistlib.get_first_4_images(
trackhashes=playlist.trackhashes
)
playlist.clear_lists()
# playlists.sort(
# key=lambda p: datetime.strptime(p.last_updated, "%Y-%m-%d %H:%M:%S"),
# reverse=True,
# )
return {"data": playlists}
class CreatePlaylistBody(BaseModel):
name: str = Field(..., description="The name of the playlist")
@api.post("/new")
def create_playlist(body: CreatePlaylistBody):
"""
New playlist
Creates a new playlist. Accepts POST method with a JSON body.
"""
exists = PlaylistTable.check_exists_by_name(body.name)
if exists:
return {"error": "Playlist already exists"}, 409
playlist = insert_playlist(body.name)
if playlist is None:
return {"error": "Playlist could not be created"}, 500
return {"playlist": playlist}, 201
class PlaylistIDPath(BaseModel):
# INFO: playlistid string examples: "recentlyadded"
playlistid: str = Field(..., description="The ID of the playlist")
class AddItemToPlaylistBody(BaseModel):
itemtype: str = Field(
default="tracks",
description="The type of item to add",
examples=["tracks", "folder", "album", "artist"],
)
sortoptions: dict = Field(
default=None,
description="The sort options for the tracks",
)
itemhash: str = Field(..., description="The hash of the item to add")
@api.post("/<playlistid>/add")
def add_item_to_playlist(path: PlaylistIDPath, body: AddItemToPlaylistBody):
"""
Add to playlist.
If itemtype is not "tracks", itemhash is expected to be a folder, album or artist hash.
"""
itemtype = body.itemtype
itemhash = body.itemhash
playlist_id = int(path.playlistid)
sortoptions = body.sortoptions
if itemtype == "tracks":
trackhashes = itemhash.split(",")
if len(trackhashes) == 1 and trackhashes[0] in PlaylistTable.get_trackhashes(playlist_id):
return {"msg": "Track already exists in playlist"}, 409
elif itemtype == "folder":
trackhashes = get_path_trackhashes(
itemhash,
sortoptions.get("tracksortby") or "default",
sortoptions.get("tracksortreverse") or False,
)
elif itemtype == "album":
trackhashes = get_album_trackhashes(itemhash)
elif itemtype == "artist":
trackhashes = get_artist_trackhashes(itemhash)
else:
trackhashes = []
PlaylistTable.append_to_playlist(playlist_id, trackhashes)
return {"msg": "Done"}, 200
class GetPlaylistQuery(GenericLimitSchema):
no_tracks: bool = Field(False, description="Whether to include tracks")
start: int = Field(0, description="The start index of the tracks")
@api.get("/<playlistid>")
def get_playlist(path: PlaylistIDPath, query: GetPlaylistQuery):
"""
Get playlist by id
"""
no_tracks = query.no_tracks
playlistid = path.playlistid
custom_playlists = [
{"name": "recentlyadded", "handler": get_recently_added_playlist},
{"name": "recentlyplayed", "handler": get_recently_played_playlist},
]
is_custom = playlistid in {p["name"] for p in custom_playlists}
if is_custom:
if query.start != 0:
return {
"tracks": [],
}
handler = next(
p["handler"] for p in custom_playlists if p["name"] == playlistid
)
playlist, tracks = handler()
return format_custom_playlist(playlist, tracks)
playlist = PlaylistTable.get_by_id(int(playlistid))
if playlist is None:
return {"msg": "Playlist not found"}, 404
if query.limit == -1:
query.limit = len(playlist.trackhashes) - 1
tracks = TrackStore.get_tracks_by_trackhashes(
playlist.trackhashes[query.start : query.start + query.limit]
)
duration = sum(t.duration for t in tracks)
playlist._last_updated = date_string_to_time_passed(playlist.last_updated)
playlist.duration = duration
playlist.images = playlistlib.get_first_4_images(tracks)
playlist.clear_lists()
return {
"info": playlist,
"tracks": serialize_tracks(tracks) if not no_tracks else [],
}
class FileStorage(_FileStorage):
@classmethod
def __get_pydantic_core_schema__(
cls, _source: Any, handler: GetCoreSchemaHandler
) -> core_schema.CoreSchema:
return core_schema.with_info_plain_validator_function(cls.validate)
class UpdatePlaylistForm(BaseModel):
image: FileStorage = Field(description="The image file")
name: str = Field(..., description="The name of the playlist")
settings: str = Field(
...,
description="The settings of the playlist",
json_schema_extra={
"example": '{"has_gif": false, "banner_pos": 50, "square_img": false, "pinned": false}'
},
)
@api.put("/<playlistid>/update", methods=["PUT"])
def update_playlist_info(path: PlaylistIDPath, form: UpdatePlaylistForm):
"""
Update playlist
"""
playlistid = path.playlistid
db_playlist = PlaylistTable.get_by_id(playlistid)
if db_playlist is None:
return {"error": "Playlist not found"}, 404
image = form.image
if form.image:
image = form.image
settings = json.loads(form.settings)
settings["has_gif"] = False
playlist = {
"id": int(playlistid),
"image": db_playlist.image,
"last_updated": create_new_date(),
"name": str(form.name).strip(),
"settings": settings,
}
if image:
try:
pil_image = Image.open(image)
content_type = image.content_type
playlist["image"] = playlistlib.save_p_image(
pil_image, playlistid, content_type
)
if image.content_type == "image/gif":
playlist["settings"]["has_gif"] = True
except UnidentifiedImageError:
return {"error": "Failed: Invalid image"}, 400
p_tuple = (*playlist.values(),)
PlaylistTable.update_one(playlistid, playlist)
playlistlib.cleanup_playlist_images()
playlist = models.Playlist(*p_tuple)
playlist.last_updated = date_string_to_time_passed(playlist.last_updated)
return {
"data": playlist,
}
@api.post("/<playlistid>/pin_unpin")
def pin_unpin_playlist(path: PlaylistIDPath):
"""
Pin playlist.
"""
playlist = PlaylistTable.get_by_id(path.playlistid)
if playlist is None:
return {"error": "Playlist not found"}, 404
settings = playlist.settings
try:
settings["pinned"] = not settings["pinned"]
except KeyError:
settings["pinned"] = True
PlaylistTable.update_settings(path.playlistid, settings)
return {"msg": "Done"}, 200
@api.delete("/<playlistid>/remove-img")
def remove_playlist_image(path: PlaylistIDPath):
"""
Clear playlist image.
"""
playlist = PlaylistTable.get_by_id(path.playlistid)
if playlist is None:
return {"error": "Playlist not found"}, 404
PlaylistTable.remove_image(path.playlistid)
playlist.image = None
playlist.thumb = None
playlist.settings["has_gif"] = False
playlist.has_image = False
playlist.images = playlistlib.get_first_4_images(trackhashes=playlist.trackhashes)
playlist.last_updated = date_string_to_time_passed(playlist.last_updated)
return {"playlist": playlist}, 200
@api.delete("/<playlistid>/delete", methods=["DELETE"])
def remove_playlist(path: PlaylistIDPath):
"""
Delete playlist
"""
PlaylistTable.remove_one(path.playlistid)
playlistlib.cleanup_playlist_images()
return {"msg": "Done"}, 200
class RemoveTracksFromPlaylistBody(BaseModel):
tracks: list[dict] = Field(..., description="A list of trackhashes to remove")
@api.post("/<playlistid>/remove-tracks")
def remove_tracks_from_playlist(
path: PlaylistIDPath, body: RemoveTracksFromPlaylistBody
):
"""
Remove track from playlist
"""
# A track looks like this:
# {
# trackhash: str;
# index: int;
# }
PlaylistTable.remove_from_playlist(path.playlistid, body.tracks)
return {"msg": "Done"}, 200
class SavePlaylistAsItemBody(BaseModel):
itemtype: str = Field(..., description="The type of item", example="tracks")
playlist_name: str = Field(..., description="The name of the playlist")
itemhash: str = Field(..., description="The hash of the item to save")
sortoptions: dict = Field(
default=dict(),
description="The sort options for the tracks",
)
@api.post("/save-item")
def save_item_as_playlist(body: SavePlaylistAsItemBody):
"""
Save as playlist
Saves a track, album, artist or folder as a playlist
"""
itemtype = body.itemtype
playlist_name = body.playlist_name
itemhash = body.itemhash
sortoptions = body.sortoptions
if PlaylistTable.check_exists_by_name(playlist_name):
return {"error": "Playlist already exists"}, 409
if itemtype == "tracks":
trackhashes = itemhash.split(",")
elif itemtype == "folder":
trackhashes = get_path_trackhashes(
itemhash,
sortoptions.get("tracksortby") or "default",
sortoptions.get("tracksortreverse") or False,
)
elif itemtype == "album":
trackhashes = get_album_trackhashes(itemhash)
elif itemtype == "artist":
trackhashes = get_artist_trackhashes(itemhash)
else:
trackhashes = []
if len(trackhashes) == 0:
return {"error": "No tracks founds"}, 404
image = (
itemhash + ".webp" if itemtype != "folder" and itemtype != "tracks" else None
)
playlist = insert_playlist(playlist_name, image)
if playlist is None:
return {"error": "Playlist could not be created"}, 500
# save image
if itemtype != "folder" and itemtype != "tracks":
filename = itemhash + ".webp"
base_path = (
Paths().lg_artist_img_path
if itemtype == "artist"
else Paths().lg_thumb_path()
)
img_path = pathlib.Path(base_path + "/" + filename)
if img_path.exists():
img = Image.open(img_path)
playlistlib.save_p_image(
img, str(playlist.id), "image/webp", filename=filename
)
PlaylistTable.append_to_playlist(playlist.id, trackhashes)
playlist.count = len(trackhashes)
images = playlistlib.get_first_4_images(trackhashes=trackhashes)
playlist.images = [img["image"] for img in images]
return {"playlist": playlist}, 201
+103
View File
@@ -0,0 +1,103 @@
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import BaseModel, Field
from swingmusic.api.auth import admin_required
from swingmusic.config import UserConfig
from swingmusic.db.userdata import PluginTable
from swingmusic.plugins.lastfm import LastFmPlugin
from swingmusic.utils.auth import get_current_userid
bp_tag = Tag(name="Plugins", description="Manage plugins")
api = APIBlueprint("plugins", __name__, url_prefix="/plugins", abp_tags=[bp_tag])
@api.get("/")
def get_all_plugins():
"""
List all plugins
"""
plugins = PluginTable.get_all()
return {"plugins": plugins}
class PluginBody(BaseModel):
plugin: str = Field(description="The plugin name", example="lyrics")
class PluginActivateBody(PluginBody):
active: bool = Field(
description="New plugin active state", example=False, default=False
)
@api.post("/setactive")
@admin_required()
def activate_deactivate_plugin(body: PluginActivateBody):
"""
Activate/Deactivate plugin
"""
name = body.plugin
PluginTable.activate(name, body.active)
return {"message": "OK"}, 200
class PluginSettingsBody(PluginBody):
settings: dict = Field(
description="The new plugin settings", example={"key": "value"}
)
@api.post("/settings")
@admin_required()
def update_plugin_settings(body: PluginSettingsBody):
"""
Update plugin settings
"""
plugin = body.plugin
settings = body.settings
if not plugin or not settings:
return {"error": "Missing plugin or settings"}, 400
PluginTable.update_settings(plugin, settings)
plugin = PluginTable.get_by_name(plugin)
return {"status": "success", "settings": plugin.settings}
class LastFmSessionBody(BaseModel):
token: str = Field(description="The token to use to create the session")
@api.post("/lastfm/session/create")
def create_lastfm_session(body: LastFmSessionBody):
"""
Create a Last.fm session
"""
if not body.token:
return {"error": "Missing token"}, 400
lastfm = LastFmPlugin()
session_key = lastfm.get_session_key(body.token)
if session_key:
config = UserConfig()
current_user = get_current_userid()
config.lastfmSessionKeys[str(current_user)] = session_key
config.lastfmSessionKeys = config.lastfmSessionKeys
return {"status": "success", "session_key": session_key}
@api.post("/lastfm/session/delete")
def delete_lastfm_session():
"""
Delete the Last.fm session
"""
config = UserConfig()
current_user = get_current_userid()
config.lastfmSessionKeys[str(current_user)] = ""
config.lastfmSessionKeys = config.lastfmSessionKeys
return {"status": "success"}
+60
View File
@@ -0,0 +1,60 @@
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import Field
from swingmusic.api.apischemas import TrackHashSchema
from swingmusic.lib.lyrics import Lyrics as Lyrics_class
from swingmusic.plugins.lyrics import Lyrics
from swingmusic.settings import Defaults
from swingmusic.utils.hashing import create_hash
bp_tag = Tag(name="Lyrics Plugin", description="Musixmatch lyrics plugin")
api = APIBlueprint(
"lyricsplugin", __name__, url_prefix="/plugins/lyrics", abp_tags=[bp_tag]
)
class LyricsSearchBody(TrackHashSchema):
title: str = Field(description="The track title ", example=Defaults.API_TRACKNAME)
artist: str = Field(description="The track artist ", example=Defaults.API_ARTISTNAME)
album: str = Field(description="The track track album ", example=Defaults.API_ALBUMNAME)
filepath: str = Field(
description="Track filepath to save the lyrics file relative to",
example="/home/cwilvx/temp/crazy song.mp3",
)
@api.post("/search")
def search_lyrics(body: LyricsSearchBody):
"""
Search for lyrics by title and artist
"""
title = body.title
artist = body.artist
album = body.album
filepath = body.filepath
trackhash = body.trackhash
finder = Lyrics()
data = finder.search_lyrics_by_title_and_artist(title, artist)
if not data:
return {"trackhash": trackhash, "lyrics": None}
perfect_match = data[0]
for track in data:
i_title = track["title"]
i_album = track["album"]
if create_hash(i_title) == create_hash(title) and create_hash(i_album) == create_hash(album):
perfect_match = track
track_id = perfect_match["track_id"]
lrc = finder.download_lyrics(track_id, filepath)
if lrc is not None:
lyrics = Lyrics_class(lrc)
return {"trackhash": trackhash, "lyrics": lyrics.format_synced_lyrics()}, 200
return {"trackhash": trackhash, "lyrics": lrc}, 200
+109
View File
@@ -0,0 +1,109 @@
from typing import Literal
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import BaseModel, Field
from swingmusic.db.userdata import MixTable
from swingmusic.plugins.mixes import MixesPlugin
from swingmusic.store.homepage import HomepageStore
from swingmusic.store.tracks import TrackStore
bp_tag = Tag(name="Mixes Plugin", description="Mixes plugin hehe")
api = APIBlueprint(
"mixesplugin", __name__, url_prefix="/plugins/mixes", abp_tags=[bp_tag]
)
class GetMixesBody(BaseModel):
mixtype: Literal["artists", "tracks"] = Field(description="The type of mix")
@api.get("/<mixtype>")
def get_artist_mixes(path: GetMixesBody):
srcmixes = MixTable.get_all(with_userid=True)
mixes = []
if path.mixtype == "artists":
mixes = [mix.to_dict(convert_timestamp=True) for mix in srcmixes]
elif path.mixtype == "tracks":
plugin = MixesPlugin()
for mix in srcmixes:
custom_mix = plugin.get_track_mix(mix)
if custom_mix:
mixes.append(custom_mix.to_dict(convert_timestamp=True))
seen_mixids = set()
# filter duplicates by trackshash
final_mixes = []
for mix in mixes:
# INFO: Ignore duplicates for artist mixes
if mix["id"] in seen_mixids and path.mixtype == "tracks":
continue
final_mixes.append(mix)
seen_mixids.add(mix["id"])
return final_mixes
class MixQuery(BaseModel):
mixid: str = Field(description="The mix id")
sourcehash: str = Field(description="The sourcehash of the mix")
@api.get("/")
def get_mix(query: MixQuery):
mixtype = ""
match query.mixid[0]:
case "a":
mixtype = "artist_mixes"
case "t":
mixtype = "custom_mixes"
case _:
return {"msg": "Invalid mix ID"}, 400
# INFO: Check if the mix is already in the homepage store
mix = HomepageStore.get_mix(mixtype, query.mixid)
if mix and mix["sourcehash"] == query.sourcehash:
return mix, 200
# INF0: Get the mix from the db
mix = MixTable.get_by_sourcehash(query.sourcehash)
if not mix:
return {"msg": "Mix not found"}, 404
if mixtype == "custom_mixes":
mix = MixesPlugin.get_track_mix(mix)
if not mix:
return {"msg": "Mix not found"}, 404
return mix.to_full_dict(), 200
class SaveMixRequest(BaseModel):
mixid: str = Field(description="The id of the mix")
type: str = Field(description="The type of mix")
sourcehash: str = Field(description="The sourcehash of the mix")
@api.post("/save")
def save_mix(body: SaveMixRequest):
mix_type = body.type
mix_sourcehash = body.sourcehash
if mix_type == "artist":
state = MixTable.save_artist_mix(mix_sourcehash)
elif mix_type == "track":
state = MixTable.save_track_mix(mix_sourcehash)
mix = HomepageStore.find_mix(body.mixid)
if mix:
mix.saved = state
return {"msg": "Mixes saved"}, 200
+380
View File
@@ -0,0 +1,380 @@
from gettext import ngettext
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
import pendulum
from pydantic import Field, BaseModel
from swingmusic.api.apischemas import TrackHashSchema
from typing import Literal
import locale
from swingmusic.db.userdata import FavoritesTable, ScrobbleTable
from swingmusic.lib.extras import get_extra_info
from swingmusic.lib.recipes.recents import RecentlyPlayed
from swingmusic.models.album import Album
from swingmusic.models.stats import StatItem
from swingmusic.models.track import Track
from swingmusic.plugins.lastfm import LastFmPlugin
from swingmusic.serializers.artist import serialize_for_card
from swingmusic.serializers.album import serialize_for_card as serialize_for_album_card
from swingmusic.serializers.track import serialize_track, serialize_tracks
from swingmusic.settings import Defaults
from swingmusic.store.albums import AlbumStore
from swingmusic.store.artists import ArtistStore
from swingmusic.store.tracks import TrackStore
from swingmusic.utils.dates import (
get_date_range,
get_duration_in_seconds,
seconds_to_time_string,
)
from swingmusic.utils.stats import (
calculate_album_trend,
calculate_artist_trend,
calculate_new_albums,
calculate_new_artists,
calculate_scrobble_trend,
calculate_track_trend,
get_albums_in_period,
get_artists_in_period,
get_tracks_in_period,
)
bp_tag = Tag(name="Logger", description="Log item plays")
api = APIBlueprint("logger", __name__, url_prefix="/logger", abp_tags=[bp_tag])
class LogTrackBody(TrackHashSchema):
timestamp: int = Field(description="The timestamp of the track")
duration: int = Field(description="The duration of the track in seconds")
source: str = Field(
description="The play source of the track",
json_schema_extra={
"examples": [
f"al:{Defaults.API_ALBUMHASH}",
f"tr:{Defaults.API_TRACKHASH}",
f"ar:{Defaults.API_ARTISTHASH}",
]
},
)
def format_date(start: float, end: float):
return f"{pendulum.from_timestamp(start).format('MMM D, YYYY')} - {pendulum.from_timestamp(end).format('MMM D, YYYY')}"
@api.post("/track/log")
def log_track(body: LogTrackBody):
"""
Log a track play to the database.
"""
timestamp = body.timestamp
duration = body.duration
if not timestamp or duration < 5:
return {"msg": "Invalid entry."}, 400
trackentry = TrackStore.trackhashmap.get(body.trackhash)
if trackentry is None:
return {"msg": "Track not found."}, 404
scrobble_data = dict(body)
# REVIEW: Do we need to store the extra info in the database?
# OR .... can we just write it to the backup file on demand?
scrobble_data["extra"] = get_extra_info(body.trackhash, "track")
ScrobbleTable.add(scrobble_data)
# NOTE: Update the recently played homepage for this userid
RecentlyPlayed(userid=scrobble_data["userid"])
# Update play data on the in-memory stores
track = trackentry.tracks[0]
album = AlbumStore.albummap.get(track.albumhash)
if album:
album.increment_playcount(duration, timestamp)
for hash in track.artisthashes:
artist = ArtistStore.artistmap.get(hash)
if artist:
artist.increment_playcount(duration, timestamp)
trackentry.increment_playcount(duration, timestamp)
track = trackentry.tracks[0]
lastfm = LastFmPlugin()
if (
lastfm.enabled
and track.duration > 30
and body.duration >= min(track.duration / 2, 240)
# SEE: https://www.last.fm/api/scrobbling#when-is-a-scrobble-a-scrobble
):
lastfm.scrobble(trackentry.tracks[0], timestamp)
return {"msg": "recorded"}, 201
class ChartItemsQuery(BaseModel):
duration: Literal["week", "month", "year", "alltime"] = Field(
"year",
description="Duration to fetch data for",
)
limit: int = Field(10, description="Number of top tracks to return")
order_by: Literal["playcount", "playduration"] = Field(
"playduration", description="Property to order by"
)
# SECTION: STATS
def get_help_text(
playcount: int, playduration: int, order_by: Literal["playcount", "playduration"]
):
"""
Get the help text given the playcount and playduration.
"""
if order_by == "playcount":
if playcount == 0:
return "unplayed"
return f"{playcount} play{'' if playcount == 1 else 's'}"
if order_by == "playduration":
return seconds_to_time_string(playduration)
# DISCLAIMER: Code beyond this point was partially written by Claude 3.5 Sonnet in Cursor.
# TODO: Refactor, group and clean up
@api.get("/top-tracks")
def get_top_tracks(query: ChartItemsQuery):
"""
Get the top N tracks played within a given duration.
"""
start_time, end_time = get_date_range(query.duration)
previous_start_time = start_time - get_duration_in_seconds(query.duration)
current_period_tracks, current_period_scrobbles, duration = get_tracks_in_period(
start_time, end_time
)
previous_period_tracks, previous_period_scrobbles, _ = get_tracks_in_period(
previous_start_time, start_time
)
scrobble_trend = (
"rising"
if current_period_scrobbles > previous_period_scrobbles
else (
"falling"
if current_period_scrobbles < previous_period_scrobbles
else "stable"
)
)
sorted_tracks = sort_tracks(current_period_tracks, query.order_by)
top_tracks = sorted_tracks[: query.limit]
response = []
for track in top_tracks:
trend = calculate_track_trend(
track, current_period_tracks, previous_period_tracks
)
track = {
**serialize_track(track),
"trend": trend,
"help_text": get_help_text(
track.playcount, track.playduration, query.order_by
),
}
response.append(track)
return {
"tracks": response,
"scrobbles": {
"text": f"{current_period_scrobbles} total play{'' if current_period_scrobbles == 1 else 's'} ({seconds_to_time_string(duration)})",
"trend": scrobble_trend,
"dates": format_date(start_time, end_time),
},
}, 200
def sort_tracks(tracks: list[Track], order_by: Literal["playcount", "playduration"]):
return sorted(tracks, key=lambda x: getattr(x, order_by), reverse=True)
@api.get("/top-artists")
def get_top_artists(query: ChartItemsQuery):
"""
Get the top N artists played within a given duration.
"""
start_time, end_time = get_date_range(query.duration)
previous_start_time = start_time - get_duration_in_seconds(query.duration)
current_period_artists = get_artists_in_period(start_time, end_time)
previous_period_artists = get_artists_in_period(previous_start_time, start_time)
new_artists = calculate_new_artists(current_period_artists, start_time)
scrobble_trend = calculate_scrobble_trend(
len(current_period_artists), len(previous_period_artists)
)
sorted_artists = sort_artists(current_period_artists, query.order_by)
top_artists = sorted_artists[: query.limit]
response = []
for artist in top_artists:
trend = calculate_artist_trend(
artist, current_period_artists, previous_period_artists
)
db_artist = ArtistStore.get_artist_by_hash(artist["artisthash"])
if db_artist is None:
continue
artist = {
**serialize_for_card(db_artist),
"trend": trend,
"help_text": get_help_text(
artist["playcount"], artist["playduration"], query.order_by
),
"extra": {
"playcount": artist["playcount"],
},
}
response.append(artist)
return {
"artists": response,
"scrobbles": {
"text": f"{new_artists} {'new' if query.duration != 'alltime' else ''} {ngettext('artist', 'artists', new_artists)}",
"trend": scrobble_trend,
"dates": format_date(start_time, end_time),
},
}, 200
def sort_artists(artists, order_by):
return sorted(artists, key=lambda x: x[order_by], reverse=True)
@api.get("/top-albums")
def get_top_albums(query: ChartItemsQuery):
"""
Get the top N albums played within a given duration.
"""
start_time, end_time = get_date_range(query.duration)
previous_start_time = start_time - get_duration_in_seconds(query.duration)
current_period_albums = get_albums_in_period(start_time, end_time)
previous_period_albums = get_albums_in_period(previous_start_time, start_time)
new_albums = calculate_new_albums(current_period_albums, previous_period_albums)
scrobble_trend = calculate_scrobble_trend(
len(current_period_albums), len(previous_period_albums)
)
sorted_albums = sort_albums(current_period_albums, query.order_by)
top_albums = sorted_albums[: query.limit]
response = []
for album in top_albums:
trend = calculate_album_trend(
album, current_period_albums, previous_period_albums
)
album = {
**serialize_for_album_card(album),
"trend": trend,
"help_text": get_help_text(
album.playcount, album.playduration, query.order_by
),
}
response.append(album)
return {
"albums": response,
"scrobbles": {
"text": f"{new_albums} new album{'' if new_albums == 1 else 's'} played",
"trend": scrobble_trend,
"dates": format_date(start_time, end_time),
},
}, 200
def sort_albums(albums: list[Album], order_by: Literal["playcount", "playduration"]):
return sorted(albums, key=lambda x: getattr(x, order_by), reverse=True)
@api.get("/stats")
def get_stats():
"""
Get the stats for the user.
"""
period = "week"
start_time, end_time = get_date_range(period)
said_period = period
match period:
case "week":
said_period = "this week"
case "month":
said_period = "this month"
case "year":
said_period = "this year"
case "alltime":
said_period = "all time"
count = len(TrackStore.get_flat_list())
total_tracks = StatItem(
"trackcount",
"in your library",
locale.format_string("%d", count, grouping=True)
+ " "
+ ngettext("track", "tracks", count),
)
tracks, playcount, playduration = get_tracks_in_period(start_time, end_time)
playcount = StatItem(
"streams",
said_period,
f"{playcount} track {ngettext('play', 'plays', playcount)}",
)
playduration = StatItem(
"playtime",
said_period,
f"{seconds_to_time_string(playduration)} listened",
)
tracks = sorted(tracks, key=lambda t: t.playduration, reverse=True)
# Find the top track from the last 7 days
top_track = StatItem(
"toptrack",
f"Top track {said_period}",
(
tracks[0].title + " - " + tracks[0].artists[0]["name"]
if len(tracks) > 0
else ""
),
(tracks[0].image if len(tracks) > 0 else None),
)
fav_count = FavoritesTable.count_favs_in_period(start_time, end_time)
favorites = StatItem(
"favorites",
said_period,
f"{fav_count} {'new' if period != 'alltime' else ''} favorite{'' if fav_count == 1 else 's'}",
)
return {
"stats": [
top_track,
playcount,
playduration,
favorites,
total_tracks,
],
"dates": format_date(start_time, end_time),
}, 200
+124
View File
@@ -0,0 +1,124 @@
"""
Contains all the search routes.
"""
from typing import Any, Literal
from unidecode import unidecode
from pydantic import Field
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from swingmusic import models
from swingmusic.api.apischemas import GenericLimitSchema
from swingmusic.lib import searchlib
from swingmusic.serializers.artist import serialize_for_cards
from swingmusic.settings import Defaults
from swingmusic.store.tracks import TrackStore
tag = Tag(name="Search", description="Search for tracks, albums and artists")
api = APIBlueprint("search", __name__, url_prefix="/search", abp_tags=[tag])
SEARCH_COUNT = 30
"""
The max amount of items to return per request
"""
class SearchQuery(GenericLimitSchema):
q: str = Field(
description="The search query",
json_schema_extra={"example": "Fleetwood Mac"},
)
start: int = Field(description="The index to start from", default=0)
limit: int = Field(
description="The number of items to return", default=SEARCH_COUNT
)
class TopResultsQuery(SearchQuery):
limit: int = Field(
description="The number of items to return", default=Defaults.API_CARD_LIMIT
)
class SearchLoadMoreQuery(SearchQuery):
itemtype: Literal["tracks", "albums", "artists"] = Field(
description="The type of search",
json_schema_extra={"example": "tracks"},
)
class Search:
def __init__(self, query: str) -> None:
self.tracks: list[models.Track] = []
self.query = unidecode(query)
def search_tracks(self):
"""
Calls :class:`SearchTracks` which returns the tracks that fuzzily match
the search terms. Then adds them to the `SearchResults` store.
"""
self.tracks = TrackStore.get_flat_list()
return searchlib.TopResults().search(self.query, tracks_only=True)
def search_artists(self):
"""Calls :class:`SearchArtists` which returns the artists that fuzzily match
the search term. Then adds them to the `SearchResults` store.
"""
artists = searchlib.SearchArtists(self.query)()
return serialize_for_cards(artists)
def search_albums(self):
"""Calls :class:`SearchAlbums` which returns the albums that fuzzily match
the search term. Then adds them to the `SearchResults` store.
"""
return searchlib.TopResults().search(self.query, albums_only=True)
def get_top_results(
self,
limit: int,
):
finder = searchlib.TopResults()
return finder.search(self.query, limit=limit)
@api.get("/top")
def get_top_results(query: TopResultsQuery):
"""
Get top results
Returns the top results for the given query.
"""
if not query.q:
return {"error": "No query provided"}, 400
return Search(query.q).get_top_results(limit=query.limit)
@api.get("/")
def search_items(query: SearchLoadMoreQuery):
"""
Find tracks, albums or artists from a search query.
"""
results: Any = []
match query.itemtype:
case "tracks":
results = Search(query.q).search_tracks()
case "albums":
results = Search(query.q).search_albums()
case "artists":
results = Search(query.q).search_artists()
case _:
return {
"error": "Invalid item type. Valid types are 'tracks', 'albums' and 'artists'"
}, 400
return {
"results": results[query.start : query.start + query.limit],
"more": len(results) > query.start + query.limit,
}
# TODO: Rewrite this file using generators where possible
+174
View File
@@ -0,0 +1,174 @@
from dataclasses import asdict
from importlib import metadata
from typing import Any
from flask_openapi3 import Tag
from flask_openapi3 import APIBlueprint
from pydantic import BaseModel, Field
from swingmusic.api.auth import admin_required
from swingmusic.db.userdata import PluginTable
from swingmusic.lib.index import index_everything
from swingmusic.config import UserConfig
from swingmusic.utils.auth import get_current_userid
bp_tag = Tag(name="Settings", description="Customize stuff")
api = APIBlueprint("settings", __name__, url_prefix="/notsettings", abp_tags=[bp_tag])
def get_child_dirs(parent: str, children: list[str]):
"""Returns child directories in a list, given a parent directory"""
return [_dir for _dir in children if _dir.startswith(parent) and _dir != parent]
class AddRootDirsBody(BaseModel):
new_dirs: list[str] = Field(
description="The new directories to add",
example=["/home/user/Music", "/home/user/Downloads"],
)
removed: list[str] = Field(
description="The directories to remove",
example=["/home/user/Downloads"],
)
@api.post("/add-root-dirs")
@admin_required()
def add_root_dirs(body: AddRootDirsBody):
"""
Add custom root directories to the database.
"""
new_dirs = body.new_dirs
removed_dirs = body.removed
config = UserConfig()
db_dirs = config.rootDirs
home = "$home"
db_home = any([d == home for d in db_dirs]) # if $home is in db
incoming_home = any([d == home for d in new_dirs]) # if $home is in incoming
# handle $home case
if db_home and incoming_home:
return {"msg": "Not changed!"}, 304
# if $home is the current root dir or the incoming root dir
# is $home, remove all root dirs
if db_home or incoming_home:
config.rootDirs = []
if incoming_home:
config.rootDirs = [home]
index_everything()
return {"root_dirs": [home]}
# ---
for _dir in new_dirs:
children = get_child_dirs(_dir, db_dirs)
removed_dirs.extend(children)
for _dir in removed_dirs:
try:
db_dirs.remove(_dir)
except ValueError:
pass
db_dirs.extend(new_dirs)
config.rootDirs = [dir_ for dir_ in db_dirs if dir_ != home]
index_everything()
return {"root_dirs": config.rootDirs}
@api.get("/get-root-dirs")
def get_root_dirs():
"""
Get root directories
"""
return {"dirs": UserConfig().rootDirs}
@api.get("")
def get_all_settings():
"""
Get all settings
"""
config = asdict(UserConfig())
# Convert sets to lists for JSON serialization
for key, value in config.items():
if isinstance(value, set):
config[key] = sorted(list(value))
config["plugins"] = [p for p in PluginTable.get_all()]
config["version"] = metadata.version("swingmusic")
# only return lastfmSessionKey for the current user
current_user = get_current_userid()
config["lastfmSessionKey"] = config["lastfmSessionKeys"].get(str(current_user), "")
del config["lastfmSessionKeys"]
return config
class SetSettingBody(BaseModel):
key: str = Field(
description="The setting key",
example="artist_separators",
)
value: Any = Field(
description="The setting value",
example=",",
)
@api.get("/trigger-scan")
def trigger_scan():
"""
Triggers scan for new music
"""
index_everything()
return {"msg": "Scan triggered!"}
class UpdateConfigBody(BaseModel):
key: str = Field(
description="The setting key",
example="usersOnLogin",
)
value: Any = Field(
description="The setting value",
example=False,
)
@api.put("/update")
@admin_required()
def update_config(body: UpdateConfigBody):
"""
Update the config file
"""
config = UserConfig()
if body.key == "artistSeparators":
body.value = body.value.split(",")
setattr(config, body.key, body.value)
# INFO: Rebuild stores when these settings are updated
reset_stores_lists = {
"artistSeparators",
"artistSplitIgnoreList",
"removeProdBy",
"removeRemasterInfo",
"mergeAlbums",
"cleanAlbumTitle",
"showAlbumsAsSingles",
}
if body.key in reset_stores_lists:
index_everything()
return {
"msg": "Config updated!",
}
+330
View File
@@ -0,0 +1,330 @@
"""
Contains all the track routes.
"""
import os
from pathlib import Path
import tempfile
import time
from typing import Literal
from pydantic import BaseModel, Field
from flask_openapi3 import APIBlueprint, Tag
from swingmusic.api.apischemas import TrackHashSchema
from swingmusic.lib.transcoder import start_transcoding
from flask import request, Response, send_from_directory
from swingmusic.lib.trackslib import get_silence_paddings
from swingmusic.store.tracks import TrackStore
from swingmusic.utils.files import guess_mime_type
bp_tag = Tag(name="File", description="Audio files")
api = APIBlueprint("track", __name__, url_prefix="/file", abp_tags=[bp_tag])
class TransCodeStore:
map: dict[str, str] = {}
@classmethod
def add_file(cls, trackhash: str, filepath: str):
cls.map[trackhash] = filepath
@classmethod
def remove_file(cls, trackhash: str):
del cls.map[trackhash]
@classmethod
def find(cls, trackhash: str):
return cls.map.get(trackhash)
class SendTrackFileQuery(BaseModel):
filepath: str = Field(description="The filepath to play (if available)")
quality: str = Field(
"original",
description="The quality of the audio file. Options: original, 1411, 1024, 512, 320, 256, 128, 96",
)
container: Literal["mp3", "aac", "flac", "webm", "ogg"] = Field(
"mp3",
description="The container format of the audio file. Options: mp3, aac, flac, webm, ogg",
)
@api.get("/<trackhash>/legacy")
def send_track_file_legacy(path: TrackHashSchema, query: SendTrackFileQuery):
"""
Get a playable audio file without Range support
Returns a playable audio file that corresponds to the given filepath. Falls back to track hash if filepath is not found.
NOTE: Does not support range requests or transcoding.
"""
trackhash = path.trackhash
filepath = query.filepath
msg = {"msg": "File Not Found"}
track = None
tracks = TrackStore.get_tracks_by_filepaths([filepath])
if len(tracks) > 0 and os.path.exists(filepath):
track = tracks[0]
else:
res = TrackStore.trackhashmap.get(trackhash)
# When finding by trackhash, sort by bitrate
# and get the first track that exists
if res is not None:
tracks = sorted(res.tracks, key=lambda x: x.bitrate, reverse=True)
for t in tracks:
if os.path.exists(t.filepath):
track = t
break
if track is not None:
audio_type = guess_mime_type(filepath)
return send_from_directory(
Path(filepath).parent,
Path(filepath).name,
mimetype=audio_type,
conditional=True,
as_attachment=True,
)
return msg, 404
@api.get("/<trackhash>")
def send_track_file(path: TrackHashSchema, query: SendTrackFileQuery):
"""
Get a playable audio file with Range headers support
Returns a playable audio file that corresponds to the given filepath. Falls back to track hash if filepath is not found.
Transcoding can be done by sending the quality and container query parameters.
**NOTES:**
- Transcoded streams report incorrect duration during playback (idk why! FFMPEG gurus we need your help here).
- The quality parameter is the desired bitrate in kbps.
- The mp3 container is the best container for upto 320kbps (and has better duration reporting). The flac container allows for higher bitrates but it produces dramatically larger files (when transcoding from lossy formats).
- You can get the transcoded bitrate by checking the X-Transcoded-Bitrate header on the first request's response.
"""
trackhash = path.trackhash
filepath = query.filepath
# If filepath is provided, try to send that
track = None
tracks = TrackStore.get_tracks_by_filepaths([filepath])
if len(tracks) > 0 and os.path.exists(filepath):
track = tracks[0]
else:
res = TrackStore.trackhashmap.get(trackhash)
# When finding by trackhash, sort by bitrate
# and get the first track that exists
if res is not None:
tracks = sorted(res.tracks, key=lambda x: x.bitrate, reverse=True)
for t in tracks:
if os.path.exists(t.filepath):
track = t
break
if track is not None:
if query.quality == "original":
return send_file_as_chunks(track.filepath)
# prevent requesting over transcoding
max_bitrate = track.bitrate
requested_bitrate = int(query.quality)
if query.container != "flac":
# drop to 320 for non-flac containers
requested_bitrate = min(320, requested_bitrate)
quality = f"{min(max_bitrate, requested_bitrate)}k"
return transcode_and_stream(trackhash, track.filepath, quality, query.container)
return {"msg": "File Not Found"}, 404
def transcode_and_stream(trackhash: str, filepath: str, bitrate: str, container: str):
"""
Initiates transcoding and returns the first chunk of the transcoded file.
The other chunks are streamed on subsequent requests and are rerouted to `send_file_as_chunks`.
"""
temp_file = TransCodeStore.find(trackhash)
if temp_file is not None:
return send_file_as_chunks(temp_file)
format_params = {
"mp3": ["-c:a", "libmp3lame"],
"aac": ["-c:a", "aac"],
"webm": ["-c:a", "libopus"],
"ogg": ["-c:a", "libvorbis"],
"flac": ["-c:a", "flac"],
"wav": ["-c:a", "pcm_s16le"],
}
# Create a temporary file
format = f".{container}" if container in format_params.keys() else ".flac"
container_args = (
format_params[container]
if container in format_params.keys()
else format_params["flac"]
)
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=format)
temp_filename = temp_file.name
temp_file.close()
TransCodeStore.add_file(trackhash, temp_filename)
start_transcoding(filepath, temp_filename, bitrate, container_args)
chunk_size = 1024 * 512 # 0.5MB
file_size = os.path.getsize(filepath)
def generate():
# Poll for the output file
while (
not os.path.exists(temp_filename)
or os.path.getsize(temp_filename) < chunk_size
):
print(f"Waiting for transcoding to complete... filename: {temp_filename}")
time.sleep(0.1) # Wait for 100ms before checking again
with open(temp_filename, "rb") as file:
file.seek(0)
return file.read(chunk_size)
audio_type = guess_mime_type(temp_filename)
response = Response(
generate(),
206,
mimetype=audio_type,
content_type=audio_type,
direct_passthrough=True,
)
response.headers.add("Content-Range", f"bytes {0}-{chunk_size}/{file_size}")
response.headers.add("Accept-Ranges", "bytes")
response.headers.add("X-Transcoded-Bitrate", bitrate)
return response
def send_file_as_chunks(filepath: str) -> Response:
"""
Returns a Response object that streams the file in chunks.
"""
# NOTE: +1 makes sure the last byte is included in the range.
# NOTE: -1 is used to convert the end index to a 0-based index.
chunk_size = 1024 * 512 # 0.5MB
# Get file size
file_size = os.path.getsize(filepath)
start = 0
end = chunk_size
# Read range header
range_header = request.headers.get("Range")
if range_header:
start = get_start_range(range_header)
# If start + chunk_size is greater than file_size,
# set end to file_size - 1
_end = start + chunk_size - 1
if _end > file_size:
end = file_size - 1
else:
end = _end
def generate_chunks():
with open(filepath, "rb") as file:
file.seek(start)
remaining_bytes = end - start + 1
retry_count = 0
max_retries = 10 # 5 * 100ms = 500ms total wait time
while remaining_bytes > 0 or retry_count < max_retries:
if retry_count == max_retries:
print("💚 sending final chunk! ...")
pos = file.tell()
chunk = file.read(os.path.getsize(filepath) - pos)
return chunk, pos, True
if remaining_bytes < chunk_size:
time.sleep(0.25)
retry_count += 1
remaining_bytes = os.path.getsize(filepath) - file.tell()
continue
chunk = file.read(min(chunk_size, remaining_bytes))
if chunk:
remaining_bytes -= len(chunk)
return chunk, file.tell(), False
else:
# If no data is read, wait for 100ms before retrying
time.sleep(0.25)
retry_count += 1
# update remaining bytes
remaining_bytes = os.path.getsize(filepath) - file.tell()
print(f"▶ Remaining bytes: {remaining_bytes}")
return None, 0, True
data, position, is_final = generate_chunks()
audio_type = guess_mime_type(filepath)
response = Response(
response=data,
status=206, # Partial Content status code
mimetype=audio_type,
content_type=audio_type,
direct_passthrough=True,
)
bytes_to_add = chunk_size if not is_final else 0
response.headers.add(
"Content-Range",
f"bytes {start}-{position}/{os.path.getsize(filepath) + bytes_to_add}",
)
response.headers.add("Access-Control-Expose-Headers", "Content-Range")
response.headers.add("Accept-Ranges", "bytes")
return response
def get_start_range(range_header: str):
try:
range_start, range_end = range_header.strip().split("=")[1].split("-")
return int(range_start)
except ValueError:
return 0
class GetAudioSilenceBody(BaseModel):
ending_file: str = Field(description="The ending file's path")
starting_file: str = Field(description="The beginning file's path")
@api.post("/silence")
def get_audio_silence(body: GetAudioSilenceBody):
"""
Get silence paddings
Returns the duration of silence at the end of the current ending track and the duration of silence at the beginning of the next track.
NOTE: Durations are in milliseconds.
"""
ending_file = body.ending_file # ending file's filepath
starting_file = body.starting_file # starting file's filepath
if ending_file is None or starting_file is None:
return {"msg": "No filepath provided"}, 400
return get_silence_paddings(ending_file, starting_file)