view piecrust/processing/less.py @ 215:a47580a0955b

bake: Better error handling for the processing pipeline. Pipeline jobs now keep track of whether they've seen any errors. This is aggregated into an overall "success" flag for the processing record. Also, jobs keep going as long as there's no critical (i.e. internal) failure happening. Errors raised by processors are also better tracked: the actual processor that failed, along with the input file, are tracks in the processing record. The `bake` command returns a failure exit code if processing saw any error.
author Ludovic Chabant <ludovic@chabant.com>
date Sat, 31 Jan 2015 17:08:02 -0800
parents b4724e577a8c
children f82262f59600
line wrap: on
line source

import os
import os.path
import json
import hashlib
import logging
import platform
import subprocess
from piecrust.processing.base import SimpleFileProcessor
from piecrust.processing.tree import FORCE_BUILD


logger = logging.getLogger(__name__)


class LessProcessor(SimpleFileProcessor):
    PROCESSOR_NAME = 'less'

    def __init__(self):
        super(LessProcessor, self).__init__({'less': 'css'})
        self._conf = None
        self._map_dir = None

    def onPipelineStart(self, pipeline):
        self._map_dir = os.path.join(pipeline.tmp_dir, 'less')
        if not os.path.isdir(self._map_dir):
            os.makedirs(self._map_dir)

    def getDependencies(self, path):
        map_path = self._getMapPath(path)
        try:
            with open(map_path, 'r') as f:
                dep_map = json.load(f)

            # Check the version, since the `sources` list has changed
            # meanings over time.
            if dep_map.get('version') != 3:
                logger.warning("Unknown LESS map version. Force rebuilding.")
                return FORCE_BUILD

            # Get the sources, but make all paths absolute.
            sources = dep_map.get('sources')
            path_dir = os.path.dirname(path)
            def _makeAbs(p):
                return os.path.join(path_dir, p)
            deps = list(map(_makeAbs, sources))
            return [map_path] + deps
        except IOError:
            # Map file not found... rebuild.
            logger.debug("No map file found for LESS file '%s' at '%s'. "
                         "Rebuilding" % (path, map_path))
            return FORCE_BUILD

    def _doProcess(self, in_path, out_path):
        self._ensureInitialized()

        map_path = self._getMapPath(in_path)
        map_path = os.path.relpath(map_path)
        args = [self._conf['bin'], '--source-map=%s' % map_path]
        args += self._conf['options']
        args.append(in_path)
        args.append(out_path)
        logger.debug("Processing LESS file: %s" % args)

        # On Windows, we need to run the process in a shell environment
        # otherwise it looks like `PATH` isn't taken into account.
        shell = (platform.system() == 'Windows')
        try:
            retcode = subprocess.call(args, shell=shell)
        except FileNotFoundError as ex:
            logger.error("Tried running LESS processor with command: %s" %
                         args)
            raise Exception("Error running LESS processor. "
                            "Did you install it?") from ex
        if retcode != 0:
            raise Exception("Error occured in LESS compiler. Please check "
                            "log messages above for more information.")
        return True

    def _ensureInitialized(self):
        if self._conf is not None:
            return

        self._conf = self.app.config.get('less') or {}
        self._conf.setdefault('bin', 'lessc')
        self._conf.setdefault('options', ['--compress'])
        if not isinstance(self._conf['options'], list):
            raise Exception("The `less/options` configuration setting "
                            "must be an array of arguments.")

    def _getMapPath(self, path):
        map_name = "%s_%s.map" % (
                os.path.basename(path),
                hashlib.md5(path.encode('utf8')).hexdigest())
        map_path = os.path.join(self._map_dir, map_name)
        return map_path