import logging import re import pandas as pd from service.tasks.base_attributes_generation import get_source_data from service.utils.aws_connectors import df_to_db, pd_read_sql from service.utils.data_model_utils import get_source_from_collection logger = logging.getLogger(__name__) def guess_header_position(df): """Checking the rows with maximum number of valid values picking the first one as header""" """counting number of not nan cells per each provided row""" valid_cells = [] for row in df.itertuples(index=True): # logger.info(f' row = {row}') counter = 0 for column in row: if type(column) == str: counter += 1 valid_cells.append(counter) # logger.info(f' VALID COLUMNS = {counter}') max_valid_cells = max(valid_cells) # logger.info(f' MAX NUMBER OF VALID ROWS = {max_valid_cells}') first_index = valid_cells.index(max_valid_cells) # logger.info(f' FIRST INDEX = {first_index}') return { "number_of_valid_columns": max_valid_cells, "guessed_header_row": first_index, } def evaluate_headers(header_list: list): picked_headers = [] unknown_headers = [] status = "success" for header in header_list: if "unnamed" in str(header).lower(): unknown_headers.append( {"header_name": header, "header_index": header_list.index(header)} ) status = "failure" else: picked_headers.append( {"header_name": header, "header_index": header_list.index(header)} ) return { "status": status, "picked_headers": picked_headers, "unknown_headers": unknown_headers, } def check_for_empty_columns(df: pd.DataFrame, guessed_header_row): columns = list(df.columns) loc_index = guessed_header_row + 1 empty_cols = [ {"column_name": col, "column_index": columns.index(col)} for col in columns if df[col].iloc[loc_index:].isnull().all() ] return empty_cols def guess_fields(collection_id, schema_name, bucket=None, file_name=None): # Get file name from collection_id if file_name is None: source = get_source_from_collection(schema_name, collection_id) sp = source.split("/") bucket = sp[0] file_name = "/".join(sp[1:]) # Load the file from S3 source_df = get_source_data(bucket, file_name) field_names = list(source_df.columns) evaluation_response = evaluate_headers(field_names) response_status = evaluation_response["status"] message = "" if response_status == "failure": sample_df = source_df.iloc[:100] guessing_response = guess_header_position(sample_df) empty_cols = check_for_empty_columns( source_df, guessing_response["guessed_header_row"] ) empty_cols_numbers = [str(x["column_index"] + 1) for x in empty_cols] picked_headers_count = len(evaluation_response["picked_headers"]) unknown_headers_count = len(evaluation_response["unknown_headers"]) all_headers_count = picked_headers_count + unknown_headers_count if len(empty_cols) > 0: columns = ", ".join(empty_cols_numbers) message = f"Failed to read headers. {unknown_headers_count} out of {all_headers_count} columns have missing header. As well columns {columns} seem to be empty. Please remove empty columns and make sure that headers row starts from top left cell of the file." else: message = f"Failed to read headers. {unknown_headers_count} out of {all_headers_count} columns have missing header. Please make sure that headers row starts from top left cell of the file." elif response_status == "success": message = "success" """ If we don't get stuff from field labels, try to guess from column names and take first one that matches our search pattern. Don't use "total" as search, as this could be in various combinations for both, monetary and quantity, as we've seen on existing data sets. See datasets BAYNK Austin, Baynk Boston 1 and ULO CyberSpirits to see what I mean why not to use "total". """ all_labels_df = get_system_labels(exclude_non_regex=False) labels_df = all_labels_df[~all_labels_df["a_field_name_regex"].isna()] labels_df.sort_values( by="a_matching_order", ascending=True, inplace=True, ignore_index=True ) guesslist = {} for row in labels_df.itertuples(index=False): guesslist[row[1]] = row[2] all_guess_list = {} for row in all_labels_df.itertuples(index=False): all_guess_list[row[1]] = row[2] pre_guessed_fields = get_guessed_labels() values_list = [] guessed_fields = [] for field in field_names: system_field = "" for guess_field, regex_pattern in guesslist.items(): if guess_field not in guessed_fields: regex = re.compile(regex_pattern) result = regex.search( field.lower().replace("_", "").replace(" ", "") ) if result is not None: system_field = guess_field guessed_fields.append(guess_field) break """ If system_field was not successfully assigned attempt to find it in guessed_labels table""" if system_field == "": try: temp_field = strip_string(field) temp_guessed_field = pre_guessed_fields[temp_field] if temp_guessed_field not in guessed_fields: """Checking that the suggested guessed field is present in guesslist Checking if the field was not hidden""" all_guess_list[temp_guessed_field] system_field = temp_guessed_field guessed_fields.append(temp_guessed_field) except Exception: pass values_list.append( { "collection_id": collection_id, "column_number": field_names.index(field), "system_field_id": 0, "system_field_name": system_field, "file_field_name": field, } ) response_df = pd.DataFrame(values_list) # Insert guessed labels to table df_to_db( df=response_df, schema=schema_name, table_name="guessed_file_upload_fields", if_exists="append", ) response_dict = {"status": str(response_status), "message": str(message)} return response_dict def get_system_labels( exclude_non_regex: bool = True, exclude_hidden: bool = True ) -> pd.DataFrame: """ Getting the list of all system_labels :param exclude_non_regex: default True - Bool to avoid listing fields without regex data :param exclude_hidden: default False - Bool to avoid listing fields that are hidden from front end :return: """ filter_list = [] if exclude_non_regex: filter_list.append(" a_field_name_regex IS NOT NULL ") if exclude_hidden: filter_list.append(" a_hidden IN (0, 2) ") if exclude_non_regex or exclude_hidden: filter_string = " WHERE" + "AND".join(filter_list) else: filter_string = "" sql = f""" SELECT a_id, a_system_name, a_field_name_regex, a_matching_order, a_hidden FROM commons.system_label {filter_string}; """ df = pd_read_sql(sql) return df def get_guessed_labels() -> dict: """Getting list of file_label -> system_label pairs""" sql = """ SELECT file_label, system_label FROM commons.guessed_labels; """ df = pd_read_sql(sql) labels_dict = {} for row in df.itertuples(index=False): labels_dict[row[0]] = row[1] return labels_dict def strip_string(string: str): """Removing all special cahracters from string""" return re.sub(r"\W", "", str(string)).replace("_", "").lower()