Automate your music collection https://blog.platypush.tech/article/Automate-your-music-collection

[sudo] pip install 'platypush[tidal,rss,lastfm]'

music.tidal:
  # No configuration required

# add its credentials here
# music.spotify:
#   client_id: client_id
#   client_secret: client_secret

lastfm:
  api_key: your_api_key
  api_secret: your_api_secret
  username: your_user
  password: your_password

# Subscribe to updates from newalbumreleases.net
rss:
  subscriptions:
    - https://newalbumreleases.net/category/cat/feed/

# Optional, used to send notifications about generation issues to your
# mobile/browser. You can also use Pushbullet, an email plugin or a chatbot if
# you prefer.
ntfy:
  # No configuration required if you want to use the default server at
  # https://ntfy.sh

# Include the mpd plugin and backend if you are listening to music over
# mpd/mopidy
music.mpd:
  host: localhost
  port: 6600

backend.music.mopidy:
  host: localhost
  port: 6600

-- New listened tracks will be pushed to the tmp_music table, and normalized by
-- a trigger.
drop table if exists tmp_music cascade;
create table tmp_music(
    id          serial not null,
    artist      varchar(255) not null,
    title       varchar(255) not null,
    album       varchar(255),
    created_at  timestamp with time zone default CURRENT_TIMESTAMP,
    primary key(id)
);

-- This table will store the tracks' info
drop table if exists music_track cascade;
create table music_track(
    id          serial not null,
    artist      varchar(255) not null,
    title       varchar(255) not null,
    album       varchar(255),
    created_at  timestamp with time zone default CURRENT_TIMESTAMP,
    primary key(id),
    unique(artist, title)
);

-- Create an index on (artist, title), and ensure that the (artist, title) pair
-- is unique
create unique index track_artist_title_idx on music_track(lower(artist), lower(title));
create index track_artist_idx on music_track(lower(artist));

-- music_activity holds the listened tracks
drop table if exists music_activity cascade;
create table music_activity(
    id          serial not null,
    track_id    int not null,
    created_at  timestamp with time zone default CURRENT_TIMESTAMP,
    primary key(id)
);

-- music_similar keeps track of the similar tracks
drop table if exists music_similar cascade;
create table music_similar(
    source_track_id   int not null,
    target_track_id   int not null,
    match_score       float not null,
    primary key(source_track_id, target_track_id),
    foreign key(source_track_id) references music_track(id),
    foreign key(target_track_id) references music_track(id)
);

-- music_discovery_playlist keeps track of the generated discovery playlists
drop table if exists music_discovery_playlist cascade;
create table music_discovery_playlist(
    id          serial not null,
    name        varchar(255),
    created_at  timestamp with time zone default CURRENT_TIMESTAMP,
    primary key(id)
);

-- This table contains the track included in each discovery playlist
drop table if exists music_discovery_playlist_track cascade;
create table music_discovery_playlist_track(
    id            serial not null,
    playlist_id   int not null,
    track_id      int not null,
    primary key(id),
    unique(playlist_id, track_id),
    foreign key(playlist_id) references music_discovery_playlist(id),
    foreign key(track_id) references music_track(id)
);

-- This table contains the new releases from artist that we've listened to at
-- least once
drop table if exists new_release cascade;
create table new_release(
    id serial not null,
    artist varchar(255) not null,
    album varchar(255) not null,
    genre varchar(255),
    created_at timestamp with time zone default CURRENT_TIMESTAMP,

    primary key(id),
    constraint u_artist_title unique(artist, album)
);

-- This trigger normalizes the tracks inserted into tmp_track
create or replace function sync_music_data()
    returns trigger as
$$
declare
    track_id int;
begin
    insert into music_track(artist, title, album)
        values(new.artist, new.title, new.album)
    on conflict(artist, title) do update
        set album = coalesce(excluded.album, old.album)
    returning id into track_id;

    insert into music_activity(track_id, created_at)
        values (track_id, new.created_at);

    delete from tmp_music where id = new.id;
    return new;
end;
$$
language 'plpgsql';

drop trigger if exists on_sync_music on tmp_music;
create trigger on_sync_music
    after insert on tmp_music
    for each row
    execute procedure sync_music_data();

