"""Validation functions for Orchard bulk uploads.""" from collections import OrderedDict from os import makedirs import openpyxl as op from openpyxl.comments import Comment from openpyxl.styles.borders import ( BORDER_THICK, BORDER_THIN, ) from constants.cell_validation_map import CELL_VALIDATION_MAP from constants.intra_row_validation_map import INTRA_ROW_VALIDATION_MAP from constants.trans_row_validation_map import ( GROUP_UNIQUE_KEY, TRANS_ROW_VALIDATION_MAP ) from constants.data_headers import DATA_HEADERS from constants.colors import ( NORMAL_BORDER_COLOR, ERROR_BORDER_COLOR, ERROR_CELL_COLOR, ERROR_RELEASE_ROW_COLOR, ERROR_RELEASE_CELL_COLOR ) import config logger = config.get_logger(__name__) # =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-= # Utility Methods # # The following methods assist the engine in validating the data in the # worksheet. # # =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-= def sanitize_header(value): """Clean a value by stripping whitespace and all non-alphanumeric characters. Convert the value to lowercase. Convert to snake case. Args: value (str): The value to clean. Returns: str: The cleaned value. """ # Strip whitespace and all non-alphanumeric characters value = ''.join(e for e in value if e.isalnum()) # Convert to lowercase value = value.lower() # Convert to snake case value = value.replace(' ', '_') # Replace - with _ value = value.replace('-', '_') return value def prepare_header(ws): """Prepare the header of the worksheet. Args: ws (openpyxl.worksheet.worksheet.Worksheet): The worksheet. Returns: list: The prepared header. """ # Extract header from the first row header = [cell.value for cell in ws[1]] # Create a reverse dictionary of the data headers reverse_dict = {v: k for k, v in DATA_HEADERS.items()} # Substitute the header using the reverse dictionary or sanitize the header header = [reverse_dict.get(cell, sanitize_header(cell)) for cell in header] return header def add_validation_comment_to_cell(cell, msg): """Add a validation comment to the cell. Args: cell (openpyxl.cell.cell.Cell): The cell to add the comment to. msg (list): The validation messages. Returns: None """ # If there are any validation messages, add a comment to the cell cell.comment = Comment( '\n'.join(msg), 'The Orchard Bulk Validation Tool' ) # Change the comment size cell.comment.width = 300 cell.comment.height = 200 # Change the cell fill color to a yellow cell.fill = op.styles.PatternFill( start_color=ERROR_CELL_COLOR, end_color=ERROR_CELL_COLOR, fill_type='solid' ) def add_row_errors(ws, error_col, row, msg): """Add an error column to the worksheet and add comments to the cells with validation messages. Args: ws (openpyxl.worksheet.worksheet.Worksheet): The worksheet to modify. error_col (int): The column number for the error column. row (openpyxl.cell.cell.Cell): The row to add the error column and comments to. msg (list): The validation messages. Returns: None """ # Add an error column to the header row at the calculated column ws.cell(row=1, column=error_col, value='ROW ERRORS') # Add Errors to row at the calculated column ws.cell(row=row[0].row, column=error_col, value='Read comment for errors') ws.cell(row=row[0].row, column=error_col).comment = Comment( '\n\n'.join(msg), 'The Orchard Bulk Validation Tool' ) ws.cell(row=row[0].row, column=error_col).comment.width = 600 ws.cell(row=row[0].row, column=error_col).comment.height = 400 # Define cell styles # Normal invisible side normal_side = op.styles.Side(style=BORDER_THIN, color=NORMAL_BORDER_COLOR) # Red side red_side = op.styles.Side(style=BORDER_THICK, color=ERROR_BORDER_COLOR) # Use the max column value from the worksheet to avoid recalculating it max_col = ws.max_column # Iterate through each cell in the row for idx, cell in enumerate(row, start=1): # if last cell, skip if idx == max_col-1: continue # Set the left and right borders based on cell position left_border = red_side if idx == 1 else normal_side # We don't want to add a right border to the error reporting cell right_border = red_side if idx == max_col-2 else normal_side # Define the border for the cell cell.border = op.styles.Border( left=left_border, right=right_border, top=red_side, bottom=red_side ) def get_rows_by_unique_key(ws, header, field): """Index rows by a field. Args: ws (openpyxl.worksheet.worksheet.Worksheet): The worksheet to index. header (list): The header row of the worksheet, conformed. field (str): The field to index by. Returns: dict: A dictionary of rows of cell objects indexed by the field. """ max_row = ws.max_row rows_by_unique_key = OrderedDict() # Iterate over the rows starting from the second row for row in ws.iter_rows(min_row=2, max_row=max_row): # Create a dictionary for the current row using header values as keys row_dict = {head: cell for head, cell in zip(header, row)} row_dict['row_num'] = row[0].row # Add row number to the dictionary # Use the specified field as the unique key unique_key = row_dict[field].value if unique_key: rows_by_unique_key.setdefault(unique_key, []).append(row_dict) return rows_by_unique_key def process_cell(field, cell): """Process a cell based on the field name. Args: field (str): The name of the field. cell (str): The value of the cell. Returns: list: A list of validation message tuples. """ # Create a reverse dictionary of the data headers reverse_dict = {v: k for k, v in DATA_HEADERS.items()} field = reverse_dict.get(field, sanitize_header(field)) result = [] # Check if the field is in the validation map if field in CELL_VALIDATION_MAP.keys(): # Iterate over the validation functions for the field for validation in CELL_VALIDATION_MAP[field]: # Get name of function at validation[0] if isinstance(validation, tuple): validator_name = validation[0].__name__ args = validation[1:] validation_result = validation[0](cell.value, *args) else: validator_name = validation.__name__ validation_result = validation(cell.value) # If a validation result, add it to the result list if validation_result: if config.DEBUG: logger.info(f'{cell.coordinate} - Validation ' f'"{validator_name}" failed on {cell.value}: ' f'{validation_result}') result.append(validation_result) elif config.DEBUG: logger.info(f'{cell.coordinate} - Validation ' f'"{validator_name}" passed on {cell.value}') else: logger.warning(f'No validation rules for {field}') return result # =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-= # Validation Enigines # # The following methods are the engines that iterate over the rows in the # worksheet and apply the validation functions to the cells in the rows. # # They implement the validation rules defined in the maps in the constants. # =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-= def validate_cells(ws): """Validate the cells in the worksheet. Add comments to the cells with validation messages. This function iterates over the cells in the worksheet and passes them to a helper function that runs multiple validation functions on the cell (as specified in the CELL_VALIDATION_MAP constant). If the validation functions return a validation message, the message is added to the cell as a comment. If the validation functions return a correction, the cell value is updated with the correction. Args: ws (openpyxl.worksheet.worksheet.Worksheet): The worksheet to validate. Returns: None """ max_row = ws.max_row max_column = ws.max_column for row in ws.iter_rows(min_row=2, max_row=max_row, max_col=max_column): # Reset list of cell corrections for the row row_corrections = [] # For each cell in the row for cell in row: # Reset any correction value from a previous cell correction = None # Validate the cell based on the column heading try: # Get the column heading column_heading = ws.cell(row=1, column=cell.column).value # Process the cell based on the column heading comments = process_cell(column_heading, cell) # If there are any validation messages, combine them msg = [] for comment in comments: if len(comment) == 3: correction = comment[2] msg.append( f'{comment[0]}: AUTOCORRECTED - {comment[1]}') else: msg.append(f'{comment[0]}: {comment[1]}') if msg: if correction: cell.value = correction add_validation_comment_to_cell(cell, msg) # Collect the corrections for the row row_corrections.append( f'{column_heading} ({cell.coordinate}) - ' f'{"; ".join(msg)}') # If the value can't be converted to an integer, ignore it and # move on to the next cell except ValueError: pass # If there are any corrections for the row, add them to the last cell if row_corrections: # Add a new column for corrections in the header row. If the column # already exists. noop ws.cell(row=1, column=max_column + 1, value='CELL ERRORS') # Add a new column for corrections ws.cell(row=row[0].row, column=max_column + 1, value='Read comment for corrections') ws.cell(row=row[0].row, column=max_column + 1).comment = Comment( '\n\n'.join(row_corrections), 'The Orchard Bulk Validation Tool' ) ws.cell(row=row[0].row, column=max_column + 1).comment.width = 600 ws.cell(row=row[0].row, column=max_column + 1).comment.height = 400 def validate_rows(ws): """Validate the rows in the worksheet. Add comments to the rows with validation messages. Add a red outline around the row if there are any validation messages. This function iterates over the rows in the worksheet and for each row, runs multiple validation functions on the row (as specified in the INTRA_ROW_VALIDATION_MAP constant). If the validation functions return a validation message, the message is added to the row as a comment. If the validation functions returns no validation message, the row is marked as valid. Args: ws (openpyxl.worksheet.worksheet.Worksheet): The worksheet to validate. Returns: """ max_row = ws.max_row max_column = ws.max_column error_col = ws.max_column + 1 # Prepare the header header = prepare_header(ws) # Iterate over the rows starting from the second row for row in ws.iter_rows(min_row=2, max_row=max_row, max_col=max_column): # Process the row based on the intra-row validation map comments = [] # Restore the validations to the default validations = [v.copy() for v in INTRA_ROW_VALIDATION_MAP] for validation in validations: # Pop the description and validator from the validation description = validation.pop('description') logger.info(f'Row {row[0].row}:' f'Validating {description}') validator = validation.pop('validator') # Get the validation result using the rest of the fields as # arguments result = validator(row, header, description, **validation) # If there is a validation result, add it to the comments if result: comments.append(result) logger.info(f'Validation failed: {result}') else: logger.info('Validation passed') # If there are any validation messages, combine them msg = [] for comment in comments: msg.append(f'{comment[0]}: {comment[1]}') if msg: add_row_errors(ws, error_col, row, msg) def validate_releases(ws): """Validate the releases in the worksheet. Args: ws (openpyxl.worksheet.worksheet.Worksheet): The worksheet to validate. Returns: None """ # Prepare the header header = prepare_header(ws) header_indexes = \ {header: idx for idx, header in enumerate(header, start=1)} # Get the rows indexed by the unique key rows_by_unique_key = get_rows_by_unique_key(ws, header, GROUP_UNIQUE_KEY) # Process the rows based on the functions in TRANS_ROW_VALIDATION_MAP for unique_key, rows in rows_by_unique_key.items(): # Restore the validations to the default validations = [v.copy() for v in TRANS_ROW_VALIDATION_MAP] for validation in validations: # Pop the description and validator from the validation description = validation.pop('description') logger.info(f'Validating {description} for {unique_key}') validator = validation.pop('validator') # Get the validation result using the rest of the fields as # arguments result = validator( rows, header, GROUP_UNIQUE_KEY, description, **validation) # If there is a validation result, add it to the comments if result: # TODO: Should this be in the row, or using a msg var like in # the other validation functions? for row in rows: row['comments'] = \ row.get('comments', []) + [result[0] + ': ' + result[1]] # noqa logger.info(f'Validation failed for {unique_key}: {result}') else: logger.info(f'Validation passed for {unique_key}') # Add results as comments to the worksheet max_column = ws.max_column+1 for unique_key, rows in rows_by_unique_key.items(): first_row = 0 last_row = len(rows)-1 for row_index, row in enumerate(rows): # Add comments to the last cell in the row if 'comments' in row: ws.cell(row=1, column=max_column, value='RELEASE ERRORS') # Check if there is already a comment in the cell if ws.cell(row=row['row_num'], column=max_column).comment: # Add the new comment to the existing comment comment = \ ws.cell(row=row['row_num'], column=max_column).comment comment.text += '\n\n' + '\n\n'.join(row['comments']) ws.cell(row=row['row_num'], column=max_column).comment = \ comment else: comment = Comment( '\n\n'.join(row['comments']), 'The Orchard Bulk Validation Tool' ) ws.cell(row=row['row_num'], column=max_column).comment = \ comment comment.width = 600 comment.height = 400 # Put 'Read comment for errors' in the cell ws.cell(row=row['row_num'], column=max_column, value='Read comment for errors') # Get the unique key cell unique_key_cell = ws.cell( row=row['row_num'], column=header_indexes[GROUP_UNIQUE_KEY]) # Get the last cell in the row error_cell = ws.cell(row=row['row_num'], column=max_column) color = ERROR_RELEASE_CELL_COLOR # Highlight the unique key cell unique_key_cell.fill = op.styles.PatternFill( start_color=color, end_color=color, fill_type='solid' ) # Highlight the error cell error_cell.fill = op.styles.PatternFill( start_color=color, end_color=color, fill_type='solid' ) # Add a border to the unique key cell. Be aware of if it's the # first or last row in the group. All rows in the group will # have side borders. If it's the first, add a top border. If # it's the last, add a bottom border. normal_side = op.styles.Side( style=BORDER_THIN, color=NORMAL_BORDER_COLOR) color_side = op.styles.Side( style=BORDER_THICK, color=ERROR_RELEASE_ROW_COLOR) top_side = normal_side bottom_side = normal_side # Adjust the top and bottom sides based on the row index if row_index == first_row: top_side = color_side if row_index == last_row: bottom_side = color_side # Reset border for cell unique_key_cell.border = op.styles.Border( left=op.styles.Side(style=None), right=op.styles.Side(style=None), top=op.styles.Side(style=None), bottom=op.styles.Side(style=None) ) # Add the border to the cell unique_key_cell.border = op.styles.Border( left=color_side, right=color_side, top=top_side, bottom=bottom_side ) # Add a comment to the first cell in the row if row_index == first_row: unique_key_cell.comment = Comment( 'This Release has issues: ' + '\n\n'.join(row['comments']), 'The Orchard Bulk Validation Tool' ) unique_key_cell.comment.width = 600 unique_key_cell.comment.height = 400 def validate_worksheet(ws): """Validate the worksheet. Args: ws (openpyxl.worksheet.worksheet.Worksheet): The worksheet to validate. Returns: None """ # Find the last row that has a non-blank value in the first column and the # last column max_row = ws.max_row max_column = ws.max_column # Set Some default styles default_font = op.styles.Font() default_fill = op.styles.PatternFill(fill_type=None) default_border = op.styles.Border(left=op.styles.Side(style=None), right=op.styles.Side(style=None), top=op.styles.Side(style=None), bottom=op.styles.Side(style=None)) default_alignment = op.styles.Alignment() logger.info('Resetting all cell styles to default.') # Iterate over all cells in the worksheet for row in ws.iter_rows(max_row=max_row, max_col=max_column): for cell in row: cell.font = default_font cell.fill = default_fill cell.border = default_border cell.alignment = default_alignment logger.info('Validating cells.') # Validate the `cells validate_cells(ws) # Save the workbook locally if in debug mode if config.DEBUG: makedirs('data/output', exist_ok=True) ws.parent.save('data/output/debug.xlsx') logger.info('Validating rows.') # Validate the rows validate_rows(ws) logger.info('Validating releases.') # Save the workbook locally if in debug mode if config.DEBUG: ws.parent.save('data/output/debug.xlsx') # Validate the releases validate_releases(ws)