<commit_before>from celery import Celery
from dataactcore.config import CONFIG_DB, CONFIG_SERVICES, CONFIG_JOB_QUEUE, CONFIG_BROKER
import requests
from dataactvalidator.filestreaming.csvS3Writer import CsvS3Writer
from dataactvalidator.filestreaming.csvLocalWriter import CsvLocalWriter
from csv import reader
from dataactcore.utils.jsonResponse import JsonResponse
from dataactcore.utils.responseException import ResponseException
from dataactcore.utils.statusCode import StatusCode

class JobQueue:
    def __init__(self, job_queue_url="localhost"):
        # Set up backend persistent URL
        backendUrl = ''.join(['db+', CONFIG_DB['scheme'], '://', CONFIG_DB['username'], ':', CONFIG_DB['password'], '@',
                              CONFIG_DB['host'], '/', CONFIG_DB['job_queue_db_name']])

        # Set up url to the validator for the RESTFul calls
        validatorUrl = ''.join([CONFIG_SERVICES['validator_host'], ':', str(CONFIG_SERVICES['validator_port'])])
        if 'http://' not in validatorUrl:
            validatorUrl = ''.join(['http://', validatorUrl])

        # Set up url to the job queue to establish connection
        queueUrl = ''.join(
            [CONFIG_JOB_QUEUE['broker_scheme'], '://', CONFIG_JOB_QUEUE['username'], ':', CONFIG_JOB_QUEUE['password'],
             '@', job_queue_url, ':', str(CONFIG_JOB_QUEUE['port']), '//'])

        # Create remote connection to the job queue
        self.jobQueue = Celery('tasks', backend=backendUrl, broker=queueUrl)

        @self.jobQueue.task(name='jobQueue.enqueue')
        def enqueue(jobID):
            # Don't need to worry about the response currently
            url = ''.join([validatorUrl, '/validate/'])
            params = {
                'job_id': jobID
            }
            response = requests.post(url, params)
            return response.json()

        @self.jobQueue.task(name='jobQueue.generate_d_file')
        def generate_d_file(api_url, user_id, job_id, interface_holder, timestamped_name, isLocal):
            job_manager = interface_holder().jobDb

            try:
                xml_response = self.get_xml_response_content(api_url)
                url_start_index = xml_response.find("<results>", 0)
                offset = len("<results>")

                if url_start_index == -1:
                    raise ResponseException("Empty response. Validate if input is correct.", StatusCode.CLIENT_ERROR)

                url_start_index += offset
                file_url = xml_response[url_start_index:xml_response.find("</results>", url_start_index)]

                full_file_path = "".join([CONFIG_BROKER['d_file_storage_path'], timestamped_name])

                self.download_file(full_file_path, file_url)
                lines = self.get_lines_from_csv(full_file_path)

                headers = lines[0]

                if isLocal:
                    file_name = "".join([CONFIG_BROKER['broker_files'], timestamped_name])
                    csv_writer = CsvLocalWriter(file_name, headers)
                else:
                    file_name = "".join([str(user_id), "/", timestamped_name])
                    bucket = CONFIG_BROKER['aws_bucket']
                    region = CONFIG_BROKER['aws_region']
                    csv_writer = CsvS3Writer(region, bucket, file_name, headers)

                with csv_writer as writer:
                    for line in lines[1:]:
                        writer.write(line)
                    writer.finishBatch()

                job_manager.markJobStatus(job_id, "finished")
                return {"message": "Success", "file_name": file_name}
            except Exception as e:
                # Log the error
                JsonResponse.error(e,500)
                job_manager.getJobById(job_id).error_message = str(e)
                job_manager.markJobStatus(job_id, "failed")
                job_manager.session.commit()
                raise e

        self.enqueue = enqueue
        self.generate_d_file = generate_d_file

    def get_xml_response_content(self, api_url):
        """ Retrieve XML Response from the provided API url """
        return requests.get(api_url, verify=False).text

    def download_file(self, local_file_path, file_url):
        """ Download a file locally from the specified URL """
        with open(local_file_path, "w") as file:
            # get request
            response = requests.get(file_url)
            # write to file<commit_msg>Closing interfaces at end of file generation<commit_after>from celery import Celery
