sync_playlists.py

#!/usr/bin/env python

import sys
import os
import sqlite3

def db_get_single_value(db, query, arguments):
    db.execute(query, arguments)
    rows = db.fetchall()
    if len(rows) != 1 or len(rows[0]) < 1:
        return None
    return rows[0][0]

def read_playlist(playlist_path, absolute=True):
    tracks = []
    for line in open(playlist_path).readlines():
        line = line.strip()
        if not line:
            continue
        if absolute and not os.path.isabs(line):
            line = os.path.join(os.path.dirname(playlist_path), line)
        tracks.append(os.path.normpath(line))
    return tracks

def sync_playlist(db, playlist_path):
    track_num = 0
    playlist_name = os.path.basename(playlist_path).replace(".m3u", "")
    extra_playlist = os.path.join(os.path.dirname(playlist_path), playlist_name + "Extra.m3u")
    playlist_id = db_get_single_value(db, "select id from playlists where name=?", (playlist_name,))
    if playlist_id is None:
        print("# Warning: Playlist not found: %s" % playlist_name)
        return track_num

    playlist_tracks = []
    tracks = read_playlist(playlist_path)
    if os.path.exists(extra_playlist):
        tracks.extend(read_playlist(extra_playlist))
    for position, track in enumerate(tracks):
        location_id = db_get_single_value(db, "select id from track_locations where location=?", (track,))
        if location_id is None:
            print("# Warning: Track location not found: %s" % track)
            continue
        track_id = db_get_single_value(db, "select id from library where location=?", (location_id,))
        if track_id is None:
            print("# Warning: Track not found: %s" % track)
            continue
        track_num += 1
        playlist_tracks.append((playlist_id, track_id, position))

    db.execute("delete from playlisttracks where playlist_id=?", (playlist_id,))
    db.executemany("insert into playlisttracks (playlist_id, track_id, position) values(?, ?, ?)", playlist_tracks)
    return track_num

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: %s [playlists...]" % sys.argv[0])
        sys.exit(1)
    
    db_path = os.path.expanduser("~/.mixxx/mixxxdb.sqlite")
    if not os.path.exists(db_path):
        print("Error: Unable to find mixxx sqlite database!")
        sys.exit(1)

    conn = sqlite3.connect(db_path)
    db = conn.cursor()
    track_num = 0
    for playlist in sys.argv[1:]:
        track_num += sync_playlist(db, os.path.abspath(playlist))
    print("Added %d tracks." % track_num)
    conn.commit()
    conn.close()