-- (Optional) accessory view to easily peek the listened tracks
drop view if exists vmusic;
create view vmusic as
select t.id as track_id
     , t.artist
     , t.title
     , t.album
     , a.created_at
from music_track t
join music_activity a
on t.id = a.track_id;

# ~/.config/platypush/scripts/music/sync.py

from logging import getLogger

from platypush.context import get_plugin
from platypush.event.hook import hook
from platypush.message.event.music import NewPlayingTrackEvent

logger = getLogger('music_sync')

# SQLAlchemy connection string that points to your database
music_db_engine = 'postgresql+pg8000://dbuser:dbpass@dbhost/dbname'


# Hook that react to NewPlayingTrackEvent events
@hook(NewPlayingTrackEvent)
def on_new_track_playing(event, **_):
    track = event.track

    # Skip if the track has no artist/title specified
    if not (track.get('artist') and track.get('title')):
        return

    logger.info(
        'Inserting track: %s - %s',
        track['artist'], track['title']
    )

    db = get_plugin('db')
    db.insert(
        engine=music_db_engine,
        table='tmp_music',
        records=[
            {
                'artist': track['artist'],
                'title': track['title'],
                'album': track.get('album'),
            }
            for track in tracks
        ]
    )

# ~/.config/platypush/scripts/music/sync.py

from logging import getLogger

from platypush.context import get_plugin
from platypush.event.hook import hook
from platypush.message.event.music import NewPlayingTrackEvent

logger = getLogger('music_sync')


# Hook that react to NewPlayingTrackEvent events
@hook(NewPlayingTrackEvent)
def on_new_track_playing(event, **_):
    track = event.track

    # Skip if the track has no artist/title specified
    if not (track.get('artist') and track.get('title')):
        return

    lastfm = get_plugin('lastfm')
    logger.info(
        'Scrobbling track: %s - %s',
        track['artist'], track['title']
    )

    lastfm.scrobble(
        artist=track['artist'],
        title=track['title'],
        album=track.get('album'),
    )

# ~/.config/platypush/scripts/music/scrobble2db.py

import logging

from datetime import datetime

from platypush.context import get_plugin, Variable
from platypush.cron import cron

logger = logging.getLogger('music_sync')
music_db_engine = 'postgresql+pg8000://dbuser:dbpass@dbhost/dbname'

# Use this stored variable to keep track of the time of the latest
# synchronized scrobble
last_timestamp_var = Variable('LAST_SCROBBLED_TIMESTAMP')


# This cron executes every 30 seconds
@cron('* * * * * */30')
def sync_scrobbled_tracks(**_):
    db = get_plugin('db')
    lastfm = get_plugin('lastfm')

    # Use the last.fm plugin to retrieve all the new tracks scrobbled since
    # the last check
    last_timestamp = int(last_timestamp_var.get() or 0)
    tracks = [
        track for track in lastfm.get_recent_tracks().output
        if track.get('timestamp', 0) > last_timestamp
    ]

    # Exit if we have no new music activity
    if not tracks:
        return

    # Insert the new tracks on the database
    db.insert(
        engine=music_db_engine,
        table='tmp_music',
        records=[
            {
                'artist': track.get('artist'),
                'title': track.get('title'),
                'album': track.get('album'),
                'created_at': (
                    datetime.fromtimestamp(track['timestamp'])
                    if track.get('timestamp') else None
                ),
            }
            for track in tracks
        ]
    )

    # Update the LAST_SCROBBLED_TIMESTAMP variable with the timestamp of the
    # most recent played track
    last_timestamp_var.set(max(
        int(t.get('timestamp', 0))
        for t in tracks
    ))

    logger.info('Stored %d new scrobbled track(s)', len(tracks))

# ~/.config/platypush/scripts/music/db.py

from sqlalchemy import create_engine
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import sessionmaker, scoped_session

music_db_engine = 'postgresql+pg8000://dbuser:dbpass@dbhost/dbname'
engine = create_engine(music_db_engine)

