import csv
import json
import logging
import shutil
from pathlib import Path
from typing import Collection

from google.cloud import storage
from structlog import wrap_logger

from toolbox.bulk_processing.processor_interface import Processor
from toolbox.bulk_processing.validators import header_equal, Invalid, ValidationFailure
from toolbox.config import Config
from toolbox.utilities import db_helper
from toolbox.utilities.rabbit_context import RabbitContext

logger = wrap_logger(logging.getLogger(__name__))


class BulkProcessor:
    def __init__(self, processor: Processor):
        self.working_dir = Config.BULK_WORKING_DIRECTORY

        self.processor = processor
        self.storage_client = storage.Client(project=self.processor.project_id)
        self.storage_bucket = self.storage_client.bucket(self.processor.bucket_name)
        self.rabbit = None
        self.db_connection_pool = None

    def run(self):
        logger.info('Checking for files in bucket', bucket_name=self.processor.bucket_name,
                    prefix=self.processor.file_prefix)
        with db_helper.connect_to_read_replica_pool() as self.db_connection_pool, RabbitContext() as self.rabbit:
            blobs_to_process = self.storage_client.list_blobs(self.processor.bucket_name,
                                                              prefix=self.processor.file_prefix)

            for blob_to_process in blobs_to_process:
                self.process_bulk_file_from_bucket(blob_to_process, prefix=self.processor.file_prefix)

    def process_bulk_file_from_bucket(self, blob_to_process, prefix):
        logger.info('Processing file', file_to_process=blob_to_process.name, prefix=prefix)
        file_to_process = self.download_file_to_process(blob_to_process)
        success_file, error_file, error_detail_file = self.initialise_results_files(file_to_process.name)
        successes, errors = self.process_file(file_to_process, success_file, error_file,
                                              error_detail_file)

        logger.info('Uploading results from processing file', file_to_process=blob_to_process.name)
        self.upload_results(successes, errors, success_file, error_file, error_detail_file)

        logger.info('Deleting remote file', deleted_file=blob_to_process.name)
        blob_to_process.delete()

        self.delete_local_files((file_to_process, success_file, error_file, error_detail_file))
        logger.info('Finished processing file', file_to_process=blob_to_process.name)

    def process_file(self, file_to_process, success_file, error_file, error_detail_file):
        try:
            with open(file_to_process, encoding="utf-8") as open_file_to_process:

                file_reader = csv.DictReader(open_file_to_process, delimiter=',')
                header_validation_failures = self.find_header_validation_errors(file_reader.fieldnames)

                if header_validation_failures:
                    logger.info('Header row is invalid for file', file=file_to_process.name)
                    shutil.copy(file_to_process, error_file)
                    self.write_error_details_to_file([header_validation_failures], error_detail_file)
                    return 0, 1  # success_count, error_count

                self.write_header_ok_to_file(error_detail_file)

                return self.process_rows(file_reader, success_file, error_file, error_detail_file)

        except UnicodeDecodeError as err:
            self.write_file_encoding_error_files(err, error_file, error_detail_file, file_to_process)
            return 0, 1  # success_count, error_count

    @staticmethod
    def write_file_encoding_error_files(err, error_file, error_detail_file, file_to_process):
        shutil.copy(file_to_process, error_file)
        error_detail_file.write_text(f'Invalid file encoding, requires utf-8, error: {err}')

    def process_rows(self, file_reader, success_file, error_file, error_detail_file):
        error_count = 0
        success_count = 0
        for line_number, row in enumerate(file_reader, 1):
            row_errors = self.find_row_validation_errors(line_number, row)
            if row_errors:
                error_count += len(row_errors)
                self.write_row_errors_to_files(row, row_errors, error_file, error_detail_file)
            else:
                success_count += 1
                event_messages = self.processor.build_event_messages(row)
                self.publish_messages(event_messages, self.rabbit)
                self.write_row_success_to_file(row, success_file)
        logger.info('Processing results', rows_processed=line_number, failures=error_count,
                    bucket_name=self.processor.bucket_name)
        return success_count, error_count

    def initialise_results_files(self, file_to_process_name):
        header_row = ','.join(column for column in self.processor.schema.keys()) + '\n'
        success_file = self.working_dir.joinpath(f'PROCESSED_{file_to_process_name}')
        success_file.write_text(header_row)
        error_file = self.working_dir.joinpath(f'ERROR_{file_to_process_name}')
        error_file.write_text(header_row)
        error_detail_file = self.working_dir.joinpath(f'ERROR_DETAIL_{file_to_process_name}')
        error_detail_file.touch()
        return success_file, error_file, error_detail_file

    def find_header_validation_errors(self, header):
        try:
            header_equal(self.processor.schema.keys())(header)
        except Invalid as invalid:
            return ValidationFailure(line_number=1, column=None, description=str(invalid))

    def find_row_validation_errors(self, line_number, row):
        errors = []

        # Check we have the correct number of columns
        # If there are extra columns the values will be stored as a list in the None key
        if row.get(None):
            errors.append(ValidationFailure(line_number,
                                            column='Multiple',
                                            description='Row contains too many columns'))
            return errors

        # If we have too few columns the value for the missing end columns will be None
        # (Empty but present columns values are an empty string, not None)
        if any(value is None for value in row.values()):
            errors.append(ValidationFailure(line_number,
                                            column='Multiple',
                                            description='Row contains too few columns'))
            return errors

        for column, validators in self.processor.schema.items():
            for validator in validators:
                try:
                    validator(row[column], row=row, db_connection_pool=self.db_connection_pool, column=column)
                except Invalid as invalid:
                    errors.append(ValidationFailure(line_number, column, invalid))
        return errors

    def write_row_errors_to_files(self, errored_row, errors, error_file, error_detail_file):
        with open(error_file, 'a') as append_error_file:
            append_error_file.write(self.rebuild_errored_csv_row(errored_row))
            append_error_file.write('\n')
        self.write_error_details_to_file(errors, error_detail_file)

    @staticmethod
    def rebuild_errored_csv_row(row: dict):
        # Pop and safely join any extra, unexpected columns which are stored in a list in the 'None' key
        extra_values = row.pop(None, [])
        extra_values_str = ','.join(extra_values)

        rebuilt_row = ','.join(value for value in row.values() if value is not None)
        if extra_values_str:
            rebuilt_row = rebuilt_row + f',{extra_values_str}'

        return rebuilt_row

    @staticmethod
    def write_error_details_to_file(errors, error_detail_file):
        with open(error_detail_file, 'a') as append_error_detail_file:
            append_error_detail_file.write(
                ' | '.join(f"[Column: {error.column}, Error: {str(error.description)}]" for error in errors))
            append_error_detail_file.write('\n')

    @staticmethod
    def write_header_ok_to_file(error_detail_file):
        with open(error_detail_file, 'a') as append_error_detail_file:
            append_error_detail_file.write('Header row is valid\n')

    @staticmethod
    def write_row_success_to_file(succeeded_row, success_file):
        with open(success_file, 'a') as append_success_file:
            append_success_file.write(','.join(succeeded_row.values()))
            append_success_file.write('\n')

    def publish_messages(self, messages, rabbit):
        for message in messages:
            rabbit.publish_message(
                message=json.dumps(message),
                content_type='application/json',
                headers=None,
                exchange=self.processor.exchange,
                routing_key=self.processor.routing_key
            )

    def download_file_to_process(self, blob_to_process):
        file_to_process = self.working_dir.joinpath(blob_to_process.name)

        with open(file_to_process, 'wb') as open_file_to_process:
            self.storage_client.download_blob_to_file(blob_to_process, open_file_to_process)
        return file_to_process

    def upload_results(self, successes, errors, success_file, error_file, error_detail_file):
        # Only upload files which contain data, we don't want to make empty error or success files
        files_to_upload = []
        if successes:
            files_to_upload.append(success_file)
        if errors:
            files_to_upload.append(error_file)
            files_to_upload.append(error_detail_file)
        self.upload_files_to_bucket(files_to_upload)

    def upload_files_to_bucket(self, files: Collection[Path]):
        for file_to_upload in files:
            file_blob = self.storage_bucket.blob(file_to_upload.name)
            file_blob.upload_from_filename(str(file_to_upload))

    @staticmethod
    def delete_local_files(files_to_delete: Collection[Path]):
        logger.info('Deleting local files', files_to_delete=files_to_delete)
        for file_to_delete in files_to_delete:
            file_to_delete.unlink()

