mediagoblin/processing/__init__.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 # Use an ordered dict if we can.  If not, we'll just use a normal dict
  18 # later.
  19 try:
  20     from collections import OrderedDict
  21 except:
  22     OrderedDict = None
  23
  24 import logging
  25 import os
  26
  27 import six
  28
  29 from mediagoblin import mg_globals as mgg
  30 from mediagoblin.db.util import atomic_update
  31 from mediagoblin.db.models import MediaEntry
  32 from mediagoblin.tools.pluginapi import hook_handle
  33 from mediagoblin.tools.translate import lazy_pass_to_ugettext as _
  34
  35 _log = logging.getLogger(__name__)
  36
  37
  38 class ProgressCallback(object):
  39     def __init__(self, entry):
  40         self.entry = entry
  41
  42     def __call__(self, progress):
  43         if progress:
  44             self.entry.transcoding_progress = progress
  45             self.entry.save()
  46
  47
  48 def create_pub_filepath(entry, filename):
  49     return mgg.public_store.get_unique_filepath(
  50             ['media_entries',
  51              six.text_type(entry.id),
  52              filename])
  53
  54
  55 class FilenameBuilder(object):
  56     """Easily slice and dice filenames.
  57
  58     Initialize this class with an original file path, then use the fill()
  59     method to create new filenames based on the original.
  60
  61     """
  62     MAX_FILENAME_LENGTH = 255  # VFAT's maximum filename length
  63
  64     def __init__(self, path):
  65         """Initialize a builder from an original file path."""
  66         self.dirpath, self.basename = os.path.split(path)
  67         self.basename, self.ext = os.path.splitext(self.basename)
  68         self.ext = self.ext.lower()
  69
  70     def fill(self, fmtstr):
  71         """Build a new filename based on the original.
  72
  73         The fmtstr argument can include the following:
  74         {basename} -- the original basename, with the extension removed
  75         {ext} -- the original extension, always lowercase
  76
  77         If necessary, {basename} will be truncated so the filename does not
  78         exceed this class' MAX_FILENAME_LENGTH in length.
  79
  80         """
  81         basename_len = (self.MAX_FILENAME_LENGTH -
  82                         len(fmtstr.format(basename='', ext=self.ext)))
  83         return fmtstr.format(basename=self.basename[:basename_len],
  84                              ext=self.ext)
  85
  86
  87
  88 class MediaProcessor(object):
  89     """A particular processor for this media type.
  90
  91     While the ProcessingManager handles all types of MediaProcessing
  92     possible for a particular media type, a MediaProcessor can be
  93     thought of as a *particular* processing action for a media type.
  94     For example, you may have separate MediaProcessors for:
  95
  96     - initial_processing: the intial processing of a media
  97     - gen_thumb: generate a thumbnail
  98     - resize: resize an image
  99     - transcode: transcode a video
 100
 101     ... etc.
 102
 103     Some information on producing a new MediaProcessor for your media type:
 104
 105     - You *must* supply a name attribute.  This must be a class level
 106       attribute, and a string.  This will be used to determine the
 107       subcommand of your process
 108     - It's recommended that you supply a class level description
 109       attribute.
 110     - Supply a media_is_eligible classmethod.  This will be used to
 111       determine whether or not a media entry is eligible to use this
 112       processor type.  See the method documentation for details.
 113     - To give "./bin/gmg reprocess run" abilities to this media type,
 114       supply both gnerate_parser and parser_to_request classmethods.
 115     - The process method will be what actually processes your media.
 116     """
 117     # You MUST override this in the child MediaProcessor!
 118     name = None
 119
 120     # Optional, but will be used in various places to describe the
 121     # action this MediaProcessor provides
 122     description = None
 123
 124     def __init__(self, manager, entry):
 125         self.manager = manager
 126         self.entry = entry
 127         self.entry_orig_state = entry.state
 128
 129         # Should be initialized at time of processing, at least
 130         self.workbench = None
 131
 132     def __enter__(self):
 133         self.workbench = mgg.workbench_manager.create()
 134         return self
 135
 136     def __exit__(self, *args):
 137         self.workbench.destroy()
 138         self.workbench = None
 139
 140     # @with_workbench
 141     def process(self, **kwargs):
 142         """
 143         Actually process this media entry.
 144         """
 145         raise NotImplementedError
 146
 147     @classmethod
 148     def media_is_eligible(cls, entry=None, state=None):
 149         raise NotImplementedError
 150
 151     ###############################
 152     # Command line interface things
 153     ###############################
 154
 155     @classmethod
 156     def generate_parser(cls):
 157         raise NotImplementedError
 158
 159     @classmethod
 160     def args_to_request(cls, args):
 161         raise NotImplementedError
 162
 163     ##########################################
 164     # THE FUTURE: web interface things here :)
 165     ##########################################
 166
 167     #####################
 168     # Some common "steps"
 169     #####################
 170
 171     def delete_queue_file(self):
 172         # Remove queued media file from storage and database.
 173         # queued_filepath is in the task_id directory which should
 174         # be removed too, but fail if the directory is not empty to be on
 175         # the super-safe side.
 176         queued_filepath = self.entry.queued_media_file
 177         if queued_filepath:
 178             mgg.queue_store.delete_file(queued_filepath)      # rm file
 179             mgg.queue_store.delete_dir(queued_filepath[:-1])  # rm dir
 180             self.entry.queued_media_file = []
 181
 182
 183 class ProcessingKeyError(Exception): pass
 184 class ProcessorDoesNotExist(ProcessingKeyError): pass
 185 class ProcessorNotEligible(ProcessingKeyError): pass
 186 class ProcessingManagerDoesNotExist(ProcessingKeyError): pass
 187
 188
 189
 190 class ProcessingManager(object):
 191     """Manages all the processing actions available for a media type
 192
 193     Specific processing actions, MediaProcessor subclasses, are added
 194     to the ProcessingManager.
 195     """
 196     def __init__(self):
 197         # Dict of all MediaProcessors of this media type
 198         if OrderedDict is not None:
 199             self.processors = OrderedDict()
 200         else:
 201             self.processors = {}
 202
 203     def add_processor(self, processor):
 204         """
 205         Add a processor class to this media type
 206         """
 207         name = processor.name
 208         if name is None:
 209             raise AttributeError("Processor class's .name attribute not set")
 210
 211         self.processors[name] = processor
 212
 213     def list_eligible_processors(self, entry):
 214         """
 215         List all processors that this media entry is eligible to be processed
 216         for.
 217         """
 218         return [
 219             processor
 220             for processor in self.processors.values()
 221             if processor.media_is_eligible(entry=entry)]
 222
 223     def list_all_processors_by_state(self, state):
 224         """
 225         List all processors that this media state is eligible to be processed
 226         for.
 227         """
 228         return [
 229             processor
 230             for processor in self.processors.values()
 231             if processor.media_is_eligible(state=state)]
 232
 233
 234     def list_all_processors(self):
 235         return self.processors.values()
 236
 237     def gen_process_request_via_cli(self, subparser):
 238         # Got to figure out what actually goes here before I can write this properly
 239         pass
 240
 241     def get_processor(self, key, entry=None):
 242         """
 243         Get the processor with this key.
 244
 245         If entry supplied, make sure this entry is actually compatible;
 246         otherwise raise error.
 247         """
 248         try:
 249             processor = self.processors[key]
 250         except KeyError:
 251             raise ProcessorDoesNotExist(
 252                 "'%s' processor does not exist for this media type" % key)
 253
 254         if entry and not processor.media_is_eligible(entry):
 255             raise ProcessorNotEligible(
 256                 "This entry is not eligible for processor with name '%s'" % key)
 257
 258         return processor
 259
 260     def workflow(self, entry_id, feed_url, reprocess_action, reprocess_info=None):
 261         """
 262         Returns the Celery command needed to proceed with media processing
 263         *This method has to be implemented in all media types*
 264         """
 265         raise NotImplementedError
 266
 267
 268 def request_from_args(args, which_args):
 269     """
 270     Generate a request from the values of some argparse parsed args
 271     """
 272     request = {}
 273     for arg in which_args:
 274         request[arg] = getattr(args, arg)
 275
 276     return request
 277
 278
 279 class MediaEntryNotFound(Exception): pass
 280
 281
 282 def get_processing_manager_for_type(media_type):
 283     """
 284     Get the appropriate media manager for this type
 285     """
 286     manager_class = hook_handle(('reprocess_manager', media_type))
 287     if not manager_class:
 288         raise ProcessingManagerDoesNotExist(
 289             "A processing manager does not exist for {0}".format(media_type))
 290     manager = manager_class()
 291
 292     return manager
 293
 294
 295 def get_entry_and_processing_manager(media_id):
 296     """
 297     Get a MediaEntry, its media type, and its manager all in one go.
 298
 299     Returns a tuple of: `(entry, media_type, media_manager)`
 300     """
 301     entry = MediaEntry.query.filter_by(id=media_id).first()
 302     if entry is None:
 303         raise MediaEntryNotFound("Can't find media with id '%s'" % media_id)
 304
 305     manager = get_processing_manager_for_type(entry.media_type)
 306
 307     return entry, manager
 308
 309
 310 def mark_entry_failed(entry_id, exc):
 311     """
 312     Mark a media entry as having failed in its conversion.
 313
 314     Uses the exception that was raised to mark more information.  If
 315     the exception is a derivative of BaseProcessingFail then we can
 316     store extra information that can be useful for users telling them
 317     why their media failed to process.
 318
 319     :param entry_id: The id of the media entry
 320     :param exc: An instance of BaseProcessingFail
 321
 322     """
 323     # Was this a BaseProcessingFail?  In other words, was this a
 324     # type of error that we know how to handle?
 325     if isinstance(exc, BaseProcessingFail):
 326         # Looks like yes, so record information about that failure and any
 327         # metadata the user might have supplied.
 328         atomic_update(mgg.database.MediaEntry,
 329             {'id': entry_id},
 330             {u'state': u'failed',
 331              u'fail_error': six.text_type(exc.exception_path),
 332              u'fail_metadata': exc.metadata})
 333     else:
 334         _log.warn("No idea what happened here, but it failed: %r", exc)
 335         # Looks like no, let's record it so that admin could ask us about the
 336         # reason
 337         atomic_update(mgg.database.MediaEntry,
 338             {'id': entry_id},
 339             {u'state': u'failed',
 340              u'fail_error': u'Unhandled exception: {0}'.format(
 341                  six.text_type(exc)),
 342              u'fail_metadata': {}})
 343
 344
 345 def get_process_filename(entry, workbench, acceptable_files):
 346     """
 347     Try and get the queued file if available, otherwise return the first file
 348     in the acceptable_files that we have.
 349
 350     If no acceptable_files, raise ProcessFileNotFound
 351     """
 352     if entry.queued_media_file:
 353         filepath = entry.queued_media_file
 354         storage = mgg.queue_store
 355     else:
 356         for keyname in acceptable_files:
 357             if entry.media_files.get(keyname):
 358                 filepath = entry.media_files[keyname]
 359                 storage = mgg.public_store
 360                 break
 361
 362     if not filepath:
 363         raise ProcessFileNotFound()
 364
 365     filename = workbench.localized_file(
 366         storage, filepath,
 367         'source')
 368
 369     if not os.path.exists(filename):
 370         raise ProcessFileNotFound()
 371
 372     return filename
 373
 374
 375 def store_public(entry, keyname, local_file, target_name=None,
 376                  delete_if_exists=True):
 377     if target_name is None:
 378         target_name = os.path.basename(local_file)
 379     target_filepath = create_pub_filepath(entry, target_name)
 380
 381     if keyname in entry.media_files:
 382         _log.warn("store_public: keyname %r already used for file %r, "
 383                   "replacing with %r", keyname,
 384                   entry.media_files[keyname], target_filepath)
 385         if delete_if_exists:
 386             mgg.public_store.delete_file(entry.media_files[keyname])
 387     try:
 388         mgg.public_store.copy_local_to_storage(local_file, target_filepath)
 389     except Exception as e:
 390         _log.error(u'Exception happened: {0}'.format(e))
 391         raise PublicStoreFail(keyname=keyname)
 392     # raise an error if the file failed to copy
 393     if not mgg.public_store.file_exists(target_filepath):
 394         raise PublicStoreFail(keyname=keyname)
 395
 396     entry.media_files[keyname] = target_filepath
 397
 398
 399 def copy_original(entry, orig_filename, target_name, keyname=u"original"):
 400     store_public(entry, keyname, orig_filename, target_name)
 401
 402
 403 class BaseProcessingFail(Exception):
 404     """
 405     Base exception that all other processing failure messages should
 406     subclass from.
 407
 408     You shouldn't call this itself; instead you should subclass it
 409     and provide the exception_path and general_message applicable to
 410     this error.
 411     """
 412     general_message = u''
 413
 414     @property
 415     def exception_path(self):
 416         return u"%s:%s" % (
 417             self.__class__.__module__, self.__class__.__name__)
 418
 419     def __init__(self, message=None, **metadata):
 420         if message is not None:
 421             super(BaseProcessingFail, self).__init__(message)
 422             metadata['message'] = message
 423         self.metadata = metadata
 424
 425 class BadMediaFail(BaseProcessingFail):
 426     """
 427     Error that should be raised when an inappropriate file was given
 428     for the media type specified.
 429     """
 430     general_message = _(u'Invalid file given for media type.')
 431
 432
 433 class PublicStoreFail(BaseProcessingFail):
 434     """
 435     Error that should be raised when copying to public store fails
 436     """
 437     general_message = _('Copying to public storage failed.')
 438
 439
 440 class ProcessFileNotFound(BaseProcessingFail):
 441     """
 442     Error that should be raised when an acceptable file for processing
 443     is not found.
 444     """
 445     general_message = _(u'An acceptable processing file was not found')