code.delx.au - notipod/blob - libnotipod.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 sys
   9 import urllib
  10
  11 from Foundation import *
  12
  13
  14 def read_plist(filename):
  15         try:
  16                 data = buffer(open(filename).read())
  17         except IOError:
  18                 return None
  19         plist, fmt, err = NSPropertyListSerialization.propertyListFromData_mutabilityOption_format_errorDescription_(data, NSPropertyListMutableContainers, None, None)
  20         if err is not None:
  21                 errStr = err.encode("utf-8")
  22                 err.release() # Doesn't follow Cocoa conventions for some reason
  23                 raise TypeError(errStr)
  24         return plist
  25
  26 class Playlist(NSObject):
  27         def init(self):
  28                 return self
  29
  30         def set(self, name, pid, ptype, tracks, parent):
  31                 self.name = name
  32                 self.pid = pid
  33                 self.ptype = ptype
  34                 self.children = []
  35                 self.tracks = tracks
  36                 self.parent = parent
  37                 if parent is not None:
  38                         parent.children.append(self)
  39
  40 class ITunesLibrary(NSObject):
  41         def load_(self, filename):
  42                 if filename is None:
  43                         filename = "~/Music/iTunes/iTunes Music Library.xml"
  44                 filename = os.path.expanduser(filename)
  45                 yield "Reading library..."
  46                 plist = read_plist(os.path.expanduser(filename))
  47                 self.folder = self.loc2name(plist["Music Folder"])
  48                 pl_tracks = plist["Tracks"]
  49                 pl_lookup = {}
  50                 self.playlists = []
  51                 for pl_playlist in plist["Playlists"]:
  52                         playlist = self.make_playlist(pl_playlist, pl_tracks, pl_lookup)
  53                         if not playlist:
  54                                 continue
  55                         yield "Read playlist: " + playlist.name
  56                         self.playlists.append(playlist)
  57                         pl_lookup[playlist.pid] = playlist
  58
  59         def loc2name(self, location):
  60                 return urllib.splithost(urllib.splittype(urllib.unquote(location))[1])[1]
  61
  62         def make_playlist(self, pl_playlist, pl_tracks, pl_lookup):
  63                 if int(pl_playlist.get("Master", 0)):
  64                         return
  65                 kind = int(pl_playlist.get("Distinguished Kind", -1))
  66                 if kind == 26:
  67                         # Don't do genius
  68                         return
  69
  70                 name = pl_playlist["Name"]
  71                 pid = pl_playlist["Playlist Persistent ID"]
  72                 if kind > 0:
  73                         ptype = {
  74                                 2: "movies",
  75                                 3: "tv-shows",
  76                                 4: "music",
  77                                 5: "books",
  78                                 10: "podcasts",
  79                                 19: "purchased",
  80                                 22: "itunes-dj",
  81                                 31: "itunes-u",
  82                         }.get(kind, "playlist")
  83                 elif pl_playlist.has_key("Smart Info"):
  84                         ptype = "smart-playlist"
  85                 elif int(pl_playlist.get("Folder", 0)):
  86                         ptype = "folder"
  87                 else:
  88                         ptype = "playlist"
  89
  90                 parent = None
  91                 try:
  92                         parent_pid = pl_playlist["Parent Persistent ID"]
  93                         parent = pl_lookup[parent_pid]
  94                 except KeyError:
  95                         pass
  96                 tracks = []
  97                 for item in pl_playlist.get("Playlist Items", []):
  98                         trackID = item["Track ID"]
  99                         filename = str(pl_tracks[str(trackID)]["Location"])
 100                         filename = self.loc2name(filename)
 101                         filename = filename.decode("utf-8")
 102                         if not filename.startswith(self.folder):
 103                                 logging.warn("Skipping: " + filename)
 104                                 continue
 105                         filename = strip_prefix(filename, self.folder)
 106                         tracks.append(filename)
 107                 playlist = Playlist.alloc().init()
 108                 playlist.set(name, pid, ptype, tracks, parent)
 109                 return playlist
 110
 111         def has_playlist_name(self, name):
 112                 for p in self.get_playlists():
 113                         if p.name == name:
 114                                 return True
 115                 return False
 116
 117         def get_playlist_name(self, name):
 118                 for playlist in self.get_playlists():
 119                         if playlist.name == name:
 120                                 return playlist
 121
 122         def get_playlist_pid(self, pid):
 123                 for playlist in self.get_playlists():
 124                         if playlist.pid == pid:
 125                                 return playlist
 126
 127         def get_playlists(self):
 128                 return self.playlists
 129
 130
 131 encoded_names = {}
 132 valid_chars = frozenset("\\/-_.() abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
 133 def encode_filename(filename):
 134         try:
 135                 return encoded_names[filename]
 136         except KeyError:
 137                 pass
 138         orig_filename = filename
 139         filename = filename.encode("ascii", "ignore")
 140         filename = "".join(c for c in filename if c in valid_chars)
 141         if filename in encoded_names:
 142                 a, b = os.path.splitext(filename)
 143                 a += "-dup"
 144                 filename = a + b
 145         encoded_names[orig_filename] = filename
 146         return filename
 147
 148 def strip_prefix(s, prefix):
 149         assert s.startswith(prefix)
 150         s = s[len(prefix):]
 151         if s.startswith("/"):
 152                 s = s[1:]
 153         return s
 154
 155 def mkdirhier(path):
 156         if os.path.isdir(path):
 157                 return
 158         paths = [path]
 159         while path != "/":
 160                 path = os.path.split(path)[0]
 161                 paths.append(path)
 162         for path in reversed(paths):
 163                 try:
 164                         os.mkdir(path)
 165                 except OSError:
 166                         pass
 167
 168 def export_m3u(dry_run, dest, path_prefix, playlist_name, files):
 169         if dry_run:
 170                 return
 171         if not path_prefix:
 172                 path_prefix = "../"
 173         playlist_file = os.path.join(dest, "-Playlists-", playlist_name) + ".m3u"
 174         playlist_file = encode_filename(playlist_file)
 175         mkdirhier(os.path.dirname(playlist_file))
 176         logging.info("Writing: " + playlist_file)
 177         f = open(playlist_file, "w")
 178         for filename in files:
 179                 if path_prefix.find("\\") > 0:
 180                         filename = filename.replace("/", "\\")
 181                 filename = encode_filename(filename)
 182                 f.write("%s%s\n" % (path_prefix, filename))
 183         f.close()
 184
 185 def sync(dry_run, source, dest, files_to_copy):
 186         join = os.path.join
 187
 188         logging.info("Calculating files to sync and deleting old files")
 189         source = source.encode("utf-8")
 190         dest = dest.encode("utf-8")
 191         filemap = {}
 192         class SyncFile(object): pass
 193         for f in files_to_copy:
 194                 sf = SyncFile()
 195                 sf.orig_filename = f.encode("utf-8")
 196                 sf.encoded_filename = encode_filename(f)
 197                 filemap[sf.encoded_filename.lower()] = sf
 198         files_to_copy = set(filemap)
 199
 200         if not os.path.isdir(dest):
 201                 raise OSError("No such file or directory: '%s'" % dest)
 202         for dirpath, dirnames, filenames in os.walk(dest):
 203                 full_dirpath = dirpath
 204                 dirpath = strip_prefix(dirpath, dest)
 205
 206                 for filename in filenames:
 207                         filename = join(dirpath, filename)
 208
 209                         # Whenever 'file' is deleted OSX will helpfully remove '._file'
 210                         if not os.path.exists(join(dest, filename)):
 211                                 continue
 212
 213                         if filename.lower() in files_to_copy:
 214                                 source_filename = filemap[filename.lower()].orig_filename
 215                                 sourcestat = os.stat(join(source, source_filename))
 216                                 deststat = os.stat(join(dest, filename))
 217                                 same_time = abs(sourcestat.st_mtime - deststat.st_mtime) < 5
 218                                 same_size = sourcestat.st_size == deststat.st_size
 219                                 if same_time and same_size:
 220                                         files_to_copy.remove(filename.lower())
 221                                         yield "Keep: " + filename
 222                                 else:
 223                                         yield "Update: " + filename
 224
 225                         elif not filename.startswith("-Playlists-"):
 226                                 yield "Delete: " + filename
 227                                 if not dry_run:
 228                                         os.unlink(join(dest, filename))
 229
 230                 if len(os.listdir(full_dirpath)) == 0:
 231                         yield "Delete: " + dirpath
 232                         if not dry_run:
 233                                 os.rmdir(full_dirpath)
 234
 235
 236         logging.info("Copying new files")
 237         files_to_copy = list(files_to_copy)
 238         files_to_copy.sort()
 239         for filename in files_to_copy:
 240                 yield "Copy: " + filemap[filename].orig_filename
 241                 if not dry_run:
 242                         source_file = join(source, filemap[filename].orig_filename)
 243                         dest_file = join(dest, filemap[filename].encoded_filename)
 244                         mkdirhier(os.path.dirname(dest_file))
 245                         shutil.copy2(source_file, dest_file)
 246
 247