01f9f74f10af4f15d97aff7412f03bd190c63ab0
<mask token>


class BulkProcessor:
    <mask token>
    <mask token>

    def process_bulk_file_from_bucket(self, blob_to_process, prefix):
        logger.info('Processing file', file_to_process=blob_to_process.name,
            prefix=prefix)
        file_to_process = self.download_file_to_process(blob_to_process)
        success_file, error_file, error_detail_file = (self.
            initialise_results_files(file_to_process.name))
        successes, errors = self.process_file(file_to_process, success_file,
            error_file, error_detail_file)
        logger.info('Uploading results from processing file',
            file_to_process=blob_to_process.name)
        self.upload_results(successes, errors, success_file, error_file,
            error_detail_file)
        logger.info('Deleting remote file', deleted_file=blob_to_process.name)
        blob_to_process.delete()
        self.delete_local_files((file_to_process, success_file, error_file,
            error_detail_file))
        logger.info('Finished processing file', file_to_process=
            blob_to_process.name)
    <mask token>

    @staticmethod
    def write_file_encoding_error_files(err, error_file, error_detail_file,
        file_to_process):
        shutil.copy(file_to_process, error_file)
        error_detail_file.write_text(
            f'Invalid file encoding, requires utf-8, error: {err}')
    <mask token>
    <mask token>
    <mask token>
    <mask token>
    <mask token>

    @staticmethod
    def rebuild_errored_csv_row(row: dict):
        extra_values = row.pop(None, [])
        extra_values_str = ','.join(extra_values)
        rebuilt_row = ','.join(value for value in row.values() if value is not
            None)
        if extra_values_str:
            rebuilt_row = rebuilt_row + f',{extra_values_str}'
        return rebuilt_row

    @staticmethod
    def write_error_details_to_file(errors, error_detail_file):
        with open(error_detail_file, 'a') as append_error_detail_file:
            append_error_detail_file.write(' | '.join(
                f'[Column: {error.column}, Error: {str(error.description)}]'
                 for error in errors))
            append_error_detail_file.write('\n')
    <mask token>

    @staticmethod
    def write_row_success_to_file(succeeded_row, success_file):
        with open(success_file, 'a') as append_success_file:
            append_success_file.write(','.join(succeeded_row.values()))
            append_success_file.write('\n')

    def publish_messages(self, messages, rabbit):
        for message in messages:
            rabbit.publish_message(message=json.dumps(message),
                content_type='application/json', headers=None, exchange=
                self.processor.exchange, routing_key=self.processor.routing_key
                )

    def download_file_to_process(self, blob_to_process):
        file_to_process = self.working_dir.joinpath(blob_to_process.name)
        with open(file_to_process, 'wb') as open_file_to_process:
            self.storage_client.download_blob_to_file(blob_to_process,
                open_file_to_process)
        return file_to_process

    def upload_results(self, successes, errors, success_file, error_file,
        error_detail_file):
        files_to_upload = []
        if successes:
            files_to_upload.append(success_file)
        if errors:
            files_to_upload.append(error_file)
            files_to_upload.append(error_detail_file)
        self.upload_files_to_bucket(files_to_upload)
    <mask token>

    @staticmethod
    def delete_local_files(files_to_delete: Collection[Path]):
        logger.info('Deleting local files', files_to_delete=files_to_delete)
        for file_to_delete in files_to_delete:
            file_to_delete.unlink()

