DataFlow CLI interface

dataflow.php

<?php

/*
error_reporting(E_ALL | E_STRICT);
Mage::setIsDeveloperMode(true);
ini_set('display_errors', 1);
*/

require_once 'abstract.php';

class Mage_Shell_DataFlow extends Mage_Shell_Abstract
{
    protected $_last_profile_exceptions;
    protected $_last_batch_exceptions;

    protected function printListLine($elements) {
        printf("%6s| %-40s| %-20s| %-20s\n", $elements[0], $elements[1], $elements[2], $elements[3]);
    }

    /**
     * @param $id_or_name string
     * @return Mage_Dataflow_Model_Profile|bool
     */
    protected function getDataFlowProfile($id_or_name) {
        $profile_collection = Mage::getModel('dataflow/profile')
                                      ->getCollection();

        foreach($profile_collection as $profile) {
            if ($profile->getId() == $id_or_name || $profile->getName() == $id_or_name)
                return $profile;
        }

        return false;
    }

    protected function doGetList() {
        $profile_collection = Mage::getModel('dataflow/profile')
                              ->getCollection();

        $this->printListLine(array("Id", "Name", "Direction", "Entity"));
        printf("-------------------------------------------------------------------------------\n");
        foreach($profile_collection as $profile) {
            $this->printListLine(array($profile->getId(),
                                       $profile->getName(),
                                       $profile->getDirection(),
                                       $profile->getEntityType()));
        }
    }

    protected function convertProfileExceptionsArray($exceptions) {
        if (! $exceptions)
            return array();

        $ret = array();
        foreach ($exceptions as $e) {

            switch ($e->getLevel()) {
                case Varien_Convert_Exception::FATAL:
                    $level = Varien_Convert_Exception::FATAL;
                    $level_str = 'Fatal';
                    break;
                case Varien_Convert_Exception::ERROR:
                    $level = Varien_Convert_Exception::ERROR;
                    $level_str = 'Error';
                    break;
                case Varien_Convert_Exception::WARNING:
                    $level = Varien_Convert_Exception::WARNING;
                    $level_str = 'Warning';
                    break;
                case Varien_Convert_Exception::NOTICE:
                    $level = Varien_Convert_Exception::NOTICE;
                    $level_str = 'Notice';
                    break;
            }

            $ret[] = new Varien_Object(array(
                "level"     => $level,
                "level_str" => $level_str,
                "message"   => $e->getMessage(),
                "exception" => $e
            ));
        }

        return $ret;
    }

    /**
     * @return bool
     */
    protected function inspectProfileExceptions() {
        if (! $this->_last_profile_exceptions)
            return false;

        fprintf(STDERR, "### Profile ###\n");

        $has_critical_error_happened = false;
        foreach ($this->_last_profile_exceptions as $e) {
            fprintf(STDERR, "**%s**: %s\n\n", $e->getLevelStr(), $e->getMessage());

            if (   $e->getLevel() == Varien_Convert_Exception::ERROR
                || $e->getLevel() == Varien_Convert_Exception::FATAL)
            {
                Mage::logException($e->getException());
                $has_critical_error_happened = true;
            }
        }

        return $has_critical_error_happened;
    }

    /**
     * @return bool
     */
    protected function inspectBatchExceptions() {
        if (! $this->_last_batch_exceptions)
            return false;

        fprintf(STDERR, "### Batch ###\n");

        foreach ($this->_last_batch_exceptions as $e) {
            fprintf(STDERR, "**%s**: %s\n", "Message", $e);
        }

        fprintf(STDERR, "\n");

        return false;
    }

    /**
     * @param Mage_Dataflow_Model_Profile $profile
     * @return Mage_Dataflow_Model_Batch
     */
    protected function doRunProfile(Mage_Dataflow_Model_Profile $profile) {
        $this->_last_profile_exceptions = false;

        $user = Mage::getModel('admin/user');
        $user->setUserId(0);

        Mage::getSingleton('admin/session')
            ->setUser($user);

        $profile = Mage::getModel("dataflow/profile")
                       ->load($profile->getId());

        Mage::register('current_convert_profile', $profile);
        $profile->run();

        $this->_last_profile_exceptions = $this->convertProfileExceptionsArray($profile->getExceptions());

        return Mage::getSingleton('dataflow/batch');
    }

