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