<mask token>


class BulkProcessor:

    def __init__(self, processor: Processor):
        self.working_dir = Config.BULK_WORKING_DIRECTORY
        self.processor = processor
        self.storage_client = storage.Client(project=self.processor.project_id)
        self.storage_bucket = self.storage_client.bucket(self.processor.
            bucket_name)
        self.rabbit = None
        self.db_connection_pool = None
    <mask token>

    def process_bulk_file_from_bucket(self, blob_to_process, prefix):
        logger.info('Processing file', file_to_process=blob_to_process.name,
            prefix=prefix)
        file_to_process = self.download_file_to_process(blob_to_process)
        success_file, error_file, error_detail_file = (self.
            initialise_results_files(file_to_process.name))
        successes, errors = self.process_file(file_to_process, success_file,
            error_file, error_detail_file)
        logger.info('Uploading results from processing file',
            file_to_process=blob_to_process.name)
        self.upload_results(successes, errors, success_file, error_file,
            error_detail_file)
        logger.info('Deleting remote file', deleted_file=blob_to_process.name)
        blob_to_process.delete()
        self.delete_local_files((file_to_process, success_file, error_file,
            error_detail_file))
        logger.info('Finished processing file', file_to_process=
            blob_to_process.name)
    <mask token>

    @staticmethod
    def write_file_encoding_error_files(err, error_file, error_detail_file,
        file_to_process):
        shutil.copy(file_to_process, error_file)
        error_detail_file.write_text(
            f'Invalid file encoding, requires utf-8, error: {err}')

    def process_rows(self, file_reader, success_file, error_file,
        error_detail_file):
        error_count = 0
        success_count = 0
        for line_number, row in enumerate(file_reader, 1):
            row_errors = self.find_row_validation_errors(line_number, row)
            if row_errors:
                error_count += len(row_errors)
                self.write_row_errors_to_files(row, row_errors, error_file,
                    error_detail_file)
            else:
                success_count += 1
                event_messages = self.processor.build_event_messages(row)
                self.publish_messages(event_messages, self.rabbit)
                self.write_row_success_to_file(row, success_file)
        logger.info('Processing results', rows_processed=line_number,
            failures=error_count, bucket_name=self.processor.bucket_name)
        return success_count, error_count
    <mask token>
    <mask token>
    <mask token>

    def write_row_errors_to_files(self, errored_row, errors, error_file,
        error_detail_file):
        with open(error_file, 'a') as append_error_file:
            append_error_file.write(self.rebuild_errored_csv_row(errored_row))
            append_error_file.write('\n')
        self.write_error_details_to_file(errors, error_detail_file)

    @staticmethod
    def rebuild_errored_csv_row(row: dict):
        extra_values = row.pop(None, [])
        extra_values_str = ','.join(extra_values)
        rebuilt_row = ','.join(value for value in row.values() if value is not
            None)
        if extra_values_str:
            rebuilt_row = rebuilt_row + f',{extra_values_str}'
        return rebuilt_row

    @staticmethod
    def write_error_details_to_file(errors, error_detail_file):
        with open(error_detail_file, 'a') as append_error_detail_file:
            append_error_detail_file.write(' | '.join(
                f'[Column: {error.column}, Error: {str(error.description)}]'
                 for error in errors))
            append_error_detail_file.write('\n')
    <mask token>

    @staticmethod
    def write_row_success_to_file(succeeded_row, success_file):
        with open(success_file, 'a') as append_success_file:
            append_success_file.write(','.join(succeeded_row.values()))
            append_success_file.write('\n')

    def publish_messages(self, messages, rabbit):
        for message in messages:
            rabbit.publish_message(message=json.dumps(message),
                content_type='application/json', headers=None, exchange=
                self.processor.exchange, routing_key=self.processor.routing_key
                )

    def download_file_to_process(self, blob_to_process):
        file_to_process = self.working_dir.joinpath(blob_to_process.name)
        with open(file_to_process, 'wb') as open_file_to_process:
            self.storage_client.download_blob_to_file(blob_to_process,
                open_file_to_process)
        return file_to_process

    def upload_results(self, successes, errors, success_file, error_file,
        error_detail_file):
        files_to_upload = []
        if successes:
            files_to_upload.append(success_file)
        if errors:
            files_to_upload.append(error_file)
            files_to_upload.append(error_detail_file)
        self.upload_files_to_bucket(files_to_upload)
    <mask token>

    @staticmethod
    def delete_local_files(files_to_delete: Collection[Path]):
        logger.info('Deleting local files', files_to_delete=files_to_delete)
        for file_to_delete in files_to_delete:
            file_to_delete.unlink()

