<commit_before><?php

namespace NS\ImportBundle\Importer;

use \Ddeboer\DataImport\Filter\OffsetFilter;
use \Ddeboer\DataImport\Reader\CsvReader;
use \Ddeboer\DataImport\Step\ConverterStep;
use \Ddeboer\DataImport\Workflow;
use \Ddeboer\DataImport\Reader;
use \Ddeboer\DataImport\Step\FilterStep;
use \Ddeboer\DataImport\Step\ValueConverterStep;
use \NS\ImportBundle\Converter\DateRangeConverter;
use \NS\ImportBundle\Converter\NoFutureDateConverter;
use \NS\ImportBundle\Converter\TrimInputConverter;
use \NS\ImportBundle\Converter\WarningConverter;
use \NS\ImportBundle\Entity\Import;
use \NS\ImportBundle\Filter\Duplicate;
use \NS\ImportBundle\Filter\NotBlank;
use \NS\ImportBundle\Writer\DoctrineWriter;
use \Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Description of ImportProcessor
 *
 * @author gnat
 */
class ImportProcessor
{
    private $container;
    private $duplicateFilter;
    private $notBlankFilter;
    private $doctrineWriter;

    private $memoryLimit = '1024M';
    private $maxExecutionTime = 190;
    private $limit  = null;

    /**
     * @param ContainerInterface $container
     */
    public function __construct(ContainerInterface $container)
    {
        $this->setContainer($container);
    }

    /**
     * @param Import $import
     * @return Result
     * @throws \Exception
     */
    public function process(Import $import)
    {
        ini_set('max_execution_time', $this->maxExecutionTime);
        ini_set('memory_limit', $this->memoryLimit);

        $reader = $this->getReader($import);
        $reader->seek($import->getPosition());
        // Create the workflow from the reader
        $workflow = new Workflow\StepAggregator($reader);
        $workflow->setSkipItemOnFailure(true);
        $workflow->addWriter($this->getWriter($import->getClass()));

        $this->addSteps($workflow, $import);

        // Process the workflow
        $result = $workflow->process();
        if($this->duplicateFilter) {
            $this->duplicateFilter->finish();
        }

        return $result;
    }

    /**
     *
     * @param string $class
     * @return DoctrineWriter
     * @throws \InvalidArgumentException
     */
    public function getWriter($class)
    {
        if ($this->doctrineWriter === null || $this->doctrineWriter->getEntityName() != $class) {
            $this->doctrineWriter = new DoctrineWriter($this->container->get('doctrine.orm.entity_manager'), $class, array('getcode' => 'site', 1 => 'caseId'));
            $this->doctrineWriter->setTruncate(false);
            $this->doctrineWriter->setClearOnFlush(false);
            $this->doctrineWriter->setEntityRepositoryMethod('findWithRelations');
        }

        return $this->doctrineWriter;
    }

    /**
     * @param Import $import
     * @return Reader $csvReader
     */
    public function getReader(Import $import)
    {
        // Create and configure the reader
        $csvReader = new CsvReader($import->getSourceFile()->openFile(), ',');

        // Tell the reader that the first row in the CSV file contains column headers
        $csvReader->setHeaderRowNumber(0);

        $import->setSourceCount($csvReader->count());

        $fields = $csvReader->getFields();
        $columns = $import->getMap()->getColumns();

        foreach ($columns as $column) {
            if ($column->getName() != $fields[$column->getOrder()]) {
                throw new \InvalidArgumentException(sprintf("%s != %s probably the wrong file or missing headers", $fields[$column->getOrder()], $column->getName()));
            }
        }

        return $csvReader;
    }

