I can't edit the original post, so here's a new version that properly handles accented characters in folders/files so that they don't perpetually re-copy. (For example, "Chopin, Frédéric".) Previous disclaimers and notes apply.
Separately, BTRamp is okay but tends to crash after a few hours, and won't shuffle more than 200 songs at a time. If anyone has iOS app suggestions that focus on locally stored songs, support CarPlay, and shuffle songs matching a detected genre ID3 tag, I'd love to hear about alternatives.
import logging
import sqlite3
from os import makedirs, remove, rmdir, walk
from os.path import getmtime
from pathlib import Path
from shutil import copyfile
from subprocess import call
from unicodedata import normalize
from urllib.parse import unquote, urlparse
def error_out(error_msg, db_conn=None):
'''
Print a message, close the Strawberry DB if open, exit with an error.
'''
logging.error(error_msg)
if db_conn:
db_conn.close()
exit(1)
# The iOS identifier for your music app
ios_appname = 'com.btrlabs.btramp'
# Names of files your music app has internally that you DON'T want deleted
skip_files = ['PlayerLog.log']
# Names of playlists you DON'T want to sync
playlists_to_ignore = ['Unlistened', 'Test']
# The base folder where your collection is located
src_folder = '/mnt/music'
# The folder where your iPhone will be mounted
dst_folder = '/mnt/iphone'
# Your Strawberry SQLite DB location, relative to $HOME
music_db_file = '.local/share/strawberry/strawberry/strawberry.db'
# Adjust verbosity up to 'WARNING'/'ERROR' or down to 'DEBUG'
logging.basicConfig(format='%(message)s', level=logging.INFO)
logging.debug(f'Connecting to DB at $HOME/{music_db_file} ...')
try:
home_folder_path = str(Path.home())
db = sqlite3.connect(home_folder_path + '/' + music_db_file)
c = db.cursor()
except Exception as e:
error_out(f'Unable to connect to Strawberry DB: {e}', db)
logging.info('Building queue from DB ...')
try:
sync_queue = {}
raw_playlist_ids = c.execute(
"""
SELECT ROWID
FROM playlists
WHERE name NOT IN ({})
""".format(','.join('?' * len(playlists_to_ignore))),
tuple(playlists_to_ignore)
).fetchall()
playlist_ids = [i[0] for i in raw_playlist_ids]
logging.debug(f"\tFound {len(playlist_ids)} playlists.")
for playlist_id in playlist_ids:
playlist_songs = c.execute(
"""
SELECT
songs.url,
songs.mtime
FROM songs
LEFT JOIN playlist_items
ON songs.rowid = playlist_items.collection_id
WHERE playlist_items.playlist = ?
""",
(playlist_id,)
).fetchall()
# Add song to queue, and also convert URLs to relative paths, e.g.
# file:///path/to/artist/album/song.mp3 to artist/album/song.mp3
for song in playlist_songs:
sync_queue[str(Path(unquote(urlparse(song[0]).path)).relative_to('/mnt/music'))] = song[1]
logging.debug(f"\tAdded {len(playlist_songs)} songs from playlist #{playlist_id}.")
db.close()
logging.info(f"\tFound {str(len(sync_queue))} files for sync.")
except Exception as e:
error_out(f'Unable to build queue: {e}', db)
logging.debug('Checking iPhone mount status ...')
try:
if not Path(dst_folder).exists():
logging.debug(f' Creating {dst_folder} ...')
makedirs(dst_folder)
if not Path(dst_folder).is_mount():
logging.debug(f' Mounting iPhone at {dst_folder}.')
ret = call(f"ifuse --documents {ios_appname} {dst_folder}", shell=True)
if ret != 0:
raise Exception(f'ifuse returned {ret}')
else:
logging.debug(f"\tiPhone is already mounted at {dst_folder}.")
except Exception as e:
error_out(f'Unable to mount iPhone: {e}')
logging.info('Syncing files ...')
try:
unchanged = overwritten = removed = 0
for (path, folders, files) in walk(dst_folder):
for file in files:
# Ignore music player stuff
if file in skip_files:
continue
# We have to convert accented characters from NFC (Linux) to NFD (Mac)
# as we compare and store each file + folder
dst_path = Path(normalize('NFD', path + '/' + file))
relative_path = str(
Path(normalize('NFC', path) + '/' + normalize('NFC', file))
.relative_to(normalize('NFC', dst_folder))
)
# Same filename exists in src + dst: recopy if newer
if relative_path in sync_queue:
if sync_queue[relative_path] > getmtime(dst_path):
if Path(src_folder + '/' + relative_path).exists():
logging.warning(f"\tOverwriting older file: {relative_path}")
overwritten += 1
copyfile(Path(src_folder + '/' + relative_path), dst_path)
else:
raise Exception(f'Unable to read source file {relative_path}')
else:
logging.debug(f"\tSkipping unchanged: {relative_path}")
unchanged += 1
# Remove handled file from queue
del sync_queue[relative_path]
# Destination file isn't in the queue, delete it
else:
logging.warning(f"\tRemoving file: {relative_path}")
removed += 1
remove(dst_path)
# Clean up empty folders; yes, there's potentially minor churn here
if folders == [] and files == [] and '_btramp' not in path:
logging.warning(f"\tRemoving folder: {path}")
rmdir(path)
newsongs = len(sync_queue)
for song in sync_queue.keys():
dst_path = Path(dst_folder + '/' + song)
folder = dst_path.parent
if not folder.exists():
logging.info(f"\tAdding folder: {folder}")
makedirs(folder)
logging.info(f"\tAdding file: {song}")
copyfile(Path(src_folder + '/' + song), dst_path)
logging.info(f"\t{newsongs} new, {unchanged} unchanged, {overwritten} updated, {removed} deleted.")
except Exception as e:
error_out(f'Unable to sync files: {e}')
logging.debug('Unmounting iPhone ...')
try:
ret = call(f'fusermount -u {dst_folder}', shell=True)
if ret != 0:
raise Exception(f'fusermount returned {ret}')
except Exception as e:
error_out(f'Unable to unmount iPhone: {e}')
logging.info('Sync complete.')
exit()
