code.delx.au - notipod/blob - libsyncitunes.py

   1 #!/usr/bin/env python
   2 # Copyright 2009 James Bunton <jamesbunton@fastmail.fm>
   3 # Licensed for distribution under the GPL version 2, check COPYING for details
   4
   5 import logging
   6 import os
   7 import shutil
   8 import urllib
   9
  10 from Foundation import *
  11
  12
  13 def read_plist(filename):
  14         try:
  15                 data = buffer(open(filename).read())
  16         except IOError:
  17                 return None
  18         plist, fmt, err = NSPropertyListSerialization.propertyListFromData_mutabilityOption_format_errorDescription_(data, NSPropertyListMutableContainers, None, None)
  19         if err is not None:
  20                 errStr = err.encode("utf-8")
  21                 err.release() # Doesn't follow Cocoa conventions for some reason
  22                 raise TypeError(errStr)
  23         return plist
  24
  25 class Playlist(NSObject):
  26         def init(self):
  27                 return self
  28
  29         def set(self, name, pid, tracks, parent):
  30                 self.name = name
  31                 self.pid = pid
  32                 self.children = []
  33                 self.tracks = tracks
  34                 self.parent = parent
  35                 if parent is not None:
  36                         parent.children.append(self)
  37
  38 class ITunesLibrary(NSObject):
  39         def init(self):
  40                 return self.initWithFilename_("~/Music/iTunes/iTunes Music Library.xml")
  41
  42         def initWithFilename_(self, filename):
  43                 filename = os.path.expanduser(filename)
  44                 plist = read_plist(os.path.expanduser(filename))
  45                 self.folder = self.loc2name(plist["Music Folder"])
  46                 pl_tracks = plist["Tracks"]
  47                 self.playlists = {}
  48                 for pl_playlist in plist["Playlists"]:
  49                         playlist = self.make_playlist(pl_playlist, pl_tracks)
  50                         self.playlists[playlist.pid] = playlist
  51                 return self
  52
  53         def loc2name(self, location):
  54                 return urllib.splithost(urllib.splittype(urllib.unquote(location))[1])[1]
  55
  56         def make_playlist(self, pl_playlist, pl_tracks):
  57                 name = pl_playlist["Name"]
  58                 pid = pl_playlist["Playlist Persistent ID"]
  59                 parent = None
  60                 try:
  61                         parent_pid = pl_playlist["Parent Persistent ID"]
  62                         parent = self.playlists.get(parent_pid)
  63                 except KeyError:
  64                         pass
  65                 tracks = []
  66                 for item in pl_playlist.get("Playlist Items", []):
  67                         trackID = item["Track ID"]
  68                         filename = str(pl_tracks[str(trackID)]["Location"])
  69                         filename = self.loc2name(filename)
  70                         filename = eval(repr(filename).lstrip("u")).decode("utf-8")
  71                         if not filename.startswith(self.folder):
  72                                 logging.warn("Skipping: " + filename)
  73                                 continue
  74                         filename = filename[len(self.folder):]
  75                         if filename.startswith("/"):
  76                                 filename = filename[1:]
  77                         tracks.append(filename)
  78                 playlist = Playlist.alloc().init()
  79                 playlist.set(name, pid, tracks, parent)
  80                 return playlist
  81
  82         def has_playlist_name(self, name):
  83                 for p in self.get_playlists():
  84                         if p.name == name:
  85                                 return True
  86                 return False
  87
  88         def get_playlist_name(self, name):
  89                 for playlist in self.get_playlists():
  90                         if playlist.name == name:
  91                                 return playlist
  92
  93         def get_playlist_pid(self, pid):
  94                 for playlist in self.get_playlists():
  95                         if playlist.pid == pid:
  96                                 return playlist
  97
  98         def get_playlists(self):
  99                 return self.playlists.values()
 100
 101         def outlineView_numberOfChildrenOfItem_(self, view, item):
 102                 if item == None:
 103                         return len(self.playlists)
 104                 else:
 105                         return 0
 106
 107         def outlineView_isItemExpandable_(self, view, item):
 108                 return False
 109
 110         def outlineView_child_ofItem_(self, view, index, item):
 111                 if item == None:
 112                         return self.playlists[index]
 113                 else:
 114                         return None
 115
 116         def outlineView_objectValueForTableColumn_byItem_(self, view, column, item):
 117                 return item.name
 118
 119
 120 def export_m3u(dry_run, dest, path_prefix, playlist_name, files):
 121         if dry_run:
 122                 return
 123         playlist_file = os.path.join(dest, playlist_name) + ".m3u"
 124         logging.info("Writing: " + playlist_file)
 125         f = open(playlist_file, "w")
 126         for filename in files:
 127                 if path_prefix.find("\\") > 0:
 128                         filename = filename.replace("/", "\\")
 129                 filename = filename.encode("utf-8")
 130                 f.write("%s%s\n" % (path_prefix, filename))
 131         f.close()
 132
 133 def strip_prefix(s, prefix):
 134         assert s.startswith(prefix)
 135         s = s[len(prefix):]
 136         if s.startswith("/"):
 137                 s = s[1:]
 138         return s
 139
 140 def mkdirhier(path):
 141         if os.path.isdir(path):
 142                 return
 143         paths = [path]
 144         while path != "/":
 145                 path = os.path.split(path)[0]
 146                 paths.append(path)
 147         for path in reversed(paths):
 148                 try:
 149                         os.mkdir(path)
 150                 except OSError:
 151                         pass
 152
 153 def sync(dry_run, source, dest, files):
 154         join = os.path.join
 155
 156         logging.info("Calculating files to sync and deleting old files")
 157         files = set(files)
 158         for dirpath, dirnames, filenames in os.walk(dest):
 159                 full_dirpath = dirpath
 160                 dirpath = strip_prefix(dirpath, dest)
 161
 162                 for filename in filenames:
 163                         filename = join(dirpath, filename).decode("utf-8")
 164
 165                         # Whenever 'file' is deleted OSX will helpfully remove '._file'
 166                         if not os.path.exists(join(dest, filename)):
 167                                 continue
 168
 169                         if filename in files:
 170                                 sourcestat = os.stat(join(source, filename))
 171                                 deststat = os.stat(join(dest, filename))
 172                                 same_time = abs(sourcestat.st_mtime - deststat.st_mtime) < 5
 173                                 same_size = sourcestat.st_size == deststat.st_size
 174                                 if same_time and same_size:
 175                                         files.remove(filename)
 176                                         logging.debug("keep: " + filename)
 177                                 else:
 178                                         logging.debug("update: " + filename)
 179
 180                         elif not filename.endswith(".m3u"):
 181                                 logging.debug("delete: " + filename)
 182                                 if not dry_run:
 183                                         os.unlink(join(dest, filename))
 184
 185                 if len(os.listdir(full_dirpath)) == 0:
 186                         logging.debug("rmdir: " + dirpath)
 187                         if not dry_run:
 188                                 os.rmdir(full_dirpath)
 189
 190
 191         logging.info("Copying new files")
 192         files = list(files)
 193         files.sort()
 194         for filename in files:
 195                 logging.debug("copy: " + filename)
 196                 if not dry_run:
 197                         mkdirhier(os.path.dirname(join(dest, filename)))
 198                         shutil.copy2(join(source, filename), join(dest, filename))
 199
 200