<commit_before>import sys, os, inspect, json
from dataactcore.utils.jsonResponse import JsonResponse
from dataactcore.utils.statusCode import StatusCode
from interfaces.jobTrackerInterface import JobTrackerInterface
import struct
from dataactcore.utils.requestDictionary import RequestDictionary
from filestreaming.csvReader import CsvReader
from interfaces.stagingInterface import StagingInterface
from interfaces.validationInterface import ValidationInterface
from validator import Validator
from dataactcore.aws.s3UrlHandler import s3UrlHandler
from dataactcore.utils.responseException import ResponseException
from csv import Error
from filestreaming.csvWriter import CsvWriter
from handlers.validationError import ValidationError

class ValidationManager:
    """

    Outer level class, called by flask route

    """
    reportHeaders = ["Field name", "Error message", "Row number"]

    def markJob(self,jobId,jobTracker,status) :
        try :
            jobTracker.markStatus(jobId,status)
        except Exception as e:
            pass

    def getJobID(self,request) :
        """
        args
        request -- HTTP request containing the jobId
        returns the jobId
        if the job is not ready an exception will be raised
        """
        requestDict = RequestDictionary(request)
        if(requestDict.exists("job_id")):
            jobId = requestDict.getValue("job_id")
            tableName = "job"+str(jobId)
        else:
                # Request does not have a job ID, can't validate
            exc = ResponseException("No job ID specified in request")
            exc.status = StatusCode.CLIENT_ERROR
            raise exc
        jobTracker = JobTrackerInterface()
        if(not (jobTracker.runChecks(jobId))):
            exc = ResponseException("Checks failed on Job ID")
            exc.status = StatusCode.CLIENT_ERROR
            raise exc

        return jobId


    def threadedValidateJob(self,jobId) :
        """
        args
        jobId -- (Integer) a valid jobId
        This method runs on a new thread thus thus
        there are zero error messages other then the
        job status being updated

        """

        jobTracker = JobTrackerInterface()
        try:
            self.runValidation(jobId, jobTracker)


            return
        except ResponseException as e:
            self.markJob(jobId,jobTracker,"invalid")
        except ValueError as e:
            self.markJob(jobId,jobTracker,"invalid")
        except ValueError as e:
            self.markJob(jobId,jobTracker,"invalid")
        except Exception as e:
            #Something unkown happend we may need to try again!
            self.markJob(jobId,jobTracker,"failed")


    def runValidation(self, jobId, jobTracker):
        rowNumber = 1

        fileType = jobTracker.getFileType(jobId)
        # Get bucket name and file name
        fileName = jobTracker.getFileName(jobId)
        bucketName = s3UrlHandler.getBucketNameFromConfig()
        errorFileName = jobTracker.getReportPath(jobId)

        validationDB = ValidationInterface()
        fieldList = validationDB.getFieldsByFileList(fileType)
        csvSchema  = validationDB.getFieldsByFile(fileType)
        rules = validationDB.getRulesByFile(fileType)
        # Pull file from S3
        reader = CsvReader()
        # Use test file for now
        #fileName = "test.csv"
        reader.openFile(bucketName, fileName,fieldList)
        # Create staging table
        # While not done, pull one row and put it into staging if it passes
        # the Validator
        tableName = "job"+str(jobId)
        stagingDb = StagingInterface()
        tableName = stagingDb.createTable(fileType,fileName,jobId,tableName)

        with CsvWriter(bucketName, errorFileName, self.reportHeaders) as writer:
            while(not reader.isFinished):
                rowNumber += 1
                try :
                    record = reader.getNextRecord()
                except ResponseException as e:
                    if(not (reader.isFinished and reader.extraLine) ) :
                        #Last line may be blank dont throw an error
                        writer.write(["Formatting Error", ValidationError.readError, str(rowNumber)])
                    continue
                if(Validator.validate(record,rules,csvSchema)) :<commit_msg>moved error directory outside database hook.<commit_after>import sys, os, inspect, json