<mask token>


class BulkProcessor:

    def __init__(self, processor: Processor):
        self.working_dir = Config.BULK_WORKING_DIRECTORY
        self.processor = processor
        self.storage_client = storage.Client(project=self.processor.project_id)
        self.storage_bucket = self.storage_client.bucket(self.processor.
            bucket_name)
        self.rabbit = None
        self.db_connection_pool = None
    <mask token>

    def process_bulk_file_from_bucket(self, blob_to_process, prefix):
        logger.info('Processing file', file_to_process=blob_to_process.name,
            prefix=prefix)
        file_to_process = self.download_file_to_process(blob_to_process)
        success_file, error_file, error_detail_file = (self.
            initialise_results_files(file_to_process.name))
        successes, errors = self.process_file(file_to_process, success_file,
            error_file, error_detail_file)
        logger.info('Uploading results from processing file',
            file_to_process=blob_to_process.name)
        self.upload_results(successes, errors, success_file, error_file,
            error_detail_file)
        logger.info('Deleting remote file', deleted_file=blob_to_process.name)
        blob_to_process.delete()
        self.delete_local_files((file_to_process, success_file, error_file,
            error_detail_file))
        logger.info('Finished processing file', file_to_process=
            blob_to_process.name)

    def process_file(self, file_to_process, success_file, error_file,
        error_detail_file):
        try:
            with open(file_to_process, encoding='utf-8'
                ) as open_file_to_process:
                file_reader = csv.DictReader(open_file_to_process,
                    delimiter=',')
                header_validation_failures = (self.
                    find_header_validation_errors(file_reader.fieldnames))
                if header_validation_failures:
                    logger.info('Header row is invalid for file', file=
                        file_to_process.name)
                    shutil.copy(file_to_process, error_file)
                    self.write_error_details_to_file([
                        header_validation_failures], error_detail_file)
                    return 0, 1
                self.write_header_ok_to_file(error_detail_file)
                return self.process_rows(file_reader, success_file,
                    error_file, error_detail_file)
        except UnicodeDecodeError as err:
            self.write_file_encoding_error_files(err, error_file,
                error_detail_file, file_to_process)
            return 0, 1

    @staticmethod
    def write_file_encoding_error_files(err, error_file, error_detail_file,
        file_to_process):
        shutil.copy(file_to_process, error_file)
        error_detail_file.write_text(
            f'Invalid file encoding, requires utf-8, error: {err}')

    def process_rows(self, file_reader, success_file, error_file,
        error_detail_file):
        error_count = 0
        success_count = 0
        for line_number, row in enumerate(file_reader, 1):
            row_errors = self.find_row_validation_errors(line_number, row)
            if row_errors:
                error_count += len(row_errors)
                self.write_row_errors_to_files(row, row_errors, error_file,
                    error_detail_file)
            else:
                success_count += 1
                event_messages = self.processor.build_event_messages(row)
                self.publish_messages(event_messages, self.rabbit)
                self.write_row_success_to_file(row, success_file)
        logger.info('Processing results', rows_processed=line_number,
            failures=error_count, bucket_name=self.processor.bucket_name)
        return success_count, error_count

    def initialise_results_files(self, file_to_process_name):
        header_row = ','.join(column for column in self.processor.schema.keys()
            ) + '\n'
        success_file = self.working_dir.joinpath(
            f'PROCESSED_{file_to_process_name}')
        success_file.write_text(header_row)
        error_file = self.working_dir.joinpath(f'ERROR_{file_to_process_name}')
        error_file.write_text(header_row)
        error_detail_file = self.working_dir.joinpath(
            f'ERROR_DETAIL_{file_to_process_name}')
        error_detail_file.touch()
        return success_file, error_file, error_detail_file
    <mask token>
    <mask token>

    def write_row_errors_to_files(self, errored_row, errors, error_file,
        error_detail_file):
        with open(error_file, 'a') as append_error_file:
            append_error_file.write(self.rebuild_errored_csv_row(errored_row))
            append_error_file.write('\n')
        self.write_error_details_to_file(errors, error_detail_file)

    @staticmethod
    def rebuild_errored_csv_row(row: dict):
        extra_values = row.pop(None, [])
        extra_values_str = ','.join(extra_values)
        rebuilt_row = ','.join(value for value in row.values() if value is not
            None)
        if extra_values_str:
            rebuilt_row = rebuilt_row + f',{extra_values_str}'
        return rebuilt_row

    @staticmethod
    def write_error_details_to_file(errors, error_detail_file):
        with open(error_detail_file, 'a') as append_error_detail_file:
            append_error_detail_file.write(' | '.join(
                f'[Column: {error.column}, Error: {str(error.description)}]'
                 for error in errors))
            append_error_detail_file.write('\n')
    <mask token>

    @staticmethod
    def write_row_success_to_file(succeeded_row, success_file):
        with open(success_file, 'a') as append_success_file:
            append_success_file.write(','.join(succeeded_row.values()))
            append_success_file.write('\n')

    def publish_messages(self, messages, rabbit):
        for message in messages:
            rabbit.publish_message(message=json.dumps(message),
                content_type='application/json', headers=None, exchange=
                self.processor.exchange, routing_key=self.processor.routing_key
                )

    def download_file_to_process(self, blob_to_process):
        file_to_process = self.working_dir.joinpath(blob_to_process.name)
        with open(file_to_process, 'wb') as open_file_to_process:
            self.storage_client.download_blob_to_file(blob_to_process,
                open_file_to_process)
        return file_to_process

    def upload_results(self, successes, errors, success_file, error_file,
        error_detail_file):
        files_to_upload = []
        if successes:
            files_to_upload.append(success_file)
        if errors:
            files_to_upload.append(error_file)
            files_to_upload.append(error_detail_file)
        self.upload_files_to_bucket(files_to_upload)
    <mask token>

    @staticmethod
    def delete_local_files(files_to_delete: Collection[Path]):
        logger.info('Deleting local files', files_to_delete=files_to_delete)
        for file_to_delete in files_to_delete:
            file_to_delete.unlink()

