script that updates track ratings in banshee database from beet favorite songs

banshee-beets-sync.py

#! /usr/bin/env python
'''
Script that syncs beet favorites with banshee.
'''

import click
from beets import library


import os
import subprocess as sp
from subprocess import check_output as co
from pathlib import Path
import sqlite3
from urllib.parse import quote, unquote, urlparse
import textwrap
import yaml
import shutil
import shlex


BEET = ['beet']
FIND = ['ls', '-p']
MOD = ['modify', '--yes']
CONFIG = ['config']

BEET_FAVORITE = "favorite:true"
BEET_FAVORITE_MODIFY = "favorite=true"
DBPATH = '~/.config/banshee-1/banshee.db'


COLUMNS, LINES = shutil.get_terminal_size(fallback=(75, 22))


def wrap(text, item_prefix="", indent=0, width=COLUMNS, *args, **kwargs):
    prefix = "{}{}".format(" " * indent, item_prefix)

    wrapper = textwrap.TextWrapper(
        initial_indent=prefix,
        subsequent_indent=" " * len(prefix),
        width=width)

    click.echo(wrapper.fill(text), *args, **kwargs)


def beet_config():
    config = co(BEET + CONFIG, encoding='utf-8')
    return yaml.load(config)


def get_beet_favorites():
    favs = co(BEET + FIND + [BEET_FAVORITE], encoding='utf-8')
    return favs.splitlines()


def set_beet_favorites(music_folder, favorites):
    if not favorites:
        click.echo("No modification to beet.")
        return
    else: click.echo("Adding %d songs to beet favorites:" % len(favorites))

    for path, uri in favorites:
        wrap('Adding "%s"' % path.relative_to(music_folder),
             item_prefix="- ",
             indent=2)
        co(BEET + MOD + [BEET_FAVORITE_MODIFY] + ["path:%s" % path])


def get_banshee_favorites():
    conn = sqlite3.connect(str(DBPATH))
    c = conn.cursor()

    c.execute('SELECT URI FROM CoreTracks WHERE Rating=5')
    items = [i[0] for i in c.fetchall()]
    conn.close()

    return items


def set_banshee_ratings(music_folder, favorites):
    if not favorites:
        click.echo("No modification to banshee.")
        return
    else:
        click.echo("Setting ratings of %d songs in banshee:" % len(favorites))

    conn = sqlite3.connect(str(DBPATH))
    c = conn.cursor()

    for path, uri in favorites:
        c.execute('UPDATE CoreTracks SET Rating=5 WHERE URI = ?', (uri,))
        if c.rowcount == 0:
            wrap(
                "{}: {} couldn't be found in banshee database".format(
                    click.style('Warning', fg='red'),
                    path),
                indent=2,
                item_prefix="",
                err=True)
        else:
            wrap('Changing "{}" rating to {}'.format(
                path.relative_to(music_folder), 5),
                item_prefix="- ",
                indent=2)

    conn.commit()
    conn.close()


@click.command()
@click.option('--beet-favorite', default=BEET_FAVORITE,
              help='beet query for favorites (default "%s").' % BEET_FAVORITE)
@click.option('--beet-favorite-modify', default=BEET_FAVORITE_MODIFY,
              help="beet modify key=value pair to set favorites "
                   '(default "%s").' % BEET_FAVORITE_MODIFY)
@click.option('--banshee-database', default=DBPATH,
              help='banshee database file (default "%s").' % DBPATH)
@click.pass_context
def main(ctx, beet_favorite, beet_favorite_modify, banshee_database):
    """
    Syncs banshee and beet favorites.
    """
    global BEET_FAVORITE, BEET_FAVORITE_MODIFY, DBPATH

    BEET_FAVORITE = beet_favorite
    BEET_FAVORITE_MODIFY = beet_favorite_modify
    DBPATH = Path(banshee_database).expanduser().absolute()

    try:
        conf = beet_config()
        beet_favs_path = get_beet_favorites()
        banshee_favs_uri = get_banshee_favorites()

        # banshee uses esoteric uri paths
        def as_uri(path): return "file://" + quote(path, safe="/()&")

        def from_uri(uri): return unquote(urlparse(uri).path)

        banshee_favs = set(
            (Path(from_uri(uri)), uri) for uri in banshee_favs_uri)
        beet_favs = set((Path(path), as_uri(path)) for path in beet_favs_path)

        add_to_banshee = beet_favs - banshee_favs
        add_to_beet = banshee_favs - beet_favs

        music_folder = Path(conf['directory']).expanduser()

        set_banshee_ratings(music_folder, add_to_banshee)
        set_beet_favorites(music_folder, add_to_beet)

    except sp.CalledProcessError as e:
        message = "Exit status {} with command: {}\n".format(
            click.style(str(e.returncode), bold=True),
            click.style(
                " ".join(shlex.quote(s) for s in e.cmd),
                bold=True))

        click.echo(message, err=True)
        ctx.abort()


if __name__ == "__main__":
    main()