Base = automap_base()
Base.prepare(engine, reflect=True)
Track = Base.classes.music_track
TrackActivity = Base.classes.music_activity
TrackSimilar = Base.classes.music_similar
DiscoveryPlaylist = Base.classes.music_discovery_playlist
DiscoveryPlaylistTrack = Base.classes.music_discovery_playlist_track
NewRelease = Base.classes.new_release


def get_db_session():
    session = scoped_session(sessionmaker(expire_on_commit=False))
    session.configure(bind=engine)
    return session()

# ~/.config/platypush/scripts/music/suggestions.py

import logging

from sqlalchemy import tuple_
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.sql.expression import bindparam

from platypush.context import get_plugin, Variable
from platypush.cron import cron

from scripts.music.db import (
    get_db_session, Track, TrackActivity, TrackSimilar
)


logger = logging.getLogger('music_suggestions')

# This stored variable will keep track of the latest activity ID for which the
# suggestions were calculated
last_activity_id_var = Variable('LAST_PROCESSED_ACTIVITY_ID')


# A cronjob that runs every 5 minutes and updates the suggestions
@cron('*/5 * * * *')
def refresh_similar_tracks(**_):
    last_activity_id = int(last_activity_id_var.get() or 0)

    # Retrieve all the tracks played since the latest synchronized activity ID
    # that don't have any similar tracks being calculated yet
    with get_db_session() as session:
        recent_tracks_without_similars = \
            _get_recent_tracks_without_similars(last_activity_id)

    try:
        if not recent_tracks_without_similars:
            raise StopIteration(
                'All the recent tracks have processed suggestions')

        # Get the last activity_id
        batch_size = 10
        last_activity_id = (
            recent_tracks_without_similars[:batch_size][-1]['activity_id'])

        logger.info(
            'Processing suggestions for %d/%d tracks',
            min(batch_size, len(recent_tracks_without_similars)),
            len(recent_tracks_without_similars))

        # Build the track_id -> [similar_tracks] map
        similars_by_track = {
            track['track_id']: _get_similar_tracks(track['artist'], track['title'])
            for track in recent_tracks_without_similars[:batch_size]
        }

        # Map all the similar tracks in an (artist, title) -> info data structure
        similar_tracks_by_artist_and_title = \
            _get_similar_tracks_by_artist_and_title(similars_by_track)

        if not similar_tracks_by_artist_and_title:
            raise StopIteration('No new suggestions to process')

        # Sync all the new similar tracks to the database
        similar_tracks = \
            _sync_missing_similar_tracks(similar_tracks_by_artist_and_title)

        # Link listened tracks to similar tracks
        with get_db_session() as session:
            stmt = insert(TrackSimilar).values({
                'source_track_id': bindparam('source_track_id'),
                'target_track_id': bindparam('target_track_id'),
                'match_score': bindparam('match_score'),
            }).on_conflict_do_nothing()

            session.execute(
                stmt, [
                    {
                        'source_track_id': track_id,
                        'target_track_id': similar_tracks[(similar['artist'], similar['title'])].id,
                        'match_score': similar['score'],
                    }
                    for track_id, similars in similars_by_track.items()
                    for similar in (similars or [])
                    if (similar['artist'], similar['title'])
                    in similar_tracks
                ]
            )

            session.flush()
            session.commit()
    except StopIteration as e:
        logger.info(e)

    last_activity_id_var.set(last_activity_id)
    logger.info('Suggestions updated')


def _get_similar_tracks(artist, title):
    """
    Use the last.fm API to retrieve the tracks similar to a given
    artist/title pair
    """
    import pylast
    lastfm = get_plugin('lastfm')

    try:
        return lastfm.get_similar_tracks(
            artist=artist,
            title=title,
            limit=10,
        )
    except pylast.PyLastError as e:
        logger.warning(
            'Could not find tracks similar to %s - %s: %s',
            artist, title, e
        )


