<commit_before>package org.slc.sli.ingestion.routes;

import java.io.File;
import java.util.ArrayList;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.spring.SpringRouteBuilder;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import org.slc.sli.ingestion.BatchJob;
import org.slc.sli.ingestion.Fault;
import org.slc.sli.ingestion.landingzone.LocalFileSystemLandingZone;
import org.slc.sli.ingestion.processors.ControlFilePreProcessor;
import org.slc.sli.ingestion.processors.ControlFileProcessor;
import org.slc.sli.ingestion.processors.EdFiProcessor;
import org.slc.sli.ingestion.processors.PersistenceProcessor;
import org.slc.sli.ingestion.processors.ZipFileProcessor;

@Component
public class IngestionRouteBuilder extends SpringRouteBuilder {

    @Autowired
    EdFiProcessor xmlProcessor;

    @Autowired
    ControlFileProcessor ctlFileProcessor;

    @Autowired
    ZipFileProcessor zipFileProcessor;

    @Autowired(required = true)
    PersistenceProcessor persistenceProcessor;

    @Autowired
    LocalFileSystemLandingZone lz;

    @Autowired
    LocalFileSystemLandingZone tempLz;

    @Override
    public void configure() throws Exception {

        String inboundDir = lz.getDirectory().getPath();

        // routeId: ctlFilePoller
        from(
                "file:" + inboundDir + "?include=^(.*)\\.ctl$&move="
                        + inboundDir + "/.done&moveFailed=" + inboundDir
                        + "/.error").routeId("ctlFilePoller")
                .process(new ControlFilePreProcessor(lz))
                .to("seda:CtrlFilePreProcessor");

        // routeId: ctlFilePreprocessor
        from("seda:CtrlFilePreProcessor")
                .routeId("ctlFilePreprocessor")
                .process(ctlFileProcessor)
                .to("seda:assembledJobs");

        // routeId: zipFilePoller
        from(
                "file:" + inboundDir + "?include=^(.*)\\.zip$&move="
                        + inboundDir + "/.done&moveFailed=" + inboundDir
                        + "/.error")
                .routeId("zipFilePoller")
                .process(zipFileProcessor)
                .process(new Processor() {

                    // set temporary path to where the files were unzipped
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        File ctlFile = exchange.getIn().getBody(File.class);
                        tempLz.setDirectory(ctlFile.getParentFile());
                    }
                }).process(new ControlFilePreProcessor(tempLz))
                .choice()
                    .when(body().isInstanceOf(BatchJob.class))
                    .to("seda:assembledJobs")
                .otherwise()
                    .to("seda:CtrlFilePreProcessor");

        // routeId: jobDispatch
        from("seda:assembledJobs")
                .routeId("jobDispatch")
                .wireTap("direct:jobReporting")
                .choice()
                    .when(header("hasErrors").isEqualTo(true))
                    .stop()<commit_msg>terminate process and report error before entering into ctlFilePreProcessor if the zip file is corrupted<commit_after>package org.slc.sli.ingestion.routes;

import java.io.File;
import java.util.ArrayList;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.spring.SpringRouteBuilder;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import org.slc.sli.ingestion.BatchJob;
import org.slc.sli.ingestion.Fault;
import org.slc.sli.ingestion.landingzone.LocalFileSystemLandingZone;
import org.slc.sli.ingestion.processors.ControlFilePreProcessor;
import org.slc.sli.ingestion.processors.ControlFileProcessor;
import org.slc.sli.ingestion.processors.EdFiProcessor;
import org.slc.sli.ingestion.processors.PersistenceProcessor;
import org.slc.sli.ingestion.processors.ZipFileProcessor;

@Component
public class IngestionRouteBuilder extends SpringRouteBuilder {

    @Autowired
    EdFiProcessor xmlProcessor;

    @Autowired
    ControlFileProcessor ctlFileProcessor;

    @Autowired
    ZipFileProcessor zipFileProcessor;

    @Autowired(required = true)
    PersistenceProcessor persistenceProcessor;

    @Autowired
    LocalFileSystemLandingZone lz;

    @Autowired
    LocalFileSystemLandingZone tempLz;

    @Override
    public void configure() throws Exception {

        String inboundDir = lz.getDirectory().getPath();

        // routeId: ctlFilePoller
        from(
                "file:" + inboundDir + "?include=^(.*)\\.ctl$&move="
                        + inboundDir + "/.done&moveFailed=" + inboundDir
                        + "/.error").routeId("ctlFilePoller")
                .process(new ControlFilePreProcessor(lz))
                .to("seda:CtrlFilePreProcessor");

        // routeId: ctlFilePreprocessor
        from("seda:CtrlFilePreProcessor")
                .routeId("ctlFilePreprocessor")
                .process(ctlFileProcessor)
                .to("seda:assembledJobs");

        // routeId: zipFilePoller
        from(
                "file:" + inboundDir + "?include=^(.*)\\.zip$&move="
                        + inboundDir + "/.done&moveFailed=" + inboundDir
                        + "/.error")
                .routeId("zipFilePoller")
                .process(zipFileProcessor)
                .choice()
                    .when(body().isInstanceOf(BatchJob.class))
                    .to("seda:assembledJobs")
                .otherwise()
                    .process(new Processor() {

                        // set temporary path to where the files were unzipped
                        @Override
                        public void process(Exchange exchange) throws Exception {
                            File ctlFile = exchange.getIn().getBody(File.class);
                            tempLz.setDirectory(ctlFile.getParentFile());
                        }
                    }).process(new ControlFilePreProcessor(tempLz))
                    .to("seda:CtrlFilePreProcessor");

        // routeId: jobDispatch
        from("seda:assembledJobs")
                .routeId("jobDispatch")
                .wireTap("direct:jobReporting")
                .choice()
                    .when(header("hasErrors").isEqualTo(true))
                    .stop()
{'fraction_non_alphanumeric': 0.10391589212250495, 'fraction_numerical': 0.0, 'mean_word_length': 2.1786924939467314, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '1698452', 'n_tokens_mistral': 1865, 'n_tokens_neox': 1757, 'n_words': 285}