    /**
     * @param Workflow $workflow
     * @param Import $import
     * @return Duplicate $duplicate
     */
    public function addSteps(Workflow $workflow, Import $import)
    {
        $offsetFilter = new FilterStep();
        $offsetFilter->add(new OffsetFilter(0,$this->getLimit(),true));

        // Move to current position
        $workflow->addStep($offsetFilter,50);

        // Trim all input
        $workflow->addStep(new ConverterStep(array(new TrimInputConverter())),45);

        $preProcessor = $import->getPreprocessor();
        if ($preProcessor) {
            $workflow->addStep($preProcessor,42);
        }

        // These map column headers i.e site_Code -> site
        $workflow->addStep($import->getMappings(),40);

        // These allow us to ignore a column i.e. - region or country_ISO 
        $workflow->addStep($import->getIgnoredMapper(),30);

        $valueConverter = new ValueConverterStep();
        $valueConverterCount = 0;
        foreach ($import->getConverters() as $column) {
            $name = ($column->hasMapper()) ? $column->getMapper() : $column->getName();
            $valueConverter->add(sprintf('[%s]', str_replace('.', '][', $name)), $this->container->get($column->getConverter()));
            $valueConverterCount++;
        }

        if ($valueConverterCount > 0) {
            $workflow->addStep($valueConverter,20);
        }

        $filterStep = new FilterStep();
        $addFilter = false;

        if ($this->notBlankFilter) {
            $filterStep->add($this->notBlankFilter);
            $addFilter = true;
        }

        if ($this->duplicateFilter) {
            $filterStep->add($this->getDuplicate());
            $addFilter = true;
        }

        if ($addFilter) {
            $workflow->addStep($filterStep,10);
        }

        // Adds warnings for out of range values
        $converter = new ConverterStep();
        $converter->add(new WarningConverter());
        $converter->add(new DateRangeConverter(new \DateTime()));

        $start = clone $import->getInputDateStart();
        $start->sub(new \DateInterval('P5Y'));

        $converter->add(new DateRangeConverter($import->getInputDateEnd(),$start));

        $workflow->addStep($converter,5);
    }

    /**
     * @return string
     */
    public function getMemoryLimit()
    {
        return $this->memoryLimit;
    }

    /**
     * @return integer
     */
    public function getMaxExecutionTime()
    {
        return $this->maxExecutionTime;
    }

    /**
     * @param string $memoryLimit
     * @return \NS\ImportBundle\Services\ImportProcessor
     */
    public function setMemoryLimit($memoryLimit)
    {
        $this->memoryLimit = $memoryLimit;
        return $this;
    }

    /**
     * @param integer $maxExecutionTime
     * @return \NS\ImportBundle\Services\ImportProcessor<commit_msg>reset priorities<commit_after><?php

namespace NS\ImportBundle\Importer;

use \Ddeboer\DataImport\Filter\OffsetFilter;
use \Ddeboer\DataImport\Reader\CsvReader;
use \Ddeboer\DataImport\Step\ConverterStep;
use \Ddeboer\DataImport\Workflow;
use \Ddeboer\DataImport\Reader;
use \Ddeboer\DataImport\Step\FilterStep;
use \Ddeboer\DataImport\Step\ValueConverterStep;
use \NS\ImportBundle\Converter\DateRangeConverter;
use \NS\ImportBundle\Converter\NoFutureDateConverter;
use \NS\ImportBundle\Converter\TrimInputConverter;
use \NS\ImportBundle\Converter\WarningConverter;
use \NS\ImportBundle\Entity\Import;
use \NS\ImportBundle\Filter\Duplicate;
use \NS\ImportBundle\Filter\NotBlank;
use \NS\ImportBundle\Writer\DoctrineWriter;
use \Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Description of ImportProcessor
 *
 * @author gnat
 */
class ImportProcessor
{
    private $container;
    private $duplicateFilter;
    private $notBlankFilter;
    private $doctrineWriter;

    private $memoryLimit = '1024M';
    private $maxExecutionTime = 190;
    private $limit  = null;

    /**
     * @param ContainerInterface $container
     */
    public function __construct(ContainerInterface $container)
    {
        $this->setContainer($container);
    }

    /**
     * @param Import $import
     * @return Result
     * @throws \Exception
     */
    public function process(Import $import)
    {
        ini_set('max_execution_time', $this->maxExecutionTime);
        ini_set('memory_limit', $this->memoryLimit);

        $reader = $this->getReader($import);
        $reader->seek($import->getPosition());
        // Create the workflow from the reader
        $workflow = new Workflow\StepAggregator($reader);
        $workflow->setSkipItemOnFailure(true);
        $workflow->addWriter($this->getWriter($import->getClass()));

        $this->addSteps($workflow, $import);

        // Process the workflow
        $result = $workflow->process();
        if($this->duplicateFilter) {
            $this->duplicateFilter->finish();
        }

        return $result;
    }