from dataactcore.config import CONFIG_DB, CONFIG_SERVICES, CONFIG_JOB_QUEUE, CONFIG_BROKER
import requests
from dataactvalidator.filestreaming.csvS3Writer import CsvS3Writer
from dataactvalidator.filestreaming.csvLocalWriter import CsvLocalWriter
from csv import reader
from dataactcore.utils.jsonResponse import JsonResponse
from dataactcore.utils.responseException import ResponseException
from dataactcore.utils.statusCode import StatusCode

class JobQueue:
    def __init__(self, job_queue_url="localhost"):
        # Set up backend persistent URL
        backendUrl = ''.join(['db+', CONFIG_DB['scheme'], '://', CONFIG_DB['username'], ':', CONFIG_DB['password'], '@',
                              CONFIG_DB['host'], '/', CONFIG_DB['job_queue_db_name']])

        # Set up url to the validator for the RESTFul calls
        validatorUrl = ''.join([CONFIG_SERVICES['validator_host'], ':', str(CONFIG_SERVICES['validator_port'])])
        if 'http://' not in validatorUrl:
            validatorUrl = ''.join(['http://', validatorUrl])

        # Set up url to the job queue to establish connection
        queueUrl = ''.join(
            [CONFIG_JOB_QUEUE['broker_scheme'], '://', CONFIG_JOB_QUEUE['username'], ':', CONFIG_JOB_QUEUE['password'],
             '@', job_queue_url, ':', str(CONFIG_JOB_QUEUE['port']), '//'])

        # Create remote connection to the job queue
        self.jobQueue = Celery('tasks', backend=backendUrl, broker=queueUrl)

        @self.jobQueue.task(name='jobQueue.enqueue')
        def enqueue(jobID):
            # Don't need to worry about the response currently
            url = ''.join([validatorUrl, '/validate/'])
            params = {
                'job_id': jobID
            }
            response = requests.post(url, params)
            return response.json()

        @self.jobQueue.task(name='jobQueue.generate_d_file')
        def generate_d_file(api_url, user_id, job_id, interface_holder, timestamped_name, isLocal):
            job_manager = interface_holder().jobDb

            try:
                xml_response = self.get_xml_response_content(api_url)
                url_start_index = xml_response.find("<results>", 0)
                offset = len("<results>")

                if url_start_index == -1:
                    raise ResponseException("Empty response. Validate if input is correct.", StatusCode.CLIENT_ERROR)

                url_start_index += offset
                file_url = xml_response[url_start_index:xml_response.find("</results>", url_start_index)]

                full_file_path = "".join([CONFIG_BROKER['d_file_storage_path'], timestamped_name])

                self.download_file(full_file_path, file_url)
                lines = self.get_lines_from_csv(full_file_path)

                headers = lines[0]

                if isLocal:
                    file_name = "".join([CONFIG_BROKER['broker_files'], timestamped_name])
                    csv_writer = CsvLocalWriter(file_name, headers)
                else:
                    file_name = "".join([str(user_id), "/", timestamped_name])
                    bucket = CONFIG_BROKER['aws_bucket']
                    region = CONFIG_BROKER['aws_region']
                    csv_writer = CsvS3Writer(region, bucket, file_name, headers)

                with csv_writer as writer:
                    for line in lines[1:]:
                        writer.write(line)
                    writer.finishBatch()

                job_manager.markJobStatus(job_id, "finished")
                return {"message": "Success", "file_name": file_name}
            except Exception as e:
                # Log the error
                JsonResponse.error(e,500)
                job_manager.getJobById(job_id).error_message = str(e)
                job_manager.markJobStatus(job_id, "failed")
                job_manager.session.commit()
                raise e
            finally:
                interface_holder.close()

        self.enqueue = enqueue
        self.generate_d_file = generate_d_file

    def get_xml_response_content(self, api_url):
        """ Retrieve XML Response from the provided API url """
        return requests.get(api_url, verify=False).text

    def download_file(self, local_file_path, file_url):
        """ Download a file locally from the specified URL """
        with open(local_file_path, "w") as file:
            # get request
            response = requests.get(file_url)
            # write to file
{'fraction_non_alphanumeric': 0.10409499500610365, 'fraction_numerical': 0.002219509488403063, 'mean_word_length': 2.3353071798667653, 'pattern_counts': {'":': 4, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 12}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '335002', 'n_tokens_mistral': 2550, 'n_tokens_neox': 2426, 'n_words': 535}