Configurable media export and file synchronisation

README.md

@@ -13,5 +13,65 @@ Both of them have as final steps: Generate metadata, Zip files, send to Snowio u
 
 You can install this bundle via composer.
 ```
-composer require snowio/akeneo-bundle dev-upgrade/pim-1.7
-```
+composer require snowio/akeneo-bundle
+```
+
+### Configure threshold check step
+
+`Snowio\Bundle\CsvConnectorBundle\Step\CheckThresholdsStep` has an injectable export threshold, and checks this against the read count of the previous step.
+
+Define the class as a parameter:
+```
+parameters:
+   ...
+   snowio_connector.step.check_thresholds.class: Snowio\Bundle\CsvConnectorBundle\Step\CheckThresholdsStep
+```
+
+Create services for this class:
+```
+services:
+   ...
+   snowio_connector.step.check_threshold.products:
+      class: '%snowio_connector.step.check_thresholds.class%'
+      arguments:
+         - 'check_thresholds'
+         - '@event_dispatcher'
+         - '@akeneo_batch.job_repository'
+         - '%minimum_products_export%'
+
+   snowio_connector.step.check_threshold.attributes:
+      class: '%snowio_connector.step.check_thresholds.class%'
+      arguments:
+         - 'check_thresholds'
+         - '@event_dispatcher'
+         - '@akeneo_batch.job_repository'
+         - '%minimum_attributes_export%'
+```
+
+You need to inject the thresholds (bottom parameter) - these should be referenced by variables in `parameters.yml`.
+
+Add your services after the steps you want to check, e.g.:
+
+<pre><code>
+services:
+   ...
+   snowio_connector.job.full_export:
+       class: '%pim_connector.job.simple_job.class%'
+       arguments:
+           - '%snowio_connector.job_name.full_export%'
+           - '@event_dispatcher'
+           - '@akeneo_batch.job_repository'
+           -
+               - '@snowio_connector.step.csv_product.export'
+               <b>- '@snowio_connector.step.check_threshold.products'</b>
+               - '@snowio_connector.step.csv_variant_group.export'
+               - '@snowio_connector.step.csv_category.export'
+               - '@snowio_connector.step.csv_attribute.export'
+               <b>- '@snowio_connector.step.check_threshold.attributes'</b>
+               - '@snowio_connector.step.csv_attribute_option.export'
+               - '@snowio_connector.step.csv_family.export'
+               - '@snowio_connector.step.metadata'
+               - '@snowio_connector.step.archive'
+               - '@snowio_connector.step.media_export'
+               - '@snowio_connector.step.post'
+<pre><code>

Resources/config/jobs.yml

@@ -12,13 +12,16 @@ services:
             - '@akeneo_batch.job_repository'
             -
                 - '@snowio_connector.step.csv_product.export'
+                - '@snowio_connector.step.check_threshold.products'
                 - '@snowio_connector.step.csv_variant_group.export'
                 - '@snowio_connector.step.csv_category.export'
                 - '@snowio_connector.step.csv_attribute.export'
+                - '@snowio_connector.step.check_threshold.attributes'
                 - '@snowio_connector.step.csv_attribute_option.export'
                 - '@snowio_connector.step.csv_family.export'
                 - '@snowio_connector.step.metadata'
                 - '@snowio_connector.step.archive'
+                - '@snowio_connector.step.media_export'
                 - '@snowio_connector.step.post'
         tags:
             - { name: akeneo_batch.job, connector: '%snowio_connector.connector_name.csv%', type: '%pim_connector.job.export_type%' }

Resources/config/steps.yml

@@ -4,6 +4,8 @@ parameters:
     snowio_connector.step.metadata.class:  Snowio\Bundle\CsvConnectorBundle\Step\MetadataStep
     guzzlehttp.client.class: GuzzleHttp\Client
     snowio_connector.step.archive.zip.class: ZipArchive
+    snowio_connector.step.media_export.class: Snowio\Bundle\CsvConnectorBundle\Step\MediaExportStep
+    snowio_connector.step.check_thresholds.class: Snowio\Bundle\CsvConnectorBundle\Step\CheckThresholdsStep
 
 services:
     guzzlehttp.client:
@@ -20,6 +22,15 @@ services:
             - '@akeneo_batch.job_repository'
             - '@snowio_connector.step.archive.zip'
 
+    snowio_connector.step.media_export:
+        class: '%snowio_connector.step.media_export.class%'
+        arguments:
+            - 'media_export'
+            - '@event_dispatcher'
+            - '@akeneo_batch.job_repository'
+            - '%media_export_user%@%media_export_host%:%media_export_directory%'
+            - '%kernel.root_dir%/logs/%media_export_log_filename%'
+
     snowio_connector.step.post:
         class: '%snowio_connector.step.post.class%'
         arguments:
@@ -35,6 +46,22 @@ services:
             - '@event_dispatcher'
             - '@akeneo_batch.job_repository'
 
+    snowio_connector.step.check_threshold.products:
+        class: '%snowio_connector.step.check_thresholds.class%'
+        arguments:
+            - 'check_thresholds'
+            - '@event_dispatcher'
+            - '@akeneo_batch.job_repository'
+            - '%minimum_products_export%'
+
+    snowio_connector.step.check_threshold.attributes:
+        class: '%snowio_connector.step.check_thresholds.class%'
+        arguments:
+            - 'check_thresholds'
+            - '@event_dispatcher'
+            - '@akeneo_batch.job_repository'
+            - '%minimum_attributes_export%'
+
     # Overwrite service definition to set step name (arg 0) and writer
     snowio_connector.step.csv_product.export:
         class: '%pim_connector.step.item_step.class%'

Resources/translations/messages.en.yml

@@ -8,6 +8,7 @@ batch_jobs:
         archive.label: "Archive Step"
         metadata.label: "Metadata Step"
         post.label: "Post Step"
+        media_export.label: "Media Export Step"
     full_export:
         label: "Full Export (products, variant groups, categories, attributes, attribute options, and families)"
         product.label: "Product Export Step"
@@ -19,6 +20,8 @@ batch_jobs:
         archive.label: "Archive Step"
         metadata.label: "Metadata Step"
         post.label: "Post Step"
+        media_export.label: "Media Export Step"
+        check_thresholds.label: "Check Thresholds Step"
 
 job_execution:
     summary:
@@ -27,3 +30,6 @@ job_execution:
         response_body: Body
         zip_location: Zip file
         metadata_location: Metadata file
+        export_location: Export location
+        log_file: Log file
+        minimum_threshold: Minimum threshold

Step/CheckThresholdsStep.php

@@ -0,0 +1,95 @@
+<?php
+
+namespace Snowio\Bundle\CsvConnectorBundle\Step;
+
+use Akeneo\Component\Batch\Job\JobRepositoryInterface;
+use Akeneo\Component\Batch\Model\StepExecution;
+use Akeneo\Component\Batch\Step\AbstractStep;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+
+/**
+ * Use this class to check export threshold has been met in previous step
+ * Create a service, inject the relevant threshold, and configure the job so this comes after the step you want to check
+ */
+class CheckThresholdsStep extends AbstractStep
+{
+    /** @var int */
+    private $minimumExportThreshold;
+
+    /**
+     * CheckThresholdsStep constructor.
+     * @param string $name
+     * @param EventDispatcherInterface $eventDispatcher
+     * @param JobRepositoryInterface $jobRepository
+     * @param $minimumExportThreshold
+     */
+    public function __construct(
+        $name,
+        EventDispatcherInterface $eventDispatcher,
+        JobRepositoryInterface $jobRepository,
+        $minimumExportThreshold
+    ) {
+        parent::__construct($name, $eventDispatcher, $jobRepository);
+        $this->minimumExportThreshold = (int)$minimumExportThreshold;
+    }
+
+    /**
+     * Extension point for subclasses to execute business logic. Subclasses should set the {@link ExitStatus} on the
+     * {@link StepExecution} before returning.
+     *
+     * Do not catch exception here. It will be correctly handled by the execute() method.
+     *
+     * @param StepExecution $stepExecution the current step context
+     *
+     * @throws \Exception
+     */
+    protected function doExecute(StepExecution $stepExecution)
+    {
+        $previousStepExecution = $this->getPreviousStepExecution($stepExecution);
+
+        $stepExecution->addSummaryInfo(
+            'minimum_threshold',
+            sprintf('%s (%s)', $this->minimumExportThreshold, $previousStepExecution->getStepName())
+        );
+
+        if ($this->minimumExportThreshold > 0 && !$this->doesExportCountMeetThreshold($previousStepExecution)) {
+            throw new \Exception(
+                sprintf(
+                    'Error - attempted to export less than the minimum threshold (step: %s/threshold: %s).',
+                    $previousStepExecution->getStepName(),
+                    $this->minimumExportThreshold
+                )
+            );
+        }
+    }
+
+    /**
+     * @param StepExecution $stepExecution
+     * @return bool
+     * @author James Pollard <jp@amp.co>
+     */
+    private function doesExportCountMeetThreshold(StepExecution $stepExecution)
+    {
+        return $stepExecution->getSummaryInfo('read') >= $this->minimumExportThreshold;
+    }
+
+    /**
+     * @param StepExecution $stepExecution
+     * @return StepExecution
+     * @throws \Exception
+     * @author James Pollard <jp@amp.co>
+     */
+    private function getPreviousStepExecution(StepExecution $stepExecution)
+    {
+        $stepExecutions = $stepExecution->getJobExecution()->getStepExecutions()->toArray();
+        // set array pointer to last element i.e. the current execution
+        end($stepExecutions);
+        $previousStepExecution = prev($stepExecutions);
+
+        if (!($previousStepExecution instanceof StepExecution)) {
+            throw new \Exception('Error during threshold check step - previous execution step was not found.');
+        }
+
+        return $previousStepExecution;
+    }
+}

Step/MediaExportStep.php

@@ -0,0 +1,139 @@
+<?php
+
+namespace Snowio\Bundle\CsvConnectorBundle\Step;
+
+use Akeneo\Component\Batch\Job\JobRepositoryInterface;
+use Akeneo\Component\Batch\Model\StepExecution;
+use Akeneo\Component\Batch\Step\AbstractStep;
+use Akeneo\Component\FileStorage\Exception\FileTransferException;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Symfony\Component\Filesystem\Exception\FileNotFoundException;
+
+/**
+ * This class rsyncs media files to a configurable location and logs some info a separate log file
+ * Forcing reconnect to MySQL was looked into as part of this, which is not necessary:
+ * @see \Akeneo\Bundle\BatchBundle\Job\DoctrineJobRepository::updateStepExecution
+ */
+class MediaExportStep extends AbstractStep
+{
+    /** @var string */
+    protected $exportLocation;
+
+    /** @var string */
+    private $logFile;
+
+    /**
+     * MediaExportStep constructor.
+     * @param string $name
+     * @param EventDispatcherInterface $eventDispatcher
+     * @param JobRepositoryInterface $jobRepository
+     * @param $exportLocation
+     * @param $logFile
+     */
+    public function __construct(
+        $name,
+        EventDispatcherInterface $eventDispatcher,
+        JobRepositoryInterface $jobRepository,
+        $exportLocation,
+        $logFile
+    ) {
+        parent::__construct($name, $eventDispatcher, $jobRepository);
+        $this->exportLocation = $exportLocation;
+        $this->logFile = $logFile;
+    }
+
+    /**
+     * Extension point for subclasses to execute business logic. Subclasses should set the {@link ExitStatus} on the
+     * {@link StepExecution} before returning.
+     *
+     * Do not catch exception here. It will be correctly handled by the execute() method.
+     *
+     * @param StepExecution $stepExecution the current step context
+     *
+     * @throws \Exception
+     */
+    protected function doExecute(StepExecution $stepExecution)
+    {
+        try {
+            $currentExportDir = rtrim($stepExecution->getJobParameters()->get('exportDir'), '/');
+            $newExportDir = rtrim($this->exportLocation, '/');
+
+            $stepExecution->addSummaryInfo('log_file', $this->logFile);
+            $stepExecution->addSummaryInfo('export_location', $newExportDir);
+
+            $output = $this->syncMedia($currentExportDir, $newExportDir);
+            $this->writeLog($this->getModifiedOutputForLog($output, $stepExecution));
+
+            $stepExecution->addSummaryInfo('read', $output[1]);
+            $stepExecution->addSummaryInfo('write', $output[2]);
+        } catch(\Exception $e) {
+            $this->writeLog(['Error - something went wrong during media export.', $e->getMessage()]);
+            throw $e;
+        }
+    }
+
+    /**
+     * @param $currentExportDir
+     * @param $newExportDir
+     * @return array
+     * @throws FileTransferException
+     * @author James Pollard <jp@amp.co>
+     */
+    protected function syncMedia($currentExportDir, $newExportDir)
+    {
+        /**
+         * append files to the current export dir so that we do not unnecessarily
+         * copy over the export csv files
+         */
+        exec("rsync -avhK --stats $currentExportDir/files/ $newExportDir/", $output, $status);
+
+        if ($status !== 0) {
+            throw new FileTransferException('Error - rsync failure during media export.' . implode(" : ", $output));
+        }
+
+        return $output;
+    }
+
+    /**
+     * @param array $content
+     * @author James Pollard <jp@amp.co>
+     */
+    protected function writeLog(array $content)
+    {
+        if (!is_dir(dirname($this->logFile))) {
+            mkdir(dirname($this->logFile), 0644, true);
+        }
+
+        $handle = fopen($this->logFile, 'a+');
+        if ($handle === false) {
+            throw new FileNotFoundException(
+                sprintf('Error - log file (%s) could not be opened during media export.', $this->logFile)
+            );
+        }
+
+        fputcsv($handle, $content, PHP_EOL);
+        fclose($handle);
+    }
+
+    /**
+     * @param array $output
+     * @param StepExecution $stepExecution
+     * @return array
+     * @author James Pollard <jp@amp.co>
+     */
+    protected function getModifiedOutputForLog(array $output, StepExecution $stepExecution)
+    {
+        $jobParameters = $stepExecution->getJobParameters();
+        $jobExecution = $stepExecution->getJobExecution();
+
+        array_unshift(
+            $output,
+            '------------------------------',
+            sprintf('Export Profile: %s (%s)', $jobExecution->getLabel(), $jobParameters->get('applicationId')),
+            sprintf('Execution ID: %s', $jobExecution->getId()),
+            date('d/m/Y H:i:s')
+        );
+
+        return $output;
+    }
+}