def _get_recent_tracks_without_similars(last_activity_id):
    """
    Get all the tracks played after a certain activity ID that don't have
    any suggestions yet.
    """
    with get_db_session() as session:
        return [
            {
                'track_id': t[0],
                'artist': t[1],
                'title': t[2],
                'activity_id': t[3],
            }
            for t in session.query(
                Track.id.label('track_id'),
                Track.artist,
                Track.title,
                TrackActivity.id.label('activity_id'),
            )
            .select_from(
                Track.__table__
                .join(
                    TrackSimilar,
                    Track.id == TrackSimilar.source_track_id,
                    isouter=True
                )
                .join(
                    TrackActivity,
                    Track.id == TrackActivity.track_id
                )
            )
            .filter(
                TrackSimilar.source_track_id.is_(None),
                TrackActivity.id > last_activity_id
            )
            .order_by(TrackActivity.id)
            .all()
        ]


def _get_similar_tracks_by_artist_and_title(similars_by_track):
    """
    Map similar tracks into an (artist, title) -> track dictionary
    """
    similar_tracks_by_artist_and_title = {}
    for similar in similars_by_track.values():
        for track in (similar or []):
            similar_tracks_by_artist_and_title[
                (track['artist'], track['title'])
            ] = track

    return similar_tracks_by_artist_and_title


def _sync_missing_similar_tracks(similar_tracks_by_artist_and_title):
    """
    Flush newly calculated similar tracks to the database.
    """
    logger.info('Syncing missing similar tracks')
    with get_db_session() as session:
        stmt = insert(Track).values({
            'artist': bindparam('artist'),
            'title': bindparam('title'),
        }).on_conflict_do_nothing()

        session.execute(stmt, list(similar_tracks_by_artist_and_title.values()))
        session.flush()
        session.commit()

        tracks = session.query(Track).filter(
            tuple_(Track.artist, Track.title).in_(
                similar_tracks_by_artist_and_title
            )
        ).all()

        return {
            (track.artist, track.title): track
            for track in tracks
        }

# ~/.config/platypush/scripts/music/discovery.py

import logging
from datetime import date, timedelta

from platypush.context import get_plugin
from platypush.cron import cron

from scripts.music.db import (
    get_db_session, Track, TrackActivity, TrackSimilar,
    DiscoveryPlaylist, DiscoveryPlaylistTrack
)

logger = logging.getLogger('music_discovery')


def get_suggested_tracks(days=7, limit=25):
    """
    Retrieve the suggested tracks from the database.

    :param days: Look back at the listen history for the past <n> days
        (default: 7).
    :param limit: Maximum number of track in the discovery playlist
        (default: 25).
    """
    from sqlalchemy import func

    listened_activity = TrackActivity.__table__.alias('listened_activity')
    suggested_activity = TrackActivity.__table__.alias('suggested_activity')

    with get_db_session() as session:
        return [
            {
                'track_id': t[0],
                'artist': t[1],
                'title': t[2],
                'score': t[3],
            }
            for t in session.query(
                Track.id,
                func.min(Track.artist),
                func.min(Track.title),
                func.sum(TrackSimilar.match_score).label('score'),
            )
            .select_from(
                Track.__table__
                .join(
                    TrackSimilar.__table__,
                    Track.id == TrackSimilar.target_track_id
                )
                .join(
                    listened_activity,
                    listened_activity.c.track_id == TrackSimilar.source_track_id,
                )
                .join(
                    suggested_activity,
                    suggested_activity.c.track_id == TrackSimilar.target_track_id,
                    isouter=True
                )
                .join(
                    DiscoveryPlaylistTrack,
                    Track.id == DiscoveryPlaylistTrack.track_id,
                    isouter=True
                )
            )
            .filter(
                # The track has not been listened
                suggested_activity.c.track_id.is_(None),
                # The track has not been suggested already
                DiscoveryPlaylistTrack.track_id.is_(None),
                # Filter by recent activity
                listened_activity.c.created_at >= date.today() - timedelta(days=days)
            )
            .group_by(Track.id)
            # Sort by aggregate match score
            .order_by(func.sum(TrackSimilar.match_score).desc())
            .limit(limit)
            .all()
        ]


