playlistsync.py
by
Ken Manheimer
—
last modified
Oct 04, 2012 02:16 PM
Filed under:
development,
craft
python-appscript based script for syncing iTunes playlists with filesystem mounted on Mac OS X
playlistsync.py
—
Python Source,
69 kB (70915 bytes)
File contents
#!/bin/python
# -*- coding: utf-8 -*-
# playlistsync.py -
# Synchronize specified iTunes playlists and subplaylists to target folders.
# This python script runs on MacOS X with an operational iTunes installation,
# and is run using python (tested with python 2.5.4 and 2.6.1) that has
# appscript installed - see http://appscript.sourceforge.net/py-appscript .
# See the module docstring, after the comments, for details and usage info.
# Copyright ® Ken Manheimer, 2009-2012
# the latest released version can be found at:
# http://myriadicity.net/software-and-systems/craft/crafty-hacks/playlistsync.py
# contact me using http://myriadicity.net/contact-info
# $Revision: 1.86 $ $Date: 2012-10-04 18:14:00 $
#
# playlistsync.py is free software: you can redistribute it and/or modify
# it under either the terms of either:
# the Gnu Public License, v3: http://www.gnu.org/licenses/gpl-3.0.txt
# or:
# the Apache License, Version 2.0: http://www.apache.org/licenses/LICENSE-2.0
# playlistsync.py is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
"""Synchronize designated iTunes playlists to target directories.
The target directories can, of course, be mounted from a portable music device,
like a portable music player or smartphone, or a regular folder in the host or
a remote computer's filesystem.
= About =
This is a python script which runs on MacOS X with an operational iTunes
installation, and is run using python 2.6 or later that has appscript installed
(see http://appscript.sourceforge.net/py-appscript ).
When you run this script, a copy of an iTunes playlist which you specify, all
its offspring playlists, and all the tracks within that collection are
duplicated to a target filesystem folder of your choosing. By choosing a
folder mounted from your smartphone, you can duplicate your playlists and
tracks there.
Only those items necessary to rectify differences with the files already in the
target hierarchy are copied. The target directory is also purged of playlists,
track files, and directories which do not correspond to any elements within the
subject playlist's hierarchy. (Other files and directories are left
undisturbed.)
If COMPOSE_PLAYLIST_NAMES is True, the playlist files (m3u) we generate are
named for their corresponding playlists's location in the containment
hierarchy, eg "Rock.Alt.NewWave" for the NewWave playlist in the Alt playlist
folder in the Rock playlist folder. Like the playlist folders in iTunes, the
playlist files corresponding to each iTunes playlist folder include the tracks
of all the directly and indirectly contained subfolders and playlists.
Subject playlist tracks that lack a valid file path ("orphaned" tracks) are not
included in the target playlists.
= Operational Configuration =
Follow the python appscript instructions, at:
http://appscript.sourceforge.net/py-appscript/index.html
to install appscript in the python you'll be using to run the script.
Configure the script by setting variable values. See the section between
"Configuration Below" and "Configuration Above" - and alternately by using
command-line options, overriding the persistent configuration variables.
Invoke the script with '--help' to see the available options.
The sync target directory must be present. To avoid unintended deletion of
music files in mistaken target directories, the target directory must contain a
file with the name specified by TARGET_DIRECTORY_FAILSAFE_FILE. The user will
be prompted for creation of the failsafe file if absent. You can bypass this
feature by setting ALWAYS_CREATE_FAILSAFE True, or passing in "-c" /
"--create-failsafe".
See the set of configuration variables in the script for the set entire set of
behaviors you can adjust.
At the default verbosity level the script reports file and playlist changes as
it processes them, and a final tally indicating the amount of changes and the
specific playlists removed or added/changed. The list of removed playlists is
particularly useful for players like the default one in Android 1.x, where the
absence of previously existing playlists is sometimes not noticed.
= Shuffle Play With A Stub First Track =
Simplistic shuffle-play schemes (like that of Android's default music player)
always start with a playlist's first track, and shuffle from there. This
script provides a workaround, in which you designate a "stub" track - i use a
track that consists of just 1 second of silence - to be put at the front of
every playlist.
To use this feature, include such a track in your iTunes library and assign the
track's name to the variable USE_STUB_TRACK \(default = '00 Stub Track'). Do
not manually include the stub track on any of your iTunes playlists - it will
be included at the top of the filesystem playlists automatically.
= Continuation and an (obsolete) Appscript Bug Workaround =
The following problem seems to be gone in recent appscript versions, by 2011 or earlier.
As with many sychronization processes, you can interrupt the script at any
point and restart it without sacrificing the results of the work already done.
This is particularly handy in light of an appscript timeout bug which can occur
after a certain number of interactions (for me, at around 60 of my 80
playlists). If this problem is happening the appscript itunes application
bridge gets wedged and raises an CommandError timeout. The script catches that
timeout and, if it's not happening in quick sequence, it restarts the sync,
quickly proceeding past the already done work and returning to where it left
off.
"""
# Implementation notes:
#
# Doctests / Functional testing:
#
# The doctests mostly exercise the itunes playlist bridge and internal model,
# not the music files, and depend on having the SUBJECT_PLAYLIST (or --playlist
# setting) name an existing populous playlist - preferably one with
# sub-playlists.
#
# Efficiencies:
#
# A number of measures are taken to reduce unnecessary overhead in
# synchronizing what can be substantial numbers of tracks. Here are some worth
# knowing about, if you're thinking about messing around with the code:
#
# - We reduce a lot of overhead by using singleton-by-label classes, so that eg
# track and playlist instances are created only once for each distinct item.
# Every subsequent instantiation of an already existing track results in
# just a dictionary lookup in the class.__new__ method to obtain the already
# existing instance.
#
# - Similarly, we stash populations of references in instances the first time
# we obtain them, and use the stashed batches on subsequent references. This
# is crucial because playlist folders effectively contain all the tracks of
# their subfolders, so we do much repeated iteration over subsets of the
# collection of itunes tracks and playlists. It also means that changes in
# the iTunes data during the course of processing, after a reference has been
# obtained, will not be noticed. For our purposes as of this writing,
# that's actually desirable.
#
# - We consider that a track with unchanged if both its size and a hash of its
# first 10k bytes (see LEADING_BYTES_TO_HASH) is the same as the prior synced
# version. The limited length being hashed seems like an effective
# compromise that catches most or all non-size-changing track edits without
# exorbitant processing overhead.
#
# - Recognizing track changes via economical hashing also enables us to embed
# fingerprints for each track in the playlist file, as a comment after the
# respective track's entry. This way, the generated playlists as a whole
# will be changed when the hash fingerprints of any of their tracks change,
# providing an opportunity to skip hashing of the sync-destination tracks
# when unnecessary.
# vvvvvvvvvvvvvvvv Customization below vvvvvvvvvvvvvvvv #
# DEFAULTS_FILE:
# Name of file to seek in the same directory for site-specific settings.
# (Use a rooted path to locate the file in another directory.) This file
# will be loaded, as Python code, after evaluation of this section, so
# settings in that file will override these settings. This setting will be
# ignored if the value evaluates False or the named file is not found.
DEFAULTS_FILE = 'pls_defaults.py'
# SUBJECT_PLAYLIST:
# (Unique) name of playlist to be synced, including contained playlists.
# Overridden by command line option "-p" / "--playlist".
SUBJECT_PLAYLIST = u'_mobile'
# TARGET_DIRECTORY_PATH:
# Path to directory where playlists are to be synced. Overridden by command
# line option "-d" / "--directory".
TARGET_DIRECTORY_PATH = '/dev/null/'
# CHECKED_ONLY:
# Include only tracks those playlist tracks that are checked (enabled).
CHECKED_ONLY = True
# TARGET_DIRECTORY_FAILSAFE_FILE:
# Name of file in selected target directory which must be present for
# playlistsync to operate. If absent playlistsync will prompt to create it
# and continue, unless the "-c" / "--create-failsafe" option is passed,
# in which case the file will be created if necessary and playlistsync will
# run. (Absence of the target directory prevents operation, no matter
# what.) Setting ALWAYS_CREATE_FAILSAFE True forces automatic creation of
# the failsafe, effectively bypassing the interlock.
TARGET_DIRECTORY_FAILSAFE_FILE = "playlistsync-here"
ALWAYS_CREATE_FAILSAFE = False
# USE_STUB_TRACK:
# The name of a short, silent track to put at the front of all playlists,
# for devices with shuffle-play that always shuffles
# *after* the first track. Set it to None if you don't want
# this feature. Overridden by command line option "-s" / "--stub-track".
USE_STUB_TRACK = u'00 Stub Track'
# PLAYLISTS_REPORTING_INCREMENT:
# Number of playlists processed between basic processing reports. (Use "-q"
# / "--quiet" option to inhibit these reports.)
PLAYLISTS_REPORTING_INCREMENT = 10
# PLAIN_SEPARATOR:
# Character to commonly use between song-name path elements. Ommitted when
# the MAGIC_LEADING_DELIMITER, which see, is at the front of a path element.
PLAIN_SEPARATOR = "."
# MUSIC_FILES_ONLY
# = False - create/sync m3u playlist files to target folder
# = True - skip creating m3u playlist files, eg if you're using Google
# Music to sync your playlists, and use this script just to keep a
# fallback stash of music files.
MUSIC_FILES_ONLY = False
# COMPOSE_PLAYLIST_NAMES
# = False - use the subject playlist names for the target
# = True - synthesize target playlist names based on subject names,
# transforming leading characters according to MAGIC_LEADING_DELIMITER.
# The names of the subject playlists, themselves, are not altered.
COMPOSE_PLAYLIST_NAMES = False
# MAGIC_LEADING_DELIMITER:
# If not empty and COMPOSE_PLAYLIST_NAMES is True, playlist names
# that start with MAGIC_LEADING_DELIMITER's first character are handled
# specially. Here are the alternatives:
#
# - Empty string: always prepend PLAIN_SEPARATOR to the playlist name when
# creating the full playlist folder path name.
# - One character: when a playlist name's leading character matches
# MAGIC_LEADING_DELIMITER's first (and, in this case, only) character,
# then that character is removed from the playlist name when constructing
# the playlist folder path name, and no additional delimiter is prepended.
# (Thus the playlist name's second character serves as the delimiter.)
# - Two or more characters: like the one-character case, but the
# MAGIC_LEADING_DELIMITER's second character is prepended to matching
# playlist names, to serve as the delimiter when constructing the playlist
# folder path name.
# Lexical precedence in Google Music player: _-,.+=[A-Za-z]
# Lexical precedence in old android Music player: +,-.[A-Za-z]_~
## Value that sorts higher for the old, standard android player:
#MAGIC_LEADING_DELIMITER = "_+"
# Value that sorts higher for the new google-music player ("_," works, too):
MAGIC_LEADING_DELIMITER = "_-"
## Value that uses characters after the leading magic char, on the playlist
## names, themselves:
#MAGIC_LEADING_DELIMITER = "_"
# VERBOSE:
# When True, marginal blather() activity messages will be emitted. Can be
# set using the --verbose command line option, and is overridden by QUIET.
VERBOSE = False
# QUIET:
# Inhibit all blather() informational messages, even when VERBOSE is set.
# Error messages are still emitted via whoops().
QUIET = False
# DRYRUN:
# Inhibit actually changing any state/filesystem/etc if True. Progress
# report lines will be prefixed with '# '.
DRYRUN = False
# ^^^^^^^^^^^^^^^^ Customization above ^^^^^^^^^^^^^^^^ #
# vvvvvvvvvvvvvvvv Operational config below vvvvvvvvvvvvvvvv #
# (and/or some obscure customization options - caveat emptor) #
# These things can be changed, but you less likely need to, and can more easily
# break things than the common user customizations, above.
# PLAYLIST_EXTENSIONS, TRACK_EXTENSIONS:
# Sequences of extensiosn of filename to be removed from the target
# directory hierarchy if the files have no corresponding playlist or
# playlist track. Other files will be left alone, but directories left
# empty after tracked file deletions will be removed.
# The first PLAYLIST_EXTENSIONS entry will be appended after a '.' to the
# playlist names as part of forming the playlist filename.
PLAYLIST_EXTENSIONS = ('m3u', 'M3u', 'M3U')
TRACK_EXTENSIONS = ('mp3', 'MP3', 'm4p', 'M4P', 'm4a', 'M4A')
# PLAYLIST_FILE_NAMES_REGEXP:
# Adjust this regular expression to specify characters to to be replaced in
# playlist names for their filesystem file name. (By default we specify the
# negation of the set of allowed characters.) The elided characters are
# replaced by the value of SUBSTITUTION_CHARACTER.
PLAYLIST_FILE_NAMES_REGEXP = '[^- _A-Za-z0-9:%,.`~!@^=*+%()]'
SUBSTITUTION_CHARACTER = '~'
# LEADING_BYTES_TO_HASH:
# Used for the mathematical fingerprint (python hash) of track files used
# to detect non-size-changing track audio and metadata edits.
LEADING_BYTES_TO_HASH = 10000
# DENSITY_LOG_BASE
# The logarithmic base for biasing the play density so continued plays as a
# track ages have more importance, or conversely, so the earlier rush to
# play a new track doesn't obscure the staying power of older favorites.
# The tuning of this setting is by feel, but in my (very slight) exploration
# doesn't seem to make a lot of difference except maybe in the low "play
# densities" (see Track.get_play_density()).
DENSITY_LOG_BASE = 1.2
# ^^^^^^^^^^^^^^^^ Operational config above ^^^^^^^^^^^^^^^^ #
import appscript
import sys, os, time, shutil, re, optparse, datetime, math
ITUNES_PATH = u'iTunes.app'
re_playlist_file_names = re.compile(PLAYLIST_FILE_NAMES_REGEXP)
# SingletonByLabelAndClass belongs in a separate module, but by including it
# here we can deliver this application as a single script, so for now:
class SingletonByLabelAndClass(object):
"""Inherit for classes that have distinctly labelled unique instances.
The distinguishing label is taken from a keyword argument named 'label'.
>>> class DistinctService(SingletonByLabelAndClass):
... counter = 0
... def __init__(self, label='default'):
... self.counter += 1
...
>>> first = DistinctService(label='first')
>>> first_too = DistinctService(label='first')
>>> second = DistinctService(label='second')
>>> first.counter == first_too.counter
True
>>> first.counter != second.counter
True
>>> id(first) == id(first_too)
True
>>> id(first) != id(second)
True
>>> class DifferentDistinctService(SingletonByLabelAndClass):
... counter = 0
... def __init__(self, label='default'):
... self.counter -= 1
...
>>> other_first = DifferentDistinctService(label='first')
>>> id(other_first) != id(first)
True
>>> first._SingletonReset()
>>> first = DistinctService(label='first')
>>> id(first) == id(first_too)
False
"""
__instances = {}
def __new__(cls, *args, **kwargs):
# avoid classes colliding by using class-name/specified-name combo:
name = (cls.__name__, kwargs['label'])
got = cls.__instances.get(name, None)
if got == None:
got = cls.__instances[name] = object.__new__(cls)
return got
@classmethod
def _SingletonReset(cls):
"""Development-mode - clear the established instances.
Use for fresh instances during incremental development of
singleton-based classes."""
for i in cls.__instances.values():
del i
cls.__instances = {}
class Playlist(SingletonByLabelAndClass):
"""Encapsulation of an iTunes playlist object.
Playlists within the sync subject playlist must be initted with their
dotted name, reflecting the containment.
>>> library = Playlist(label='Library')
>>> subject = Playlist(label=SUBJECT_PLAYLIST)
"""
label = None
# Short-circuit redundant visits and accounting distortion after
# appscript-timeout restarts:
_already_done = None
_is_terminal = None
_pedigree = None
_contained_tracks = None
_children_playlists = None
_playlist_file_string = None
_playlist_file_name = None
def __init__(self, label=None, ref=None):
"""Realize playlist for string LABEL, passed in as a keyword argument.
To disambiguate playlists with the same terminal names, LABEL must be
the dotted playlist containment path from the subject playlist.
REF is for internal use on contained playlists, to disambiguate."""
if self.label is not None:
# we're a singleton that's already been initted.
return
self.label = label
self._mgr = mgr = PlaylistSyncManager(label=SUBJECT_PLAYLIST)
if not ref:
ref = mgr.get_playlist_refs(label)
if not ref:
raise ValueError, ("No playlists labelled '%s' found" % label)
elif type(ref) == appscript.reference.Reference:
self._ref = ref
else:
if len(ref) > 1:
raise ValueError, ("%s ambiguous - mutiple found" % label)
else:
self._ref = ref[0]
mgr.distinct_playlist(self)
def __repr__(self):
return "<%s.%s instance '%s'>" % (self.__module__,
self.__class__.__name__,
self.label)
def is_terminal_playlist(self):
if self._is_terminal is None:
self._is_terminal = (self._ref.special_kind.get()
== appscript.k.none)
return self._is_terminal
def sync_to_directory(self, target_dir_path):
"""Sync our playlist constituents to TARGET_DIRECTORY_PATH.
A playlist file will not be created for this encompassing playlist,
itself, but will be created for its constituents."""
# Employ helpers, including ._sync_to_directory(), to do actual work.
fldr_mgr = PlayfolderManager(label=target_dir_path)
self._sync_to_directory(fldr_mgr, save_playlist=False)
def _sync_to_directory(self, fldr_mgr, save_playlist=True):
"""Recursively sync using FOLDER_MANAGER with already-set dir path.
Set optional SAVE_PLAYLIST False to overriding saving of the playlist
file, itself. (Useful for skipping save of the top-level, which
generally is redundant with universal shuffle that most players
have."""
# We use discrepancies in playlist files to lead to those terminal
# (non-folder) playlists where discrepancies in track collections and
# individual tracks are reconciled. Hence, in playlist folders,
# attention is devoted only to the playlist files for syncronization,
# and, if the folder's playlist has changed, to their subfolders and
# playlists to narrow down which harbors the constituent changes.
#
# So we recursively drill down the subject playlist-folder/playlist
# hierarchy, synthesizing the playlist files and comparing them with
# the existing files in the target directory hierarchy.
if self._already_done is not None:
# Redundant visits happen after appscript-timeout restarts.
return
playflnm = self.get_playlist_file_name()
tally_name = self.get_tally_name()
playflstr = self.get_playlist_file_string()
dotted = self.get_dotted_name()
changed = fldr_mgr.playlist_file_is_changed(playflnm, playflstr)
if save_playlist:
if changed:
status = 'changed'
else:
status = 'unchanged'
else:
status = 'top-level'
#blather('%s playlist %s\n' % (status, playflnm), marginal=True)
blather('%s playlist %s\n' % (status, playflnm), marginal=False)
if self.is_terminal_playlist():
if changed:
# ... process contained tracks. tracks can be situated
# repeatedly, work is expended only the first time.
for track in self.get_contained_tracks():
fldr_mgr.situate_track_file(track)
else:
# ... mark contained tracks as situated.
for track in self.get_contained_tracks():
fldr_mgr.mark_as_situated(track.get_tally_name())
else:
# ... process contained playlist folders/playlists.
for sub in self.get_child_playlists():
sub._sync_to_directory(fldr_mgr, save_playlist=True)
# last, situate the changed playlist file:
if save_playlist:
if changed:
fldr_mgr.situate_playlist_file(self)
else:
fldr_mgr.mark_as_situated(tally_name)
elif fldr_mgr.is_marked_as_situated(tally_name):
fldr_mgr.mark_as_not_situated(tally_name)
self._already_done = True
def get_child_playlists(self):
"""Return a list of immediate offspring playlists and playlist folder.
>>> type(Playlist(label=SUBJECT_PLAYLIST).get_child_playlists())
<type 'list'>
"""
if self._children_playlists is None:
got = []
for ref in self._mgr.get_offspring(self._ref):
subdotted = self.get_sub_dotted_name(ref.name.get())
got.append(Playlist(label=subdotted, ref=ref))
self._children_playlists = got
return self._children_playlists
def get_contained_playlists(self, flat=False):
"""Return nesting of all contained playlists, including self.
When True, optional FLAT (default False) means return a flat rather
than nested list.
>>> type(Playlist(label=SUBJECT_PLAYLIST).get_contained_playlists())
<type 'list'>
"""
got = [self]
op = got.extend if flat else got.append
for offspring in self.get_child_playlists():
op(offspring.get_contained_playlists(flat=flat))
return got
def get_contained_tracks(self):
"""Return list of all contained tracks, including self.
We exclude orphaned tracks, ie those lacking a music file."""
# get the contained refs from iTunes only the first time.
if self._contained_tracks:
return self._contained_tracks
else:
tracks = []
for ref in self._ref.tracks.get():
if CHECKED_ONLY and not ref.enabled.get():
blather("(Noticed disabled track: %s.)\n" % ref.name.get())
continue
gotloc = ref.location.get()
if gotloc == appscript.k.missing_value:
blather("Track missing: %s..." % ref.name.get())
continue
tracks.append(Track(label=gotloc.path.encode('utf-8',
'replace'),
ref=ref))
if USE_STUB_TRACK:
stub_track = self._mgr.get_stub_track()
if stub_track:
tracks.insert(0, stub_track)
self._contained_tracks = tracks
return tracks
def get_dotted_name(self):
"""Return the playlist's dotted name, with respect to the subject."""
return ".".join(self.get_pedigree())
def get_sub_dotted_name(self, subname):
"""Return a sub-list NAME as dotted name w.r.t. the subject playlist."""
return self.get_dotted_name() + '.' + subname
def get_pedigree(self):
"""Return a list of strings of the playlist containment to subject, if
COMPOSE_PLAYLIST_NAMES is False. Otherwise, return the playlist name.
"""
# We cache the computed pedigree; clear the value if playlists change.
if self._pedigree:
return self._pedigree
else:
pedigree = self._pedigree = self._mgr.get_pedigree(self._ref)
return pedigree
def get_playlist_file_name(self):
"""Return playlist dotted name sanitized for use as a filesystem name.
>>> Playlist(label='Library').get_playlist_file_name()
u'Library.m3u'
"""
if self._playlist_file_name is not None:
return self._playlist_file_name
sanitized = [re_playlist_file_names.sub(SUBSTITUTION_CHARACTER, pl)
for pl in self.get_pedigree()]
if not MAGIC_LEADING_DELIMITER or not COMPOSE_PLAYLIST_NAMES:
separator = PLAIN_SEPARATOR
constructed = sanitized
else:
leading = MAGIC_LEADING_DELIMITER[0]
if len(MAGIC_LEADING_DELIMITER) > 1:
separator = MAGIC_LEADING_DELIMITER[1]
else:
separator = ""
constructed = []
got = False
for element in sanitized:
if element and element[0] == leading:
# omit element's magic leading char
# and include alt (or no) separator:
got = True
constructed += separator + element[1:]
else:
if constructed:
# add in a PLAIN_SEPARATOR explicitly:
constructed += PLAIN_SEPARATOR
constructed += element
self._playlist_file_name = "%s.%s" % ("".join(constructed),
PLAYLIST_EXTENSIONS[0])
return self._playlist_file_name
get_tally_name = get_playlist_file_name
def get_playlist_file_string(self):
"""Return an m3u playlist as a string."""
if self._playlist_file_string is None:
global USE_STUB_TRACK
tracks = self.get_contained_tracks()
got = ["#EXTM3U"] + [track.get_playlist_file_entry()
for track in tracks
# omit tracks that lack a valid path:
if track.get_fs_path()]
self._playlist_file_string = "\n".join(got)
return self._playlist_file_string
numtracks = 0
class Track(SingletonByLabelAndClass):
"""Encapsulation of an iTunes track."""
_refs = None
_name = None
_size = None
_played_count = None
_played_density = None
_days_age = None
_fs_path = None
_relative_fs_path = None
_playlist_file_entry = None
_partial_hash = None
def __init__(self, label=None, ref=None):
"""Manifest an appscript itunes track with path LABEL and REFERENCE.
\(NAME keyword arg for SingletonByLabelAndClass operation.)"""
if self._refs is not None:
# we're a singleton that's already been initted.
self._refs.append(ref)
return None
self._fs_path = label
self._refs = [ref]
def get_size(self):
"""Return the size in bytes."""
if self._size is None:
self._size = self._refs[0].size.get()
return self._size
def get_name(self):
"""Return the track name, utf-8 encoded."""
if self._name is None:
self._name = self._refs[0].name.get().encode('utf-8', 'replace')
return self._name
def get_artist(self):
"""Return the track artist, utf-8 encoded."""
return self._refs[0].artist.get().encode('utf-8', 'replace')
def get_play_count(self):
"""Return the tracks played_count value."""
if self._played_count is None:
self._played_count = self._refs[0].played_count.get()
return self._played_count
def get_play_density(self):
"""Return an index related to average play count per day.
We use an exponential curve so continued plays as a track ages have
greater importance. In addition, the density index is limited to not
exceed the number of days, so reduce exaggeration of new tracks
importance in the enthusiastic replays of their first acquisition."""
if self._played_density is None:
days_age = self.get_days_age()
density = (self.get_play_count()
/ max(math.log(days_age, DENSITY_LOG_BASE), 1))
# density = self.get_play_count() / max(days_age,2)
if density > days_age:
density = days_age
self._played_density = density
return self._played_density
def get_days_age(self):
"""Return age in decimal days since added to the library."""
if self._days_age is None:
age = datetime.datetime.now() - self._refs[0].date_added.get()
self._days_age = age.days + (age.seconds / 867400.0)
return self._days_age
def get_fs_path(self):
return self._fs_path
def get_partial_hash(self):
"""Return a hash of some leading portion of the track.
The leading portion is dictated by LEADING_BYTES_TO_HASH"""
if self._partial_hash is None:
path = self.get_fs_path()
if path:
self._partial_hash = get_file_partial_hash(path)
else:
self._partial_hash = 0
return self._partial_hash
def open(self, mode='r', buffering=None):
"""Return a file object opened with optional MODE and BUFFERING."""
if buffering is None:
return open(self.get_fs_path(), mode=mode)
else:
return open(self.get_fs_path(), mode=mode, buffering=buffering)
def get_relative_fs_path(self):
"""Return path of track relative to iTunes/iTunes Music folder."""
if self._relative_fs_path is None:
path = self.get_fs_path()
in_itunes_folder = path.find('iTunes/iTunes Music')
if in_itunes_folder:
# Trim a lot of common excess.
path = path[in_itunes_folder+20:]
self._relative_fs_path = path
syncmgr.distinct_track(self)
return self._relative_fs_path
get_tally_name = get_relative_fs_path
def get_playlist_file_entry(self):
"""Return a string suitable for use as an m3u playlist entry."""
# #EXTINF:#seconds,Artist - Title
# relative/path/to/file
if self._playlist_file_entry is None:
ref = self._refs[0]
file_string = ("#EXTINF:%i,%s - %s\n%s\n# %s"
% (ref.duration.get(),
ref.artist.get().encode('utf-8', 'replace'),
self.get_name(),
self.get_relative_fs_path(),
self.get_partial_hash()))
self._playlist_file_entry = file_string
return self._playlist_file_entry
def __repr__(self):
return "<%s.%s instance '%s' @ %s>" % (self.__module__,
self.__class__.__name__,
self.get_name(),
self.get_relative_fs_path())
class PlayfolderTally(object):
"""Utlity by which PlayfolderManager resources are tracked.
The resources include files and directories in the target playfolder
hierarchy."""
_playfolder_map = None
# _playfolder_map: {'dotted.name': {'type': 'playlist' | 'track',
# 'situated': False | True},
# 'removed': False | True},
# ...}
def __init__(self):
self.scan()
def scan(self):
"""Tally directory hierarchy contents for tracking."""
self._playfolder_map = map = {}
base_path = self._base_path
count = 0
blather("existing target dirs: ")
for dirpath, dirnames, filenames in os.walk(base_path):
count += 1
if (count % 100) == 0:
blather(".", noflag=True)
blather(" %d" % count, noflag=True, marginal=True)
relpath = dirpath[len(base_path)+1:]
if relpath: relpath += '/'
for fname in filenames:
split = fname.split('.')
if len(split) == 1:
continue
else:
extension = split[-1]
if extension in PLAYLIST_EXTENSIONS:
map[relpath + fname] = {'type': 'playlist',
'situated': False,
'removed': False}
elif extension in TRACK_EXTENSIONS:
map[relpath + fname] = {'type': 'track',
'situated': False,
'removed': False}
blather(" %d.\n" % count, noflag=True)
def mark_as_situated(self, tracking_name):
"""Register TRACKING_NAME for retention in cleanup phase."""
pfmap = self._playfolder_map
entry = pfmap.get(tracking_name, None)
if not entry:
pfmap[tracking_name] = {'situated': True, 'removed': False}
else:
entry.update({'situated': True, 'removed': False})
def is_marked_as_situated(self, tracking_name):
"""Register TRACKING_NAME for retention in cleanup phase."""
entry = self._playfolder_map.get(tracking_name, None)
if not entry:
return False
else:
return entry['situated']
def mark_as_not_situated(self, tracking_name):
"""Register TRACKING_NAME for removal in cleanup phase."""
pfmap = self._playfolder_map
entry = pfmap.get(tracking_name, None)
if not entry:
pfmap[tracking_name] = {'situated': False, 'removed': False}
else:
entry['situated'] = False
def mark_as_removed(self, tracking_name):
"""Register removed state of TRACKING_NAME during cleanup."""
pfmap = self._playfolder_map
entry = pfmap.get(tracking_name, None)
if not entry:
pfmap[tracking_name] = {'removed': True}
else:
entry['removed'] = True
def get_orphaned_file_tally(self):
"""List of unclaimed file resource paths in target hierarchy.
Returns a list of (relative_path, kind) pairs, where kind is either
'playlist' or 'track'.
Resources include tracks and playlist files having extensions among
those listed in TRACK_EXTENSIONS and PLAYLIST_EXTENSIONS."""
items = self._playfolder_map.items()
# impose some order, rather than scattershot:
def tupsanitize(pair):
"""Insulate sort comparison function from unicode problems."""
return pair[0].decode("utf-8", "replace")
# XXX I'm not certain this is the *right* solution, but it prevents
# tracebacks. It may cause file leakage, where tracks are removed, but
# their files, having funky characters in their names, are not removed.
# Without tupsanitize, sort on some filename strings can result in:
# UnicodeWarning: Unicode equal comparison failed to convert
# both arguments to Unicode - interpreting them as being unequal
# and then a UnicodeDecodeError exception.
# Python 2.5.4, Apple Inc. build 5646.
items.sort(key=tupsanitize)
got = [(path, state['type'])
for (path, state) in items
if not state['situated']
and not state['removed']]
return got
def get_orphaned_directory_tally(self):
"""Generator of empty directory paths in target hierarchy."""
for (path, dirs, files) in os.walk(self._base_path, topdown=False):
for dir in dirs:
if not os.path.exists(os.path.join(path, dir)):
# exclude directory that's already been removed:
dirs.remove(dir)
if (not dirs) and (not files):
yield path
class PlayfolderManager(SingletonByLabelAndClass, PlayfolderTally):
"""Maintain playlist folder contents and tally for removal of unused items.
"Playfolder" is used to mean the filesystem directory hierarchy to which
iTunes playlist contents - tracks and m3u-type playlist files - are
synced."""
_base_path = None
# NOTE: we mustn't instantiate Playlist objects during our __init__ because
# Playlist obtains the PlayfolderManager during its' __init__.
def __init__(self, label=None):
"""Initialize rooted at LABEL = target sync directory path.
DIRECTORY_PATH is '~x' user and '$var' expanded."""
if self._playfolder_map is not None:
# we're a singleton that's already been initted.
return
blather("PlayfolderManager init")
directory_path = os.path.abspath(label)
expanded = os.path.expanduser(os.path.expandvars(directory_path))
if not DRYRUN:
if not os.path.exists(expanded):
raise ValueError, ('Target sync directory %s not found'
% directory_path)
elif not os.path.isdir(expanded):
raise ValueError, ('Target sync path %s not directory'
% directory_path)
self._base_path = expanded
PlayfolderTally.__init__(self)
def __repr__(self):
return ("<%s.%s instance for '%s'>"
% (self.__module__, self.__class__.__name__,
self._base_path))
def situate_track_file(self, track):
"""Locate Track instance properly in target directory, and register.
"Properly" means in the same location relative to the _base_path as the
track is relative to the iTunes music library.
If the track is already marked as situated we don't do anything.
If not present or different size than existing target track, then the
music file is copied to the target location. Otherwise, no changes are
made.
In either case, the tally of files is updated to indicate that the file
is not orphaned, and should be retained during the clean-up phase."""
tally_name = track.get_tally_name()
if self.is_marked_as_situated(tally_name):
return
abs_track_path = track.get_fs_path()
rel_track_path = track.get_relative_fs_path()
target_track_path = os.path.join(self._base_path, rel_track_path)
do_copy = False
if not os.path.exists(target_track_path):
# target copy absent:
if not os.path.exists(os.path.dirname(target_track_path)):
# create intervening dirs.
try:
self.create_directory(os.path.dirname(rel_track_path))
do_copy = True
except OSError:
# eg, aberrant deleted tracks still found on paths starting
# with /cache/, on smart playlists.
syncmgr.skipped_track(track)
else:
# Directory exists, file does not.
do_copy = True
elif (track.get_partial_hash()
!= get_file_partial_hash(target_track_path)):
# target copy is different:
do_copy = True
syncmgr.changed_track(track)
if do_copy:
do_file_copy(abs_track_path, target_track_path)
syncmgr.added_track(track)
self.mark_as_situated(tally_name)
def situate_playlist_file(self, playlist):
"""Ensure that current version of playlist file is properly located."""
if MUSIC_FILES_ONLY:
return
path = os.path.join(self._base_path,
playlist.get_tally_name().decode("utf-8",
"replace"))
contents = playlist.get_playlist_file_string()
if self.playlist_file_is_changed(path, contents):
do_file_write(path, contents)
syncmgr.changed_playlist(playlist)
self.mark_as_situated(playlist.get_tally_name().decode("utf-8",
"replace"))
def playlist_file_is_changed(self, filename, contents):
"""True iff playlist FILENAME's CONTENTS different than established one.
"Changed" includes new additions."""
base = self._base_path
path = os.path.join(base, filename)
return (not os.path.exists(path)) or (open(path).read() != contents)
def remove_unused_tracks_and_playlists(self):
"""Remove unused files and directories from target directory."""
# - determined orphaned items from tallies and remove them
# - remove remaining empty directories *excluding* the sync root,
# using os.walk for depth-first directory removal.
base_path = self._base_path
for fpath, kind in self.get_orphaned_file_tally():
path = os.path.join(base_path, fpath)
if kind == 'track':
syncmgr.removed_track(os.stat(path).st_size, path)
elif kind == 'playlist':
syncmgr.removed_playlist(path)
do_remove(path)
for dpath in self.get_orphaned_directory_tally():
do_rmdir(dpath)
def create_directory(self, relative_path):
"""Create a directory at PATH relative to base dir.
Intervening directories are created if necessary.
Return True if a directory was created, else False if unnecessary."""
base = self._base_path
container, dir = os.path.split(relative_path)
abs_container = os.path.join(base, container)
# XXX put tracks hierarchy in subdir to prevent collisions w/playlists?
if (container and (container != os.path.sep)
and not os.path.exists(abs_container)):
self.create_directory(container)
do_mkdir(os.path.join(base, relative_path))
def get_dotted_name(self, path):
"""Return the dotted name for PATH w.r.t. the base path.
PATH should lie within the base path.
If not, the result starts with a '.'"""
if path[:len(self._base_path)] == self._base_path:
path = path[len(self._base_path)+1:]
return path.replace('/', '.')
class PlaylistSyncTally(object):
"""Utility by which PlaylistSyncManager resources are tracked."""
itunes = None
_name_to_refs = None
_dotted_name_to_refs = None
# mapping from playlist NAME to list of pl refs where pl NAME is parent.
_ref_to_child_refs = None
_stub_track = None
def __init__(self):
self.itunes = itunes = AppInstance(label=ITUNES_PATH)
self.scan()
def scan(self):
"""Infer the playlists organization - name-to-ref and hierarchy."""
ntr = self._name_to_refs = {}
rtor = self._ref_to_child_refs = {}
count = 0
blather("All playlists in Library: ")
for ref in self.itunes.playlists.get():
count += 1
if (count % 100) == 0:
blather(".", noflag=True)
blather(" %d" % count, noflag=True, marginal=True)
# We don't exclude any by kind.
rtor[ref] = []
name = ref.name.get()
if ntr.has_key(name):
ntr[name].append(ref)
else:
ntr[name] = [ref]
# the number of refs apparently is twice the number of playlists?
blather(" %d.\n" % (count/2), noflag=True)
for reflist in self._name_to_refs.values():
for ref in reflist:
parent = ref.parent
if ref.parent.exists():
parent = ref.parent.get()
if not rtor.has_key(parent):
# iTunes 9.0(70): 'Genius Mixes' parent is a *different*
# 'Genius Mixes' object, not in rtor?
rtor[parent] = [ref]
else:
rtor[parent].append(ref)
def contained_playlist_names(self):
"Return the names of all the contained regular playlists."
return self._name_to_refs.keys()
def get_playlist_refs(self, name):
"""Return playlist references for contained playlists having NAME."""
refs = self._name_to_refs.get(name)
if not refs:
raise ValueError, "Playlist '%s' not found." % name
return refs
def get_offspring(self, ref):
"""Return list of playlist REF's immediate offspring references."""
return self._ref_to_child_refs[ref]
MB = 2**20
KB = 2**10
class ProgressMeter(object):
"""Collect and incrementally emit progress statistics."""
last_reported_count = 0
initial_report = True
tracks_reporting_increment = 0
_track_paths_vs_entries = None
tracks_added_count = 0
tracks_changed_count = 0
tracks_skipped_count = 0
tracks_written_bytes = 0.0
tracks_removed_count = 0
tracks_removed_bytes = 0.0
total_playlists = -1 # top-level playlist is always "new"
playlists_written = []
playlists_removed = []
def __init__(self,
playlists_reporting_increment=PLAYLISTS_REPORTING_INCREMENT):
self.playlists_reporting_increment = playlists_reporting_increment
self._track_paths_vs_entries = {}
def tick(self, force=False, identify_playlist_changes=False):
total = len(self._track_paths_vs_entries)
if QUIET:
return
if (force or identify_playlist_changes
or (self.initial_report and total)
or ((self.total_playlists - self.last_reported_count
>= self.playlists_reporting_increment))):
self.initial_report = False
self.last_reported_count = self.total_playlists
blather("processed %i playlist%s, %i added or changed, %i removed\n"
% (self.total_playlists,
'' if self.total_playlists == 1 else 's',
len(self.playlists_written),
len(self.playlists_removed)))
if self.tracks_skipped_count:
skipped_or_blank = ", %i skipped" % self.tracks_skipped_count
else:
skipped_or_blank = ""
blather(("tracks: %i; %i added + %i chngd = %.2f ~MB,"
+ " %i removed = -%.2f MB%s\n")
% (total,
self.tracks_added_count, self.tracks_changed_count,
self.tracks_written_bytes/MB,
self.tracks_removed_count,
self.tracks_removed_bytes/MB,
skipped_or_blank))
if identify_playlist_changes:
removed_num = len(self.playlists_removed)
plural = '' if removed_num == 1 else 's'
punc = ':' if removed_num else '.'
blather("Removed %i playlist%s%s\n"
% (removed_num, plural, punc))
for pl_path in self.playlists_removed:
blather(" %s\n"
% os.path.basename(pl_path.decode("utf-8",
"replace")))
written_num = len(self.playlists_written)
plural = '' if written_num == 1 else 's'
punc = ':' if written_num else '.'
blather("Changed or added %i playlist%s%s\n"
% (written_num, plural, punc))
for pl in self.playlists_written:
blather(" %s\n"
% pl.get_tally_name().decode("utf-8", "replace"))
def distinct_track(self, path):
vs = self._track_paths_vs_entries
if vs.has_key(path):
vs[path] += 1
else:
vs[path] = 1
def added_track(self, track):
self.tracks_added_count += 1
bytes = track.get_size()
self.tracks_written_bytes += bytes
blather("added_track %s (%i)\n"
% (track.get_tally_name().decode("utf-8", "replace"), bytes),
marginal=True) # klm debug
def changed_track(self, track):
self.tracks_changed_count += 1
bytes = track.get_size()
self.tracks_written_bytes += track.get_size()
blather("changed_track %s (%iB)\n"
% (track.get_tally_name().decode("utf-8", "replace"), bytes),
marginal=True) # klm debug
def skipped_track(self, track):
self.tracks_skipped_count += 1
bytes = track.get_size()
blather("skipped_track %s (%iB)\n"
% (track.get_tally_name().decode("utf-8", "replace"), bytes))
def removed_track(self, bytes, path):
self.tracks_removed_count += 1
self.tracks_removed_bytes += bytes
blather("removed_track %s (%iB)\n" % (path, bytes))
def distinct_playlist(self, playlist):
self.total_playlists += 1
blather("distinct_playlist %s\n"
% playlist.get_tally_name().decode("utf-8", "replace"),
marginal=True)
self.tick()
def changed_playlist(self, playlist):
self.playlists_written.append(playlist)
blather("changed_playlist %s\n"
% playlist.get_tally_name().decode("utf-8", "replace"))
def removed_playlist(self, path):
self.playlists_removed.append(path)
blather("removed_playlist %s\n" % (path))
class PlaylistSyncManager(SingletonByLabelAndClass, PlaylistSyncTally,
ProgressMeter):
"""Represent itunes playlists included within a named playlist.
Initting with the same label gets the same PlaylistSyncManager instance."""
label = None
_stub_track = None
subject_playlist = None
def __init__(self, label=None):
"""Initialize with label of the subject playlist."""
global syncmgr
if self.label is not None:
# we're a singleton that's already been initted.
return
blather("PlaylistSyncManager init")
if label == None:
raise ValueError,("%s must be initted with label=<subject_playlist>"
% self.__class__.__name__)
self.label = label
PlaylistSyncTally.__init__(self)
ProgressMeter.__init__(self)
subject = self.itunes.playlists[label]
if not subject.exists():
raise ValueError, ("Configured subject playlist '%s' not found"
% label)
else:
self.subject_playlist = subject.get()
syncmgr = self
def get_pedigree(self, ref):
"""Return a list of names of the playlists containing REF to subject."""
subject = self.subject_playlist
current = ref
pedigree = [current.name.get()]
if not COMPOSE_PLAYLIST_NAMES:
return pedigree
done = False
while True:
if current.parent.exists():
current = current.parent.get()
if current != subject:
pedigree.insert(0, current.name.get())
else:
break
else:
break
return pedigree
def get_dotted_name(self, ref):
"""Return playlist REF's dotted name with respect to sync subject."""
return ".".join(self.get_pedigree(ref))
def get_stub_track(self):
"""Return track for front of m3u lists, or None. See USE_STUB_TRACK"""
global USE_STUB_TRACK
if not USE_STUB_TRACK:
return None
if self._stub_track is None:
stub = self.itunes.tracks[USE_STUB_TRACK]
if stub.exists():
utf8_path = stub.location.get().path.encode('utf-8', 'replace')
self._stub_track = Track(label=utf8_path, ref=stub.get())
else:
whoops("Specified USE_STUB_TRACK '%s' not found, zeroing\n"
% USE_STUB_TRACK)
USE_STUB_TRACK = None
return self._stub_track
def __repr__(self):
return "<%s.%s instance '%s'>" % (self.__module__,
self.__class__.__name__,
self.label)
class AppInstance(SingletonByLabelAndClass):
"""Access to a mac application object for given app path.
Initting with the same pathname gets the same python object.
>>> itunes = AppInstance(label=u'iTunes.app')
"""
# getattr/setattr so appscript.app interface can be used via the instance.
def __init__(self, label=None):
if self.__dict__.has_key('_app'):
# we're a singleton that's already been initted.
return None
self.__dict__['_app'] = appscript.app(label)
def __getattr__(self, attr):
_app = self.__dict__['_app']
if hasattr(_app, attr):
return getattr(_app, attr)
raise AttributeError, ("No such attribute '%s' on %s"
% (attr, _app))
def __setattr__(self, attr, value):
if attr == '_app':
self.__dict__['_app'] = value
else:
_app = self.__dict__['_app']
if hasattr(_app, attr):
return _app.set(attr, value)
else:
raise AttributeError, ("No such attribute '%s' on %s"
% (attr, _app))
def get_file_partial_hash(path, bytes=LEADING_BYTES_TO_HASH):
"""For file PATH, return python hash of the first BYTES bytes.
BYTES defaults to LEADING_BYTES_TO_HASH."""
return hash(open(path).read(bytes))
def get_partial_hash(string, bytes=LEADING_BYTES_TO_HASH):
"""For STRING, return python hash of the first BYTES bytes.
BYTES defaults to LEADING_BYTES_TO_HASH."""
return hash(string[:bytes])
## we could use md5 digest, but it implies way waay more precision than we
## actually have - the partial-file hash allows false negatives to a degree
## that would completely obscure any gains with a higher-entropy hash.
#from hashlib import md5
#def get_file_partial_digest(path, bytes=LEADING_BYTES_TO_DIGEST):
# """For file PATH, return python hash of the first BYTES bytes.
#
# BYTES defaults to LEADING_BYTES_TO_DIGEST."""
# digester = md5()
# digester.update(open(path).read(bytes))
# return digester.hexdigest()
#
#def get_partial_digest(string, bytes=LEADING_BYTES_TO_DIGEST):
# """For STRING, return python hash of the first BYTES bytes.
#
# BYTES defaults to LEADING_BYTES_TO_DIGEST."""
# digester = md5()
# digester.update(string[:bytes])
# return digester.hexdigest()
def do_file_copy(from_path, to_path):
blather("shutil.copy2(%s, %s)\n" % (from_path, to_path), marginal=True)
if not DRYRUN:
shutil.copy2(from_path, to_path)
def do_file_write(path, contents):
blather("open(%s, 'w').write(playlist_contents)\n" % path, marginal=True)
if not DRYRUN:
open(path, 'w').write(contents)
def do_mkdir(path):
blather("os.mkdir(%s)\n" % path, marginal=True)
if not DRYRUN:
os.mkdir(path)
def do_remove(path):
blather("os.remove(%s)\n" % path, marginal=True)
if not DRYRUN:
os.remove(path)
def do_rmdir(path):
blather("os.rmdir(%s)\n" % path, marginal=True)
if not DRYRUN:
os.rmdir(path)
def do_failsafe_check():
"""Return True if the failsafe file is present or if we create it.
We offer to create it if not found, returning true if we create it or it
was already present."""
if DRYRUN:
return True
failsafe_path = os.path.join(TARGET_DIRECTORY_PATH,
TARGET_DIRECTORY_FAILSAFE_FILE)
create = False
if os.path.exists(failsafe_path):
if create_failsafe:
blather("failsafe %s already present\n" % failsafe_path,
marginal=True)
return True
if not os.path.exists(TARGET_DIRECTORY_PATH):
raise EnvironmentError, (2, "No such file or directory",
TARGET_DIRECTORY_PATH)
if create_failsafe:
create = True
else:
sys.stdout.write("Create failsafe file:\n '%s'? [y/yes/n/NO] "
% failsafe_path)
create = sys.stdin.readline().lower()[:-1] in ['y', 'yes']
if create:
blather("create failsafe %s\n" % failsafe_path, marginal=True)
if not DRYRUN:
open(failsafe_path, 'w').close()
return True
else:
blather('Failsafe absent.\n', marginal=True)
return False
quiet_stack = []
def push_quiet(setting):
"""Set QUIET state to SETTING, preserving prior setting for pop_quiet()."""
global QUIET
quiet_stack.append(QUIET)
QUIET = setting
def pop_quiet():
"""Restore prior QUIET setting, removing the setting from the stack."""
global QUIET
QUIET = quiet_stack.pop()
def blather(msg, marginal=False, force=False, noflag=False):
"""Print informational MESSAGE unless in QUIET mode.
If optional MARGINAL is True (default, False), then the message is emitted
only when VERBOSE obtains (and only if QUIET is not set).
Optional FORCE means to print regardless of modes *except* QUIET.
Blathered messages are preceded by a '# ' (during dry-run) or a ': '
otherwise. Optional NOFLAG, however, inhibits both flags when set (which
it is not, by default)."""
if QUIET:
return
if not force:
if marginal and not VERBOSE:
return
flag = '' if noflag else '# ' if DRYRUN else ': '
# these awful contortions are to ensure string encoding safety - this is
# the minimum i could devise that satisfies the encoding variations across
# my trial playlist's 1800+ track filenames. there must be a better way?
try: composed = ("%s%s" % (flag, msg.encode("utf-8", "replace")))
except UnicodeError:
composed = ("%s%s" % (flag, msg.decode("utf-8", "replace")))
try: sys.stderr.write(composed)
except UnicodeError:
sys.stderr.write(composed.encode('ascii', 'replace'))
def whoops(msg):
sys.stderr.write(msg)
# developer convenience:
def _reset_singletons():
for klass in [Playlist, PlayfolderManager, PlaylistSyncManager,
Track, AppInstance]:
klass._SingletonReset()
def present_track_stats(playlist, amount):
"""Present the AMOUNT tracks that are enrolled in the most playlists.
If AMOUNT is a negative number, we show the tracks with lowest enrollment.
Note that we only count enrollment in terminal playlists, not playlist
folders."""
blather("getting target playlist enrollment statistics for top %i tracks\n"
% amount)
tracks = {}
highest = True
if amount < 0:
amount *= -1
highest = False
push_quiet(True)
target_playlists = playlist.get_contained_playlists(flat=True)
pop_quiet()
for l in target_playlists:
if not l.is_terminal_playlist():
blather('_', noflag=True)
else:
blather('+', noflag=True)
for t in l.get_contained_tracks():
if tracks.has_key(t):
tracks[t] += 1
else:
tracks[t] = 1
blather('\n', noflag=True)
tl = [(y, x) for x, y in tracks.items()]
tl.sort(reverse=highest)
print "#lists density - Track-name (artist)"
for i, t in tl[:amount]:
print "%i %.3f - %s (%s)" % (i, t.get_play_density(),
t.get_name(), t.get_artist())
def present_play_stats(playlist, amount, ditch_outliers=0):
"""Present AMOUNT playlists having greatest average number of track plays.
If AMOUNT is negative show abs(AMOUNT) playlists having least track plays.
DITCH_OUTLIERS specifies the number of top and bottom track play-count
values to discard as outliers.
We average the total amount of contained track plays versus number of
tracks, and count folders as well as terminal playlists."""
if amount == 0:
sys.stderr.write("play stats amount must be positive or negative in\n")
raise
blather("getting %i %s playlist by average number of track plays...\n"
% (abs(amount), "top" if (amount > 0) else "bottom"))
if ditch_outliers:
blather(" (sans top and bottom outliers...)\n")
totals = {}
highest = (amount > 0)
if not highest:
amount *= -1
push_quiet(True)
target_playlists = playlist.get_contained_playlists(flat=True)
pop_quiet()
did = 0
blather('on playlist ')
for l in target_playlists:
did += 1
if (did - 1) % 10 == 0:
blather('%i ' % did, noflag=True)
playdensities = [t.get_play_density() for t in l.get_contained_tracks()]
playdensities.sort()
if ditch_outliers and len(playdensities) > ditch_outliers:
playdensities = playdensities[ditch_outliers:-1*ditch_outliers]
pl_sum = sum(playdensities)
totals[l] = pl_sum
blather('%i\n' % did, noflag=True)
pl = [(float(total)/len(playlist.get_contained_tracks()), playlist)
for playlist, total in totals.items()]
pl.sort(reverse=highest)
for i, p in pl[:amount]:
print ("%s: %f avg - %f total / %i tracks"
% (p.get_dotted_name(), i, totals[p],
len(p.get_contained_tracks())))
def first_timeout():
"""Gratuitous itunes appscript reference cycle to elicit a timeout.
It usually happens on some systems after just about 2**16 (= 2**2**2**2)
interactions, and sometimes substantially before that."""
import appscript
import sys
itunes = appscript.app('itunes')
t = itunes.tracks[1]
count = 0
class PassedLimit(Exception): pass
try:
while True:
count += 1
if count % 1000 == 0:
sys.stderr.write("%i\n" % count)
if count == 65530:
sys.stderr.write("threshold: %i\n" % count)
elif count > 66000:
raise PassedLimit
x = t.name.get()
except appscript.reference.CommandError:
sys.stderr.write("got a CommandError at count %i\n" % count)
raise
except PassedLimit:
sys.stderr.write("exceeded threshold without triggering the timeout\n")
if DEFAULTS_FILE:
import sys
from os import path
if DEFAULTS_FILE.find(path.sep) == 0:
thefile = DEFAULTS_FILE
else:
d, f = path.split(sys.argv[0])
thefile = path.join(d, DEFAULTS_FILE)
if path.exists(thefile):
defaults = open(thefile, 'r').read()
exec(defaults, globals())
if __name__ == "__main__":
usage = """usage: %prog [ options ]
for details: %prog --help"""
parser = optparse.OptionParser(usage=usage)
parser.add_option("-p", "--playlist", metavar="PLAYLIST",
default = SUBJECT_PLAYLIST,
help=("playlist to be synced, including sub-playlists"
+ " [default: '%s']" % SUBJECT_PLAYLIST))
parser.add_option("-d", "--directory", metavar="DIRECTORY-PATH",
default = TARGET_DIRECTORY_PATH,
help=("directory to be used as sync destination"
+ " [default: '%default']"))
parser.add_option("-c", "--create-failsafe",
dest="create_failsafe", action="store_true",
default = ALWAYS_CREATE_FAILSAFE,
help=("should we automatically create absent failsafe?"
+ " otherwise we'll prompt to create if necessary"
+ " [default: '%default']"))
parser.add_option("-s", "--stub-track", dest="stub",
metavar="STUB-TRACK-NAME",
default=USE_STUB_TRACK,
help=("identify shuffle-play lead track fodder"
" [default: '%default']"))
parser.add_option("--no-stub-track", dest="nostub", action="store_true",
default=False,
help="inhibit special stub track provision")
parser.add_option("-t", "--doctest", action="store_true",
default=False,
help="run only the module doctests")
parser.add_option("-q", "--quiet", action="store_true",
default=False,
help=("inhibit activity reports, including those"
"from --verbose and --dry-run"
+ " [default: '%default']"))
parser.add_option("-n", "--dry-run", action="store_true", dest="dryrun",
default=DRYRUN,
help=("do everything short of changing any files"
" and report a lot (except when --quiet is set);"
" target directory and failsafe are ignored"))
parser.add_option("-v", "--verbose", action="store_true",
default=VERBOSE,
help=("report more activity as it happens;"
" --quiet overriddes this"
+ " [default: '%default']"))
parser.add_option("-T", "--timeout-test", action="store_true",
default=False, dest="timeouttest",
help=("run only a test for a funky appscript timeout bug"
+ " [default: '%default']"))
parser.add_option("--track-stats", type="int",
default=False, dest="trackstats",
help=("only show top NUMBER track target playlist"
+ " enrollment statistics"))
parser.add_option("--play-stats", type="int",
default=False, dest="playstats",
help=("show top-played N (positive N) tracks or averaged"
+ " playlists (negative N)"))
parser.add_option("--play-stats-sans-outliers", type=int,
default=0, dest="trimplaystats",
help=("for --play-stats, remove top and bottom N outliers"
+ " [default '%default'"))
# MAYBE: --target-dir=
(options, args) = parser.parse_args()
DRYRUN = (options.dryrun or options.trackstats
or options.playstats or options.trimplaystats)
push_quiet(options.quiet)
VERBOSE = options.verbose
USE_STUB_TRACK = options.stub
subject = SUBJECT_PLAYLIST = options.playlist
directory = TARGET_DIRECTORY_PATH = options.directory
create_failsafe = options.create_failsafe
# In verbose mode only, we blather about some important settings:
blather(" dryrun: %s, verbose: %s,stub: '%s',\n"
" playlist: '%s', auto-create failsafe: %s\n"
" dir / failsafe: %s / %s\n"
% (DRYRUN, VERBOSE, USE_STUB_TRACK, subject, create_failsafe,
directory, TARGET_DIRECTORY_FAILSAFE_FILE),
marginal=True)
if options.doctest:
push_quiet(True)
import doctest
blather("running doctests\n", marginal=True)
doctest.testmod()
elif options.timeouttest:
first_timeout()
else:
PlaylistSyncManager(label=subject)
playlist = Playlist(label=subject)
if options.trackstats or options.playstats or options.trimplaystats:
if options.trackstats:
playlist = Playlist(label=subject)
PlaylistSyncManager(label=subject)
present_track_stats(playlist, options.trackstats)
if options.playstats or options.trimplaystats:
playlist = Playlist(label=subject)
PlaylistSyncManager(label=subject)
present_play_stats(playlist,
options.playstats or options.trimplaystats,
ditch_outliers=options.trimplaystats)
raise SystemExit
if not do_failsafe_check():
failsafe_path = os.path.join(TARGET_DIRECTORY_PATH,
TARGET_DIRECTORY_FAILSAFE_FILE)
raise EnvironmentError, (1, "Failsafe file is absent",
failsafe_path)
done = False
timeouts = 0
while not done:
try:
laststart = time.time()
playlist.sync_to_directory(directory)
syncmgr.tick(force=True)
done = True
except appscript.reference.CommandError:
# appscript default timeout is 60 seconds:
if time.time() - laststart < 75:
# failing too quickly, give up:
whoops("appscript timeout too soon, bailing out...\n")
raise
timeouts += 1
whoops("resuming after appscript timeout %i\n" % timeouts)
fm = PlayfolderManager(label=directory)
fm.remove_unused_tracks_and_playlists()
syncmgr.tick(identify_playlist_changes=True)
Creative Commons Attribution-NonCommercial-ShareAlike 4.0
Unported License