from dataactcore.utils.jsonResponse import JsonResponse
from dataactcore.utils.statusCode import StatusCode
from interfaces.jobTrackerInterface import JobTrackerInterface
import struct
from dataactcore.utils.requestDictionary import RequestDictionary
from filestreaming.csvReader import CsvReader
from interfaces.stagingInterface import StagingInterface
from interfaces.validationInterface import ValidationInterface
from validator import Validator
from dataactcore.aws.s3UrlHandler import s3UrlHandler
from dataactcore.utils.responseException import ResponseException
from csv import Error
from filestreaming.csvWriter import CsvWriter
from handlers.validationError import ValidationError

class ValidationManager:
    """

    Outer level class, called by flask route

    """
    reportHeaders = ["Field name", "Error message", "Row number"]

    def markJob(self,jobId,jobTracker,status) :
        try :
            jobTracker.markStatus(jobId,status)
        except Exception as e:
            pass

    def getJobID(self,request) :
        """
        args
        request -- HTTP request containing the jobId
        returns the jobId
        if the job is not ready an exception will be raised
        """
        requestDict = RequestDictionary(request)
        if(requestDict.exists("job_id")):
            jobId = requestDict.getValue("job_id")
            tableName = "job"+str(jobId)
        else:
                # Request does not have a job ID, can't validate
            exc = ResponseException("No job ID specified in request")
            exc.status = StatusCode.CLIENT_ERROR
            raise exc
        jobTracker = JobTrackerInterface()
        if(not (jobTracker.runChecks(jobId))):
            exc = ResponseException("Checks failed on Job ID")
            exc.status = StatusCode.CLIENT_ERROR
            raise exc

        return jobId


    def threadedValidateJob(self,jobId) :
        """
        args
        jobId -- (Integer) a valid jobId
        This method runs on a new thread thus thus
        there are zero error messages other then the
        job status being updated

        """

        jobTracker = JobTrackerInterface()
        try:
            self.runValidation(jobId, jobTracker)


            return
        except ResponseException as e:
            self.markJob(jobId,jobTracker,"invalid")
        except ValueError as e:
            self.markJob(jobId,jobTracker,"invalid")
        except ValueError as e:
            self.markJob(jobId,jobTracker,"invalid")
        except Exception as e:
            #Something unkown happend we may need to try again!
            self.markJob(jobId,jobTracker,"failed")


    def runValidation(self, jobId, jobTracker):
        rowNumber = 1

        fileType = jobTracker.getFileType(jobId)
        # Get bucket name and file name
        fileName = jobTracker.getFileName(jobId)
        bucketName = s3UrlHandler.getBucketNameFromConfig()
        errorFileName = "errors/"+jobTracker.getReportPath(jobId)

        validationDB = ValidationInterface()
        fieldList = validationDB.getFieldsByFileList(fileType)
        csvSchema  = validationDB.getFieldsByFile(fileType)
        rules = validationDB.getRulesByFile(fileType)
        # Pull file from S3
        reader = CsvReader()
        # Use test file for now
        #fileName = "test.csv"
        reader.openFile(bucketName, fileName,fieldList)
        # Create staging table
        # While not done, pull one row and put it into staging if it passes
        # the Validator
        tableName = "job"+str(jobId)
        stagingDb = StagingInterface()
        tableName = stagingDb.createTable(fileType,fileName,jobId,tableName)

        with CsvWriter(bucketName, errorFileName, self.reportHeaders) as writer:
            while(not reader.isFinished):
                rowNumber += 1
                try :
                    record = reader.getNextRecord()
                except ResponseException as e:
                    if(not (reader.isFinished and reader.extraLine) ) :
                        #Last line may be blank dont throw an error
                        writer.write(["Formatting Error", ValidationError.readError, str(rowNumber)])
                    continue
                if(Validator.validate(record,rules,csvSchema)) :
{'fraction_non_alphanumeric': 0.06802799128140415, 'fraction_numerical': 0.0013766203969255478, 'mean_word_length': 2.437697160883281, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '3718672', 'n_tokens_mistral': 2215, 'n_tokens_neox': 2140, 'n_words': 643}