def search_remote_tracks(tracks):
    """
    Search for Tidal tracks given a list of suggested tracks.
    """
    # If you use Spotify instead of Tidal, simply replacing `music.tidal`
    # with `music.spotify` here should suffice.
    tidal = get_plugin('music.tidal')
    found_tracks = []

    for track in tracks:
        query = track['artist'] + ' ' + track['title']
        logger.info('Searching "%s"', query)
        results = (
            tidal.search(query, type='track', limit=1).output.get('tracks', [])
        )

        if results:
            track['remote_track_id'] = results[0]['id']
            found_tracks.append(track)
        else:
            logger.warning('Could not find "%s" on TIDAL', query)

    return found_tracks


def refresh_discover_weekly():
    # If you use Spotify instead of Tidal, simply replacing `music.tidal`
    # with `music.spotify` here should suffice.
    tidal = get_plugin('music.tidal')

    # Get the latest suggested tracks
    suggestions = search_remote_tracks(get_suggested_tracks())
    if not suggestions:
        logger.info('No suggestions available')
        return

    # Retrieve the existing discovery playlists
    # Our naming convention is that discovery playlist names start with
    # "Discover Weekly" - feel free to change it
    playlists = tidal.get_playlists().output
    discover_playlists = sorted(
        [
            pl for pl in playlists
            if pl['name'].lower().startswith('discover weekly')
        ],
        key=lambda pl: pl.get('created_at', 0)
    )

    # Delete all the existing discovery playlists
    # (except the latest one). We basically keep two discovery playlists at the
    # time in our collection, so you have two weeks to listen to them before they
    # get deleted. Feel free to change this logic by modifying the -1 parameter
    # with e.g. -2, -3 etc. if you want to store more discovery playlists.
    for playlist in discover_playlists[:-1]:
        logger.info('Deleting playlist "%s"', playlist['name'])
        tidal.delete_playlist(playlist['id'])

    # Create a new discovery playlist
    playlist_name = f'Discover Weekly [{date.today().isoformat()}]'
    pl = tidal.create_playlist(playlist_name).output
    playlist_id = pl['id']

    tidal.add_to_playlist(
        playlist_id,
        [t['remote_track_id'] for t in suggestions],
    )

    # Add the playlist to the database
    with get_db_session() as session:
        pl = DiscoveryPlaylist(name=playlist_name)
        session.add(pl)
        session.flush()
        session.commit()

    # Add the playlist entries to the database
    with get_db_session() as session:
        for track in suggestions:
            session.add(
                DiscoveryPlaylistTrack(
                    playlist_id=pl.id,
                    track_id=track['track_id'],
                )
            )

        session.commit()

    logger.info('Discover Weekly playlist updated')


@cron('0 6 * * 1')
def refresh_discover_weekly_cron(**_):
    """
    This cronjob runs every Monday at 6 AM.
    """
    try:
        refresh_discover_weekly()
    except Exception as e:
        logger.exception(e)

        # (Optional) If anything went wrong with the playlist generation, send
        # a notification over ntfy
        ntfy = get_plugin('ntfy')
        ntfy.send_message(
            topic='mirrored-notifications-topic',
            title='Discover Weekly playlist generation failed',
            message=str(e),
            priority=4,
        )

>>> import os
>>>
>>> # Move to the Platypush config directory
>>> path = os.path.join(os.path.expanduser('~'), '.config', 'platypush')
>>> os.chdir(path)
>>>
>>> # Import and run the cron function
>>> from scripts.music.discovery import refresh_discover_weekly_cron
>>> refresh_discover_weekly_cron()

# ~/.config/platypush/scripts/music/releases.py

import html
import logging
import re
import threading
from datetime import date, timedelta
from typing import Iterable, List

from platypush.context import get_plugin
from platypush.cron import cron
from platypush.event.hook import hook
from platypush.message.event.rss import NewFeedEntryEvent

from scripts.music.db import (
    music_db_engine, get_db_session, NewRelease
)


create_lock = threading.RLock()
logger = logging.getLogger(__name__)


def _split_html_lines(content: str) -> List[str]:
    """
    Utility method used to convert and split the HTML lines reported
    by the RSS feed.
    """
    return [
        l.strip()
        for l in re.sub(
            r'(</?p[^>]*>)|(<br\s*/?>)',
            '\n',
            content
        ).split('\n') if l
    ]


