audiomangler/cli.py

   1 # -*- coding: utf-8 -*-
   2 ###########################################################################
   3 #    Copyright (C) 2008 by Andrew Mahone
   4 #    <andrew.mahone@gmail.com>
   5 #
   6 # Copyright: See COPYING file that comes with this distribution
   7 #
   8 ###########################################################################
   9 import sys
  10 import getopt
  11 import shutil
  12 import os
  13 import os.path
  14 import errno
  15 from functools import wraps
  16 from audiomangler.config import Config
  17 from audiomangler import util
  18 from audiomangler.codecs import sync_sets, get_codec
  19 from audiomangler.scanner import scan
  20 from audiomangler.task import PoolTask
  21 from audiomangler.logging import *
  22
  23 def parse_options(options = []):
  24     def decorator(f):
  25         @wraps(f)
  26         def proxy(*args):
  27             if not args:
  28                 if len(sys.argv) == 1:
  29                     print_usage(options)
  30                     sys.exit(0)
  31                 else:
  32                     args = sys.argv[1:]
  33             name_map = {}
  34             s_opts = []
  35             l_opts = []
  36             for (s_opt, l_opt, name, desc) in options:
  37                 if s_opt:
  38                     name_map['-'+s_opt.rstrip(':')] = name
  39                     s_opts.append(s_opt)
  40                 if l_opt:
  41                     name_map['--'+l_opt.rstrip('=')] = name
  42                     l_opts.append(l_opt)
  43             s_opts = ''.join(s_opts)
  44             try:
  45                 (opts, args) = getopt.getopt(args, s_opts, l_opts)
  46             except getopt.GetoptError:
  47                 import traceback
  48                 traceback.print_exception(*sys.exc_info())
  49                 print_usage(options)
  50                 sys.exit(0)
  51             for k, v in opts:
  52                 k = name_map[k]
  53                 Config[k] = v
  54             f(*args)
  55         return proxy
  56     return decorator
  57
  58 def print_usage(opts):
  59     print """usage:
  60     %s [options] [files or directories to process]
  61
  62 options:""" % sys.argv[0]
  63     for short, long_, name, desc in opts:
  64         print "    -%s, --%-10s  %s" % (short.rstrip(':'), long_.rstrip('='), desc)
  65
  66 common_opts = (
  67     ('b:', 'base=', 'base', 'base directory for target files'),
  68     ('p:', 'profile=', 'profile', 'profile to load settings from'),
  69     ('f:', 'filename=', 'filename', 'format for target filenames'),
  70 )
  71
  72 @parse_options(common_opts)
  73 def rename(*args):
  74     dir_list = scan(args)[1]
  75     util.test_splits(dir_list)
  76     onsplit = Config['onsplit']
  77     for (dir_, files) in dir_list.items():
  78         dir_p = util.fsdecode(dir_)
  79         msg(consoleformat=u"from dir %(dir_p)s:",
  80             format="enter: %(dir_)r", dir_=dir_, dir_p=dir_p, loglevel=INFO)
  81         dstdirs = set()
  82         moves = []
  83         for file_ in files:
  84             src = file_.filename
  85             dst = util.fsencode(file_.format())
  86             src_p = util.fsdecode(src)
  87             dst_p = util.fsdecode(dst)
  88             if src == dst:
  89                 msg(consoleformat=u"  skipping %(src_p)s, already named correctly",
  90                     format="skip: %(src)r",
  91                     src_p=srcp_p, src=src, loglevel=INFO)
  92                 continue
  93             dstdir = os.path.split(dst)[0]
  94             if dstdir not in dstdirs and dstdir != dir_:
  95                 try:
  96                     os.makedirs(dstdir)
  97                 except OSError, e:
  98                     if e.errno != errno.EEXIST or not os.path.isdir(dstdir):
  99                         raise
 100                 dstdirs.add(dstdir)
 101             msg(consoleformat=u"  %(src_p)s -> %(dst_p)s",
 102                 format="move: %(src)r, %(dst)r",
 103                 src=src, dst=dst, src_p=src_p, dst_p=dst_p, loglevel=INFO)
 104             util.move(src, dst)
 105         if len(dstdirs) == 1:
 106             dstdir = dstdirs.pop()
 107             for file_ in os.listdir(dir_):
 108                 src = os.path.join(dir_, file_)
 109                 dst = os.path.join(dstdir, file_)
 110                 src_p = util.fsdecode(src)
 111                 dst_p = util.fsdecode(dst)
 112                 msg(consoleformat=u"  %(src_p)s -> %(dst_p)s",
 113                     format="move: %(src)r, %(dst)r", src=src, dst=dst,
 114                     src_p=src_p, dst_p=dst_p, loglevel=INFO)
 115                 util.move(src, dst)
 116             while len(os.listdir(dir_)) == 0:
 117                 dir_p = util.fsdecode(dir_)
 118                 msg(consoleformat=u"  remove empty directory: %(dir_p)s",
 119                     format="rmdir: %(dir_)r",
 120                     dir_=dir_, dir_p=dir_p, loglevel=INFO)
 121                 try:
 122                     os.rmdir(dir_)
 123                 except Exception:
 124                     break
 125                 newdir = os.path.split(dir_)[0]
 126                 if newdir != dir_:
 127                     dir_ = newdir
 128                 else:
 129                     break
 130         else:
 131             if onsplit == 'warn':
 132                 msg(consoleformat=u"WARNING: tracks in %(dir_p)s were placed in different directories, other files may be left in the source directory",
 133                     format="split: %(dir_)r",
 134                     dir_=dir_, dir_p=dir_p, loglevel=WARNING)
 135
 136 @parse_options(common_opts + (
 137         ('t:', 'type=', 'type', 'type of audio to encode to'),
 138         ('s:', 'preset=', 'preset', 'codec preset to use'),
 139         ('e:', 'encopts=', 'encopts', 'encoder options to use'),
 140         ('j:', 'jobs=', 'jobs', 'number of jobs to run'),
 141    )
 142 )
 143 def sync(*args):
 144     (album_list, dir_list) = scan(args)[:2]
 145     targettids = scan(Config['base'])[2]
 146     sync_sets(album_list.values(), targettids)
 147
 148 def replaygain_task_generator(album_list):
 149     for key, album in album_list.items():
 150         profiles = set()
 151         for track in album:
 152             profiles.add((
 153                getattr(track, 'type_', None),
 154                getattr(getattr(track, 'info', None), 'sample_rate', None),
 155                getattr(getattr(track, 'info', None), 'channels', None)
 156             ))
 157         if len(profiles) != 1:
 158             continue
 159         profile = profiles.pop()
 160         dir_ = album[0].meta.flat()['dir']
 161         if profile[1] not in (8000, 11025, 12000, 16000, 22050, 24, 32, 44100, 48000):
 162             print "invalid bitrate for %s" % dir_
 163             continue
 164         codec = get_codec(profile[0])
 165         if not codec or not codec.has_replaygain:
 166             print "replaygain not supported for %s" % dir_
 167             continue
 168         if reduce(lambda x, y: x and y.has_replaygain(), album, True):
 169             print "all tracks have replaygain for %s" % dir_
 170             continue
 171         msg(consoleformat=u"Adding replaygain values to %(albumtitle)s",
 172             format="rg: %(tracks)r",
 173             albumtitle=album[0].meta.flat().get('album', '[unknown]'), tracks=tuple(t.filename for t in album))
 174         yield codec.add_replaygain([t.filename for t in album])
 175
 176 @parse_options(common_opts[:2] + (
 177         ('j:', 'jobs=', 'jobs', 'number of jobs to run'),
 178     )
 179 )
 180 def replaygain(*args):
 181     if not args:
 182         args = (Config['base'], )
 183     (album_list) = scan(args)[0]
 184     PoolTask(replaygain_task_generator(album_list)).run()
 185
 186 __all__ = []