Validate.py

Go to the documentation of this file.

import sys
import traceback
 
 
lines_to_drop = []
class validate_data:
    """
    When data is processed, it reads all lines and verify it data is complete or not
    """
    def upc_validation(self, df, release_id):
        """ Sets 'undefined' value for null release_id
 
        Args:
            df (pandas dataframe): processed file loaded as dataframe
            release_id (str): release_id column
        Returns: 
            release_id (str)
        """
        release_id = str(release_id)
        if (release_id == "") or ("na" in release_id) or (release_id == "0") or (release_id == "None"):
            # print("Undefined")
            pass
            release_id = "undefined"
            # return release_id
        elif len(release_id) < 12 and len(release_id) >= 14:
            print("Error in {}. release_id invalid".format(release_id))
            # self.delete_rows(df, "release_id", release_id)
        # elif len(release_id) <= 12:
        #     top = 13
        #     n_zero = top-len(release_id)
        #     release_id = "0"*n_zero + release_id
        return release_id
 
    def isrc_validation(self, df, isrc):
        """ Sets 'undefined' value for null isrc_id
 
        Args:
            df (pandas dataframe): processed file loaded as dataframe
            isrc (str): release_id column
        Returns: 
            isrc (str)
        """
        isrc = str(isrc)
        if isrc == "" or isrc == "nan" or isrc == "None":
            # return isrc
            # isrc = "undefined"
            pass
        elif len(isrc) != 12:
            pass
            # print("Error in {}. ISRC invalid".format(isrc))
            # self.delete_rows(df, "isrc_id", isrc)
        return isrc
 
    def territory_validation(self, df, territory):
        """ Sets 'undefined' value for null territory_code
 
        Args:
            df (pandas dataframe): processed file loaded as dataframe
            territory (str): release_id column
        Returns: 
            territory (str)
        """
        territory = str(territory)
        if territory == "-" or territory == "N/A" or territory == "" or territory == "nan":
            territory = "XX"
            # print(territory)
        elif len(territory) != 2:
            print("Error in {}. territory invalid".format(territory))
            # self.delete_rows(df, "territory_code", territory)
        return territory
    
    def delete_rows(self, df, col, value):
        """ Deletes rows which don't satisfy the OTTO standard
 
        Args:
            df (pandas dataframe): processed file loaded as dataframe
            col (str): column name
            value (str): determines which lines should be deleted 
        Returns: lines_to_drop (list)
        """
        # print("DELETING ROWS")
        df_idx = list(df[df[col] == value].index)
        lines_to_drop[len(lines_to_drop):] = df_idx
        return lines_to_drop
 
    # def validate_columns(self, df, columns_type):
    #     """ executes de data validation
 
    #     Args:
    #         df (pandas dataframe): processed file loaded as dataframe
    #         columns_type (str): column name
    #         value (str): determines which lines should be deleted 
    #     Returns: lines_to_drop (list)
    #     """
    #     print("DATA BEFORE CLEANING: {}".format(df.shape))
    #     null_values = list(df.columns[df.isnull().any()])
    #     # print(null_values)
    #     for col in null_values:
    #         try:
    #             if list(filter(lambda item: item["value"].lower() == col and item["flagNull"] == "false", columns_type)):
    #                 df = df.dropna(subset=[col])
    #         except Exception: 
    #             continue
    #     df.reset_index(inplace=True)
    #     print("DATA AFTER CLEANING: {}".format(df.shape))
    #     return df
 
    def validation(self, df):
        """ Executes de data validation
 
        Args:
            df (pandas dataframe): processed file loaded as dataframe
        Returns: df (pandas dataframe)
                 status (str)
        """
        try:
            df["release_id"] = df["release_id"].apply(lambda x: str(x).replace(".0", ""))
            df["release_id"] = df.apply(lambda Row: self.upc_validation(df, Row["release_id"]), axis=1)
            df["isrc_id"] = df.apply(lambda Row: self.isrc_validation(df, Row["isrc_id"]), axis=1)
            df["territory_code"] = df.apply(lambda Row: self.territory_validation(df, Row["territory_code"]), axis=1)
            # df.drop(lines_to_drop, inplace=True)
            # valid = valid.unique()
            # valid = valid.all()
            status = "Ok"
            return df, status
        except Exception:
            print(sys.exc_info()[2])
            print(traceback.format_exc())
            status = "Validation Failed"
            return None, status