<mask token>


class BulkProcessor:

    def __init__(self, processor: Processor):
        self.working_dir = Config.BULK_WORKING_DIRECTORY
        self.processor = processor
        self.storage_client = storage.Client(project=self.processor.project_id)
        self.storage_bucket = self.storage_client.bucket(self.processor.
            bucket_name)
        self.rabbit = None
        self.db_connection_pool = None

    def run(self):
        logger.info('Checking for files in bucket', bucket_name=self.
            processor.bucket_name, prefix=self.processor.file_prefix)
        with db_helper.connect_to_read_replica_pool(
            ) as self.db_connection_pool, RabbitContext() as self.rabbit:
            blobs_to_process = self.storage_client.list_blobs(self.
                processor.bucket_name, prefix=self.processor.file_prefix)
            for blob_to_process in blobs_to_process:
                self.process_bulk_file_from_bucket(blob_to_process, prefix=
                    self.processor.file_prefix)

    def process_bulk_file_from_bucket(self, blob_to_process, prefix):
        logger.info('Processing file', file_to_process=blob_to_process.name,
            prefix=prefix)
        file_to_process = self.download_file_to_process(blob_to_process)
        success_file, error_file, error_detail_file = (self.
            initialise_results_files(file_to_process.name))
        successes, errors = self.process_file(file_to_process, success_file,
            error_file, error_detail_file)
        logger.info('Uploading results from processing file',
            file_to_process=blob_to_process.name)
        self.upload_results(successes, errors, success_file, error_file,
            error_detail_file)
        logger.info('Deleting remote file', deleted_file=blob_to_process.name)
        blob_to_process.delete()
        self.delete_local_files((file_to_process, success_file, error_file,
            error_detail_file))
        logger.info('Finished processing file', file_to_process=
            blob_to_process.name)

    def process_file(self, file_to_process, success_file, error_file,
        error_detail_file):
        try:
            with open(file_to_process, encoding='utf-8'
                ) as open_file_to_process:
                file_reader = csv.DictReader(open_file_to_process,
                    delimiter=',')
                header_validation_failures = (self.
                    find_header_validation_errors(file_reader.fieldnames))
                if header_validation_failures:
                    logger.info('Header row is invalid for file', file=
                        file_to_process.name)
                    shutil.copy(file_to_process, error_file)
                    self.write_error_details_to_file([
                        header_validation_failures], error_detail_file)
                    return 0, 1
                self.write_header_ok_to_file(error_detail_file)
                return self.process_rows(file_reader, success_file,
                    error_file, error_detail_file)
        except UnicodeDecodeError as err:
            self.write_file_encoding_error_files(err, error_file,
                error_detail_file, file_to_process)
            return 0, 1

    @staticmethod
    def write_file_encoding_error_files(err, error_file, error_detail_file,
        file_to_process):
        shutil.copy(file_to_process, error_file)
        error_detail_file.write_text(
            f'Invalid file encoding, requires utf-8, error: {err}')

    def process_rows(self, file_reader, success_file, error_file,
        error_detail_file):
        error_count = 0
        success_count = 0
        for line_number, row in enumerate(file_reader, 1):
            row_errors = self.find_row_validation_errors(line_number, row)
            if row_errors:
                error_count += len(row_errors)
                self.write_row_errors_to_files(row, row_errors, error_file,
                    error_detail_file)
            else:
                success_count += 1
                event_messages = self.processor.build_event_messages(row)
                self.publish_messages(event_messages, self.rabbit)
                self.write_row_success_to_file(row, success_file)
        logger.info('Processing results', rows_processed=line_number,
            failures=error_count, bucket_name=self.processor.bucket_name)
        return success_count, error_count

    def initialise_results_files(self, file_to_process_name):
        header_row = ','.join(column for column in self.processor.schema.keys()
            ) + '\n'
        success_file = self.working_dir.joinpath(
            f'PROCESSED_{file_to_process_name}')
        success_file.write_text(header_row)
        error_file = self.working_dir.joinpath(f'ERROR_{file_to_process_name}')
        error_file.write_text(header_row)
        error_detail_file = self.working_dir.joinpath(
            f'ERROR_DETAIL_{file_to_process_name}')
        error_detail_file.touch()
        return success_file, error_file, error_detail_file

    def find_header_validation_errors(self, header):
        try:
            header_equal(self.processor.schema.keys())(header)
        except Invalid as invalid:
            return ValidationFailure(line_number=1, column=None,
                description=str(invalid))

    def find_row_validation_errors(self, line_number, row):
        errors = []
        if row.get(None):
            errors.append(ValidationFailure(line_number, column='Multiple',
                description='Row contains too many columns'))
            return errors
        if any(value is None for value in row.values()):
            errors.append(ValidationFailure(line_number, column='Multiple',
                description='Row contains too few columns'))
            return errors
        for column, validators in self.processor.schema.items():
            for validator in validators:
                try:
                    validator(row[column], row=row, db_connection_pool=self
                        .db_connection_pool, column=column)
                except Invalid as invalid:
                    errors.append(ValidationFailure(line_number, column,
                        invalid))
        return errors

    def write_row_errors_to_files(self, errored_row, errors, error_file,
        error_detail_file):
        with open(error_file, 'a') as append_error_file:
            append_error_file.write(self.rebuild_errored_csv_row(errored_row))
            append_error_file.write('\n')
        self.write_error_details_to_file(errors, error_detail_file)

    @staticmethod
    def rebuild_errored_csv_row(row: dict):
        extra_values = row.pop(None, [])
        extra_values_str = ','.join(extra_values)
        rebuilt_row = ','.join(value for value in row.values() if value is not
            None)
        if extra_values_str:
            rebuilt_row = rebuilt_row + f',{extra_values_str}'
        return rebuilt_row

    @staticmethod
    def write_error_details_to_file(errors, error_detail_file):
        with open(error_detail_file, 'a') as append_error_detail_file:
            append_error_detail_file.write(' | '.join(
                f'[Column: {error.column}, Error: {str(error.description)}]'
                 for error in errors))
            append_error_detail_file.write('\n')
    <mask token>

    @staticmethod
    def write_row_success_to_file(succeeded_row, success_file):
        with open(success_file, 'a') as append_success_file:
            append_success_file.write(','.join(succeeded_row.values()))
            append_success_file.write('\n')

    def publish_messages(self, messages, rabbit):
        for message in messages:
            rabbit.publish_message(message=json.dumps(message),
                content_type='application/json', headers=None, exchange=
                self.processor.exchange, routing_key=self.processor.routing_key
                )

    def download_file_to_process(self, blob_to_process):
        file_to_process = self.working_dir.joinpath(blob_to_process.name)
        with open(file_to_process, 'wb') as open_file_to_process:
            self.storage_client.download_blob_to_file(blob_to_process,
                open_file_to_process)
        return file_to_process

    def upload_results(self, successes, errors, success_file, error_file,
        error_detail_file):
        files_to_upload = []
        if successes:
            files_to_upload.append(success_file)
        if errors:
            files_to_upload.append(error_file)
            files_to_upload.append(error_detail_file)
        self.upload_files_to_bucket(files_to_upload)

    def upload_files_to_bucket(self, files: Collection[Path]):
        for file_to_upload in files:
            file_blob = self.storage_bucket.blob(file_to_upload.name)
            file_blob.upload_from_filename(str(file_to_upload))

    @staticmethod
    def delete_local_files(files_to_delete: Collection[Path]):
        logger.info('Deleting local files', files_to_delete=files_to_delete)
        for file_to_delete in files_to_delete:
            file_to_delete.unlink()