    /**
     *
     * @param string $class
     * @return DoctrineWriter
     * @throws \InvalidArgumentException
     */
    public function getWriter($class)
    {
        if ($this->doctrineWriter === null || $this->doctrineWriter->getEntityName() != $class) {
            $this->doctrineWriter = new DoctrineWriter($this->container->get('doctrine.orm.entity_manager'), $class, array('getcode' => 'site', 1 => 'caseId'));
            $this->doctrineWriter->setTruncate(false);
            $this->doctrineWriter->setClearOnFlush(false);
            $this->doctrineWriter->setEntityRepositoryMethod('findWithRelations');
        }

        return $this->doctrineWriter;
    }

    /**
     * @param Import $import
     * @return Reader $csvReader
     */
    public function getReader(Import $import)
    {
        // Create and configure the reader
        $csvReader = new CsvReader($import->getSourceFile()->openFile(), ',');

        // Tell the reader that the first row in the CSV file contains column headers
        $csvReader->setHeaderRowNumber(0);

        $import->setSourceCount($csvReader->count());

        $fields = $csvReader->getFields();
        $columns = $import->getMap()->getColumns();

        foreach ($columns as $column) {
            if ($column->getName() != $fields[$column->getOrder()]) {
                throw new \InvalidArgumentException(sprintf("%s != %s probably the wrong file or missing headers", $fields[$column->getOrder()], $column->getName()));
            }
        }

        return $csvReader;
    }

    /**
     * @param Workflow $workflow
     * @param Import $import
     * @return Duplicate $duplicate
     */
    public function addSteps(Workflow $workflow, Import $import)
    {
        $offsetFilter = new FilterStep();
        $offsetFilter->add(new OffsetFilter(0, $this->getLimit(), true));

        // Move to current position
        $workflow->addStep($offsetFilter, 80);

        // Trim all input
        $workflow->addStep(new ConverterStep(array(new TrimInputConverter())), 70);

        $preProcessor = $import->getPreprocessor();
        if ($preProcessor) {
            $workflow->addStep($preProcessor, 60);
        }

        // These map column headers i.e site_Code -> site
        $workflow->addStep($import->getMappings(), 50);

        // These allow us to ignore a column i.e. - region or country_ISO 
        $workflow->addStep($import->getIgnoredMapper(), 40);

        $valueConverter = new ValueConverterStep();
        $valueConverterCount = 0;
        foreach ($import->getConverters() as $column) {
            $name = ($column->hasMapper()) ? $column->getMapper() : $column->getName();
            $valueConverter->add(sprintf('[%s]', str_replace('.', '][', $name)), $this->container->get($column->getConverter()));
            $valueConverterCount++;
        }

        if ($valueConverterCount > 0) {
            $workflow->addStep($valueConverter, 30);
        }

        $filterStep = new FilterStep();
        $addFilter = false;

        if ($this->notBlankFilter) {
            $filterStep->add($this->notBlankFilter);
            $addFilter = true;
        }

        if ($this->duplicateFilter) {
            $filterStep->add($this->getDuplicate());
            $addFilter = true;
        }

        if ($addFilter) {
            $workflow->addStep($filterStep, 20);
        }

        // Adds warnings for out of range values
        $converter = new ConverterStep();
        $converter->add(new WarningConverter());
        $converter->add(new DateRangeConverter(new \DateTime()));

        $start = clone $import->getInputDateStart();
        $start->sub(new \DateInterval('P5Y'));

        $converter->add(new DateRangeConverter($import->getInputDateEnd(), $start));

        $workflow->addStep($converter, 10);
    }

    /**
     * @return string
     */
    public function getMemoryLimit()
    {
        return $this->memoryLimit;
    }

    /**
     * @return integer
     */
    public function getMaxExecutionTime()
    {
        return $this->maxExecutionTime;
    }

    /**
     * @param string $memoryLimit
     * @return \NS\ImportBundle\Services\ImportProcessor
     */
    public function setMemoryLimit($memoryLimit)
    {
        $this->memoryLimit = $memoryLimit;
        return $this;
    }

    /**
     * @param integer $maxExecutionTime
     * @return \NS\ImportBundle\Services\ImportProcessor
{'fraction_non_alphanumeric': 0.1400046257034924, 'fraction_numerical': 0.0043944183177858295, 'mean_word_length': 2.6695898161244696, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 163, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7313594', 'n_tokens_mistral': 3738, 'n_tokens_neox': 3608, 'n_words': 701}