def _get_summary_field(title: str, lines: Iterable[str]) -> str | None:
    """
    Parse the fields of a new album from the feed HTML summary.
    """
    for line in lines:
        m = re.match(rf'^{title}:\s+(.*)$', line.strip(), re.IGNORECASE)
        if m:
            return html.unescape(m.group(1))


@hook(NewFeedEntryEvent, feed_url='https://newalbumreleases.net/category/cat/feed/')
def save_new_release(event: NewFeedEntryEvent, **_):
    """
    This hook is triggered whenever the newalbumreleases.net has new entries.
    """
    # Parse artist and album
    summary = _split_html_lines(event.summary)
    artist = _get_summary_field('artist', summary)
    album = _get_summary_field('album', summary)
    genre = _get_summary_field('style', summary)

    if not (artist and album):
        return

    # Check if we have listened to this artist at least once
    db = get_plugin('db')
    num_plays = int(
        db.select(
            engine=music_db_engine,
            query=
            '''
            select count(*)
            from music_activity a
            join music_track t
            on a.track_id = t.id
            where artist = :artist
            ''',
            data={'artist': artist},
        ).output[0].get('count', 0)
    )

    # If not, skip it
    if not num_plays:
        return

    # Insert the new release on the database
    with create_lock:
        db.insert(
            engine=music_db_engine,
            table='new_release',
            records=[{
                'artist': artist,
                'album': album,
                'genre': genre,
            }],
            key_columns=('artist', 'album'),
            on_duplicate_update=True,
        )


def get_new_releases(days=7):
    """
    Retrieve the new album releases from the database.

    :param days: Look at albums releases in the past <n> days
        (default: 7)
    """
    with get_db_session() as session:
        return [
            {
                'artist': t[0],
                'album': t[1],
            }
            for t in session.query(
                NewRelease.artist,
                NewRelease.album,
            )
            .select_from(
                NewRelease.__table__
            )
            .filter(
                # Filter by recent activity
                NewRelease.created_at >= date.today() - timedelta(days=days)
            )
            .all()
        ]


def search_tidal_new_releases(albums):
    """
    Search for Tidal albums given a list of objects with artist and title.
    """
    tidal = get_plugin('music.tidal')
    expanded_tracks = []

    for album in albums:
        query = album['artist'] + ' ' + album['album']
        logger.info('Searching "%s"', query)
        results = (
            tidal.search(query, type='album', limit=1)
            .output.get('albums', [])
        )

        if results:
            album = results[0]

            # Skip search results older than a year - some new releases may
            # actually be remasters/re-releases of existing albums
            if date.today().year - album.get('year', 0) > 1:
                continue

            expanded_tracks += (
                tidal.get_album(results[0]['id']).
                output.get('tracks', [])
            )
        else:
            logger.warning('Could not find "%s" on TIDAL', query)

    return expanded_tracks


def refresh_release_radar():
    tidal = get_plugin('music.tidal')

    # Get the latest releases
    tracks = search_tidal_new_releases(get_new_releases())
    if not tracks:
        logger.info('No new releases found')
        return

    # Retrieve the existing new releases playlists
    playlists = tidal.get_playlists().output
    new_releases_playlists = sorted(
        [
            pl for pl in playlists
            if pl['name'].lower().startswith('new releases')
        ],
        key=lambda pl: pl.get('created_at', 0)
    )

    # Delete all the existing new releases playlists
    # (except the latest one)
    for playlist in new_releases_playlists[:-1]:
        logger.info('Deleting playlist "%s"', playlist['name'])
        tidal.delete_playlist(playlist['id'])

    # Create a new releases playlist
    playlist_name = f'New Releases [{date.today().isoformat()}]'
    pl = tidal.create_playlist(playlist_name).output
    playlist_id = pl['id']

    tidal.add_to_playlist(
        playlist_id,
        [t['id'] for t in tracks],
    )


@cron('0 7 * * 1')
def refresh_release_radar_cron(**_):
    """
    This cron will execute every Monday at 7 AM.
    """
    try:
        refresh_release_radar()
    except Exception as e:
        logger.exception(e)
        get_plugin('ntfy').send_message(
            topic='mirrored-notifications-topic',
            title='Release Radar playlist generation failed',
            message=str(e),
            priority=4,
        )