import csv
import json
import logging
import shutil
from pathlib import Path
from typing import Collection

from google.cloud import storage
from structlog import wrap_logger

from toolbox.bulk_processing.processor_interface import Processor
from toolbox.bulk_processing.validators import header_equal, Invalid, ValidationFailure
from toolbox.config import Config
from toolbox.utilities import db_helper
from toolbox.utilities.rabbit_context import RabbitContext

logger = wrap_logger(logging.getLogger(__name__))


class BulkProcessor:
    def __init__(self, processor: Processor):
        self.working_dir = Config.BULK_WORKING_DIRECTORY

        self.processor = processor
        self.storage_client = storage.Client(project=self.processor.project_id)
        self.storage_bucket = self.storage_client.bucket(self.processor.bucket_name)
        self.rabbit = None
        self.db_connection_pool = None

    def run(self):
        logger.info('Checking for files in bucket', bucket_name=self.processor.bucket_name,
                    prefix=self.processor.file_prefix)
        with db_helper.connect_to_read_replica_pool() as self.db_connection_pool, RabbitContext() as self.rabbit:
            blobs_to_process = self.storage_client.list_blobs(self.processor.bucket_name,
                                                              prefix=self.processor.file_prefix)

            for blob_to_process in blobs_to_process:
                self.process_bulk_file_from_bucket(blob_to_process, prefix=self.processor.file_prefix)

    def process_bulk_file_from_bucket(self, blob_to_process, prefix):
        logger.info('Processing file', file_to_process=blob_to_process.name, prefix=prefix)
        file_to_process = self.download_file_to_process(blob_to_process)
        success_file, error_file, error_detail_file = self.initialise_results_files(file_to_process.name)
        successes, errors = self.process_file(file_to_process, success_file, error_file,
                                              error_detail_file)

        logger.info('Uploading results from processing file', file_to_process=blob_to_process.name)
        self.upload_results(successes, errors, success_file, error_file, error_detail_file)

        logger.info('Deleting remote file', deleted_file=blob_to_process.name)
        blob_to_process.delete()

        self.delete_local_files((file_to_process, success_file, error_file, error_detail_file))
        logger.info('Finished processing file', file_to_process=blob_to_process.name)

    def process_file(self, file_to_process, success_file, error_file, error_detail_file):
        try:
            with open(file_to_process, encoding="utf-8") as open_file_to_process:

                file_reader = csv.DictReader(open_file_to_process, delimiter=',')
                header_validation_failures = self.find_header_validation_errors(file_reader.fieldnames)

                if header_validation_failures:
                    logger.info('Header row is invalid for file', file=file_to_process.name)
                    shutil.copy(file_to_process, error_file)
                    self.write_error_details_to_file([header_validation_failures], error_detail_file)
                    return 0, 1  # success_count, error_count

                self.write_header_ok_to_file(error_detail_file)

                return self.process_rows(file_reader, success_file, error_file, error_detail_file)

        except UnicodeDecodeError as err:
            self.write_file_encoding_error_files(err, error_file, error_detail_file, file_to_process)
            return 0, 1  # success_count, error_count

    @staticmethod
    def write_file_encoding_error_files(err, error_file, error_detail_file, file_to_process):
        shutil.copy(file_to_process, error_file)
        error_detail_file.write_text(f'Invalid file encoding, requires utf-8, error: {err}')

    def process_rows(self, file_reader, success_file, error_file, error_detail_file):
        error_count = 0
        success_count = 0
        for line_number, row in enumerate(file_reader, 1):
            row_errors = self.find_row_validation_errors(line_number, row)
            if row_errors:
                error_count += len(row_errors)
                self.write_row_errors_to_files(row, row_errors, error_file, error_detail_file)
            else:
                success_count += 1
                event_messages = self.processor.build_event_messages(row)
                self.publish_messages(event_messages, self.rabbit)
                self.write_row_success_to_file(row, success_file)
        logger.info('Processing results', rows_processed=line_number, failures=error_count,
                    bucket_name=self.processor.bucket_name)
        return success_count, error_count

    def initialise_results_files(self, file_to_process_name):
        header_row = ','.join(column for column in self.processor.schema.keys()) + '\n'
        success_file = self.working_dir.joinpath(f'PROCESSED_{file_to_process_name}')
        success_file.write_text(header_row)
        error_file = self.working_dir.joinpath(f'ERROR_{file_to_process_name}')
        error_file.write_text(header_row)
        error_detail_file = self.working_dir.joinpath(f'ERROR_DETAIL_{file_to_process_name}')
        error_detail_file.touch()
        return success_file, error_file, error_detail_file

    def find_header_validation_errors(self, header):
        try:
            header_equal(self.processor.schema.keys())(header)
        except Invalid as invalid:
            return ValidationFailure(line_number=1, column=None, description=str(invalid))

    def find_row_validation_errors(self, line_number, row):
        errors = []

        # Check we have the correct number of columns
        # If there are extra columns the values will be stored as a list in the None key
        if row.get(None):
            errors.append(ValidationFailure(line_number,
                                            column='Multiple',
                                            description='Row contains too many columns'))
            return errors

        # If we have too few columns the value for the missing end columns will be None
        # (Empty but present columns values are an empty string, not None)
        if any(value is None for value in row.values()):
            errors.append(ValidationFailure(line_number,
                                            column='Multiple',
                                            description='Row contains too few columns'))
            return errors

        for column, validators in self.processor.schema.items():
            for validator in validators:
                try:
                    validator(row[column], row=row, db_connection_pool=self.db_connection_pool, column=column)
                except Invalid as invalid:
                    errors.append(ValidationFailure(line_number, column, invalid))
        return errors

    def write_row_errors_to_files(self, errored_row, errors, error_file, error_detail_file):
        with open(error_file, 'a') as append_error_file:
            append_error_file.write(self.rebuild_errored_csv_row(errored_row))
            append_error_file.write('\n')
        self.write_error_details_to_file(errors, error_detail_file)

    @staticmethod
    def rebuild_errored_csv_row(row: dict):
        # Pop and safely join any extra, unexpected columns which are stored in a list in the 'None' key
        extra_values = row.pop(None, [])
        extra_values_str = ','.join(extra_values)

        rebuilt_row = ','.join(value for value in row.values() if value is not None)
        if extra_values_str:
            rebuilt_row = rebuilt_row + f',{extra_values_str}'

        return rebuilt_row

    @staticmethod
    def write_error_details_to_file(errors, error_detail_file):
        with open(error_detail_file, 'a') as append_error_detail_file:
            append_error_detail_file.write(
                ' | '.join(f"[Column: {error.column}, Error: {str(error.description)}]" for error in errors))
            append_error_detail_file.write('\n')

    @staticmethod
    def write_header_ok_to_file(error_detail_file):
        with open(error_detail_file, 'a') as append_error_detail_file:
            append_error_detail_file.write('Header row is valid\n')

    @staticmethod
    def write_row_success_to_file(succeeded_row, success_file):
        with open(success_file, 'a') as append_success_file:
            append_success_file.write(','.join(succeeded_row.values()))
            append_success_file.write('\n')

    def publish_messages(self, messages, rabbit):
        for message in messages:
            rabbit.publish_message(
                message=json.dumps(message),
                content_type='application/json',
                headers=None,
                exchange=self.processor.exchange,
                routing_key=self.processor.routing_key
            )

    def download_file_to_process(self, blob_to_process):
        file_to_process = self.working_dir.joinpath(blob_to_process.name)

        with open(file_to_process, 'wb') as open_file_to_process:
            self.storage_client.download_blob_to_file(blob_to_process, open_file_to_process)
        return file_to_process

    def upload_results(self, successes, errors, success_file, error_file, error_detail_file):
        # Only upload files which contain data, we don't want to make empty error or success files
        files_to_upload = []
        if successes:
            files_to_upload.append(success_file)
        if errors:
            files_to_upload.append(error_file)
            files_to_upload.append(error_detail_file)
        self.upload_files_to_bucket(files_to_upload)

    def upload_files_to_bucket(self, files: Collection[Path]):
        for file_to_upload in files:
            file_blob = self.storage_bucket.blob(file_to_upload.name)
            file_blob.upload_from_filename(str(file_to_upload))

    @staticmethod
    def delete_local_files(files_to_delete: Collection[Path]):
        logger.info('Deleting local files', files_to_delete=files_to_delete)
        for file_to_delete in files_to_delete:
            file_to_delete.unlink()