    protected function doRunBatch(Mage_Dataflow_Model_Batch $batch) {
        if (! $batch->getId()) {
            throw new Exception("Uninitialized batch-model for import");
        }

        $adapter = Mage::getModel($batch->getAdapter());
        $adapter->setBatchParams($batch->getParams());

        $import_model = $batch->getBatchImportModel();
        $import_ids = $import_model->getIdCollection();

        $this->_last_batch_exceptions = array();
        $saved = 0;

        foreach ($import_ids as $id) {
            $import_model->load($id);
            if (! $import_model->getid()) {
                $this->_last_batch_exceptions[] = sprintf("Skipping undefined row");
                continue;
            }

            try {
                $import_data = $import_model->getBatchData();
                $adapter->saveRow($import_data);
            } catch (Exception $ex) {
                $this->_last_batch_exceptions[] = $ex->getMessage();
                continue;
            }

            $saved++;
        }

        if (method_exists($adapter, "'getEventPrefix'")) {
            /**
             * Event for process rules relations after products import
             */
            Mage::dispatchEvent($adapter->getEventPrefix() . '_finish_before', array(
                'adapter' => $adapter
            ));

            /**
             * Clear affected ids for adapter possible reuse
             */
            $adapter->clearAffectedEntityIds();
        }

        return $saved;
    }

    /**
     * @param $id_or_name string
     * @throws Exception
     */
    protected function doImport($id_or_name) {
        $profile = $this->getDataFlowProfile($id_or_name);
        if (! $profile)
            throw new Exception(sprintf("Unable to find profile '%s'", $id_or_name));

        if ($profile->getEntityType() && ! $profile->getDirection() == "import")
            throw new Exception(sprintf("Selected profile '%s' is not an import", $id_or_name));

        fprintf(STDERR, "\n");

        /** @var $batch Mage_Dataflow_Model_Batch_Import */
        $batch = $this->doRunProfile($profile);

        $has_critical_error_happened = $this->inspectProfileExceptions();
        if ($has_critical_error_happened)
            return;

        $n_saved = $this->doRunBatch($batch);
        $this->inspectBatchExceptions();
    }

    /**
     * @param $id_or_name string
     * @throws Exception
     */
    protected function doExport($id_or_name) {
        $profile = $this->getDataFlowProfile($id_or_name);
        if (! $profile)
            throw new Exception(sprintf("Unable to find profile '%s'", $id_or_name));

        if ($profile->getEntityType() && ! $profile->getDirection() == "export")
            throw new Exception(sprintf("Selected profile '%s' is not an export", $id_or_name));

        /** @var $batch Mage_Dataflow_Model_Batch_Export */
        $batch = $this->doRunProfile($profile);

        print_r($batch->getData());
    }

    public function run()
    {
        try {
            if ($this->getArg("list")){
                $this->doGetList();
            }
            else if ($this->getArg("import")) {
                $this->doImport($this->getArg("import"));
            }
            else if ($this->getArg("export")) {
                $this->doExport($this->getArg("export"));
            }
            else {
                echo $this->usageHelp();
            }
        } catch (Exception $ex) {
            fprintf(STDERR, "\n**Error** During execution: \n");
            fprintf(STDERR, "%s\n\n", $ex->__toString());
            Mage::logException($ex);
        }
    }

   /**
     * Retrieve Usage Help Message
     *
     */
    public function usageHelp()
    {
        return <<<USAGE

Usage:  php -f dataflow.php -- [options]

  --import <profile>            Run DataFlow import profile
  --export <profile>            Run DataFlow export profile
  --list                        List available profiles

  <profile>     Id or Name of a DataFlow-profile


USAGE;
    }
}

$shell = new Mage_Shell_DataFlow();
$shell->run();