Active_prgs/Redgrave/Amazon_NamesNormQC.py

"""

Amazon-NamesNormQC

Created by:
Emanuel Borges
11.21.2024

This program will assist with the process of performing Names Normalization QC on the Amazon privilege logs.

"""

import os, uuid, pickle, re
import MyCode.Active_prgs.Redgrave.Amazon_PrivLogQC
from dataclasses import dataclass, field
from typing import List, Optional
from collections import namedtuple
from win32com.client import Dispatch


@dataclass
class Person:
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    work_email_address: Optional[str] = None
    alt_work_email_address: Optional[str] = None
    _id: uuid.UUID = field(default_factory=uuid.uuid4)
    is_attorney: Optional[str] = None
    split_role_date_range: Optional[str] = None
    sidley_validated: Optional[str] = None
    category: Optional[str] = None
    organization: Optional[str] = None
    job_title: Optional[str] = None
    business_title: Optional[str] = None
    full_name_preferred: Optional[str] = None
    login: Optional[str] = None
    department_fine: Optional[str] = None
    addressed_during_caag: Optional[str] = None
    #last_updated: Optional[str] = None
    full_name_overide: Optional[str] = None

    def __post_init__(self):
        """Convert all string fields to uppercase."""
        if self.first_name:
            self.first_name = self.first_name.strip().upper()
        if self.last_name:
            self.last_name = self.last_name.strip().upper()
        if self.work_email_address:
            self.work_email_address = self.work_email_address.strip().upper()
        if self.alt_work_email_address:
            self.alt_work_email_address = self.alt_work_email_address.strip().upper()
        if self.is_attorney:
            self.is_attorney = self.is_attorney.strip().upper()
        if self.split_role_date_range:
            self.split_role_date_range = self.split_role_date_range.strip().upper()
        if self.sidley_validated:
            self.sidley_validated = self.sidley_validated.strip().upper()
        if self.category:
            self.category = self.category.strip().upper()
        if self.organization:
            self.organization = self.organization.strip().upper()
        if self.job_title:
            self.job_title = self.job_title.strip().upper()
        if self.business_title:
            self.business_title = self.business_title.strip().upper()
        if self.full_name_preferred:
            self.full_name_preferred = self.full_name_preferred.strip().upper()
        if self.login:
            self.login = self.login.strip().upper()
        if self.department_fine:
            self.department_fine = self.department_fine.strip().upper()
        if self.addressed_during_caag:
            self.addressed_during_caag = self.addressed_during_caag.strip().upper()
        #if self.last_updated:
        #    self.last_updated = self.last_updated.strip().upper()

@dataclass
class PeopleList:
    people: List[Person] = field(default_factory=list)

    def add_person(self, person: Person):
        self.people.append(person)
        #print(f"Added person: {person}")


    def search_by_email(self, emailAddress:str) -> Optional[Person]:
        for person in self.people:
            if person.work_email_address == emailAddress:
                return person
        return None
        
    def list_people(self):
        for person in self.people:
            print(person)

    def update_full_Name_overide(self, emailAddress:str, fullNameOverideValue) -> Optional[Person]:
        valueUpdated = False
        for person in self.people:
            if person.work_email_address == emailAddress.upper():
                person.full_name_overide = fullNameOverideValue.upper()
                valueUpdated = True
                ## Give a quik warning as you add the override value into the database if the last name differs.
                if "," in fullNameOverideValue:
                    lastName = fullNameOverideValue.split(",")[0]
                else:
                    lastName = fullNameOverideValue.split(" ")[-1]
                if lastName.upper() == person.last_name:
                    pass
                else:
                    print(f"WARNING: Overide last name value {lastName.upper()} does not match {person.last_name}.")
        if valueUpdated == False:
            print(f"WARNING: No email address match for {emailAddress} found.")


class NamesVerification(object):
    """A class for automating the process of performing QC on the names within the Amazon privilege logs."""
    version = '0.6.0'


    def __init__(self, cleanedDatExportFileName, masterAttorneyListFileName,fullNameOveridesFileName, forceNewPklFile = False, Encoding = 'UTF8'):
        """Initializes the data structures. cleanedDatExportFileName should be the full path to the file.
        Assumes the first row of the data file is the header and first column is DocID.
        Assumes the MAL is a spreadsheet (for now).
        MAL gets saved to a pkl file for performance reasons.  pkl will be used unless forceNewPklFile is set to true"""
        pklFileName = os.path.splitext(masterAttorneyListFileName)[0] + ".pkl"

        print("Initializing data structures...")
        if forceNewPklFile:
            print("Creating MAL structure...")
            self.malPeopleList = PeopleList()
            self.__IngestMALSpreadsheet(masterAttorneyListFileName)
            print("MAL structure created.")
            print("Loading full name overide values...")
            self.__LoadFullNameOverideValues(fullNameOveridesFileName)
            print("Full name overide values loaded.")
            print("Creating pickle backup...")
            self.__SaveMalToPkl(pklFileName)
            print("Pickle backup created.")
        else:
            if os.path.exists(pklFileName):
                print("Loading MAL structure from pickle file...")
                self.malPeopleList = self.__LoadMalFromPkl(pklFileName)
                print("MAL structure loaded.")
            else:
                print("Pickle file doesnt exist.")
                print("Creating MAL structure...")
                self.malPeopleList = PeopleList()
                self.__IngestMALSpreadsheet(masterAttorneyListFileName)
                print("MAL structure created.")
                print("Loading full name overide values...")
                self.__LoadFullNameOverideValues(fullNameOveridesFileName)
                print("Full name overide values loaded.")
                print("Creating pickle backup...")
                self.__SaveMalToPkl(pklFileName)
                print("Pickle backup created.")
                
##        self.malPeopleList = PeopleList()
##        
##        print("Creating MAL structure...")
##        self.__IngestMALSpreadsheet(masterAttorneyListFileName)
##        print("MAL structure created.")
##        print("Creating pickle backup...")


    def __IngestMALSpreadsheet(self, masterAttorneyListFileName):
        """Pseudo-private method which will open an Excel spreadsheet and ingest the values into the peoplelist dataclass."""
        ## There doenst seem to be a consistent value in the "row" column in the MAL, so setting these parameters here to avoid gap issues.

        excelTabParametersList = [{"tabName":"Attorneys", "beginRowNumber":2, "endRowNumber":10919, "beginColNumber":2, "endColNumber":17},
                               {"tabName":"Downgrades", "beginRowNumber":2, "endRowNumber":572, "beginColNumber":2, "endColNumber":16}]

 #       excelTabParametersList = [{"tabName":"Attorneys", "beginRowNumber":2, "endRowNumber":30, "beginColNumber":2, "endColNumber":16},
 #                              {"tabName":"Downgrades", "beginRowNumber":2, "endRowNumber":30, "beginColNumber":2, "endColNumber":15}]

        spreadsheetFileMappingMatrix = {"First Name":"first_name", "Last Name":"last_name", "Work Email":"work_email_address", "Alt Work Email":"alt_work_email_address", "Is Attorney": "is_attorney",
                                  "Split Role -  Attorney Capacity Date Range":"split_role_date_range", " Validated by OC??":"sidley_validated", "Category": "category", "Organization":"organization", "Job Title":"job_title",
                                  "Business Title":"business_title", "Full Name (Preferred)":"full_name_preferred", "Login":"login", "Department (Fine)":"department_fine", "Addressed during CAAG":"addressed_during_caag",
                                        "Last Updated":"last_updated"}

        xlApp = Dispatch('Excel.Application')
        xlBook = xlApp.Workbooks.Open(masterAttorneyListFileName)
        
        for excelTab in excelTabParametersList:
            sht = xlBook.Worksheets(excelTab['tabName'])
            print(f"Ingesting sheet {excelTab['tabName']}.")
            excelFieldPositionMatrix = {}
            for col in range (excelTab['beginColNumber'], excelTab['endColNumber'] +1):
                excelFieldPositionMatrix[sht.Cells(1,col).Value] = col
            for row in range(excelTab['beginRowNumber'], excelTab['endRowNumber'] +1):
                #print(row)
                ##  TODO: Refactor the excelTabParametersList later. Didnt realize columns were not consistent.
                if excelTab['tabName'] == 'Attorneys':
                    self.malPeopleList.add_person(Person(is_attorney = sht.Cells(row,excelFieldPositionMatrix['Is Attorney']).Value,
                                                         split_role_date_range = sht.Cells(row,excelFieldPositionMatrix['Split Role -  Attorney Capacity Date Range']).Value,
                                                         sidley_validated =  sht.Cells(row,excelFieldPositionMatrix[' Validated by OC?']).Value,
                                                         category = sht.Cells(row,excelFieldPositionMatrix['Category']).Value,
                                                         organization = sht.Cells(row,excelFieldPositionMatrix['Organization']).Value,
                                                         last_name = sht.Cells(row,excelFieldPositionMatrix['Last Name']).Value,
                                                         first_name = sht.Cells(row,excelFieldPositionMatrix['First Name']).Value,
                                                         work_email_address = sht.Cells(row,excelFieldPositionMatrix['Work Email']).Value,
                                                         alt_work_email_address = sht.Cells(row,excelFieldPositionMatrix['Alt Work Email']).Value,
                                                         job_title = sht.Cells(row,excelFieldPositionMatrix['Job Title']).Value,
                                                         business_title = sht.Cells(row,excelFieldPositionMatrix['Business Title']).Value,
                                                         full_name_preferred = sht.Cells(row,excelFieldPositionMatrix['Full Name (Preferred)']).Value,
                                                         login = sht.Cells(row,excelFieldPositionMatrix['Login']).Value,
                                                         department_fine = sht.Cells(row,excelFieldPositionMatrix['Department (Fine)']).Value,
                                                         addressed_during_caag = sht.Cells(row,excelFieldPositionMatrix['Comments']).Value))
                                                         #addressed_during_caag = sht.Cells(row,excelFieldPositionMatrix['Addressed during CAAG']).Value))
                                                         #last_updated = sht.Cells(row,excelFieldPositionMatrix['Last Updated']).Value ))
                
                else:
                    self.malPeopleList.add_person(Person(is_attorney = sht.Cells(row,excelFieldPositionMatrix['Is Attorney']).Value,
                                                         split_role_date_range = sht.Cells(row,excelFieldPositionMatrix['Split Role -  Attorney Capacity Date Range']).Value,
                                                         sidley_validated =  sht.Cells(row,excelFieldPositionMatrix['Validated by OC?']).Value,
                                                         organization = sht.Cells(row,excelFieldPositionMatrix['Organization']).Value,
                                                         last_name = sht.Cells(row,excelFieldPositionMatrix['Last Name']).Value,
                                                         first_name = sht.Cells(row,excelFieldPositionMatrix['First Name']).Value,
                                                         work_email_address = sht.Cells(row,excelFieldPositionMatrix['Work Email']).Value,
                                                         alt_work_email_address = sht.Cells(row,excelFieldPositionMatrix['Alt Work Email']).Value,
                                                         job_title = sht.Cells(row,excelFieldPositionMatrix['Job Title']).Value,
                                                         business_title = sht.Cells(row,excelFieldPositionMatrix['Business Title']).Value,
                                                         full_name_preferred = sht.Cells(row,excelFieldPositionMatrix['Full Name (Preferred)']).Value,
                                                         login = sht.Cells(row,excelFieldPositionMatrix['Login']).Value,
                                                         department_fine = sht.Cells(row,excelFieldPositionMatrix['Department (Fine)']).Value,
                                                         addressed_during_caag = sht.Cells(row,excelFieldPositionMatrix['Addressed during CAAG']).Value))                                         

        xlBook.Close()

    def __SaveMalToPkl(self, pklFileName):
        """Pseudo-private method which will save the current MAL people list object to a pkl file, for performance reasons."""
        outputFile = open(pklFileName,'wb')
        pickle.dump(self.malPeopleList,outputFile)
        outputFile.close()

    def __LoadMalFromPkl(self, pklFileName):
        """Pseudo-private method which will load a MAL people list object from a pkl file, for performance reasons."""
        contents = open(pklFileName, 'rb')
        obj = pickle.load(contents)
        contents.close()
        return obj

    def __LoadFullNameOverideValues(self, fullNameOveridesFileName):
        """Pseudo-private method which will update the MAL people list object with the full name overide values."""
        contents = open(fullNameOveridesFileName).readlines()
        for line in contents:
            line = line.replace("\n","")
            emailAddress,fullNameOverideValue = line.split("|")
            
            self.malPeopleList.update_full_Name_overide(emailAddress, fullNameOverideValue)

    def SmartDedupeSet(self, currentSet):
        """Pseudo-private method that attempts to do some additional deduplication of the values in a set by lowering all values and deduplicating.  Returns a lowered deduplicated set."""
        newSet = set()
        for val in currentSet:
            newSet.add(val.lower())
        return newSet

if __name__ == '__main__':
    cleanedDatExportFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\Data Exports\VEAS\VEAS_Log_Data_Export_Converted.txt"
    #cleanedDatExportFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\20241202 - FTC-CID\PLOG All IDs (20241202)\PLOG All IDs (20241202)_Converted_SubSetOnly.txt"
    #cleanedDatExportFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\20241202 - FTC-CID\PLOG All IDs (20241202)\TEST-PLOG.txt"
    #cleanedDatExportFileName = r"C:\Users\eborges\AppData\Local\Programs\Python\Python312\MyCode\JN\_Temp2\20241115_PrivLogWorking_CAAG\PrivLogExport_20241113_CAAG_Converted.txt"
    #cleanedDatExportFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\20241122 - VEAS CAAG 20241206\export_20241122_160117_Converted.txt"
    #cleanedDatExportFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\20241122 - VEAS CAAG 20241206\TEST.txt"
    #masterAttorneyListFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\Amazon_ Master Attorney List  2024.11.06(7045550.3).xlsx"
    masterAttorneyListFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\Amazon_ Master Attorney List  2024.12.1(7045413.15).xlsx"
    #masterAttorneyListFileName = r"C:\Test_Dir\Amazon\TEST-MAL.xlsx"
    #fullNameOveridesFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\FullNameOverides.txt"
    #fullNameOveridesFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\FullNameOverides - Copy.txt"
    fullNameOveridesFileName = r"C:\Users\eborges\OneDrive - Redgrave LLP\Documents\Cases\Amazon\_PrivLogQCProcess\Consilio\VEAS-MasterAttorneyList\FullNameOverides.txt"
    

    nv = NamesVerification(cleanedDatExportFileName, masterAttorneyListFileName, fullNameOveridesFileName)
    #nv.malPeopleList.list_people()

    qcP = MyCode.Active_prgs.Redgrave.Amazon_PrivLogQC.QcPrivLog(cleanedDatExportFileName)
    print(nv.malPeopleList.search_by_email('crespojp@amazon.com'.upper()))
    #print(nv.malPeopleList.search_by_email('crespojp@amazon.com'.upper()))
    workList = qcP.metadataValuesDict.keys()
    outputFile = open(r"C:\Test_Dir\Amazon\NameNormOutputText.txt",'w')
    for docID in workList:
        #metadataFieldValues = qcP.metadataValuesDict[docID]._asdict()['toValues']
        #formattedFieldValues = qcP.formattedValuesDict[docID]._asdict()['toValues']
        #metadataFieldValues = qcP.metadataValuesDict[docID]._asdict()['ccValues']
        #formattedFieldValues = qcP.formattedValuesDict[docID]._asdict()['ccValues']
        #metadataFieldValues = qcP.metadataValuesDict[docID]._asdict()['bccValues']
        #formattedFieldValues = qcP.formattedValuesDict[docID]._asdict()['bccValues']
        metadataFieldValues = qcP.metadataValuesDict[docID]._asdict()['fromValues']
        formattedFieldValues = qcP.formattedValuesDict[docID]._asdict()['fromValues']
        formattedAttorneyValues = set()
        for formattedValue in formattedFieldValues:
            if "*" in formattedValue:
                formattedAttorneyValues.add(formattedValue.upper())
        
        if metadataFieldValues:
            matchedMetadataValues = set()
            for nameItem in metadataFieldValues:
                ## First test to see if there is a valid email address.
                resultSet = set()
                results = re.findall(qcP.allPossibleEmailAddressesRegExPattern, nameItem)
                if results:
                    for result in results:
                        resultSet.add(result)
                    if len(resultSet) >1:
                        resultSet = nv.SmartDedupeSet(resultSet)
                    if len(resultSet) >1:
                        print("ERROR multiple email **unique** email addresses in one item.")
                        print(resultSet)
                        print("\n")
                    else:
                        personMatch = nv.malPeopleList.search_by_email(resultSet.pop().upper())
                        if personMatch:
                            if personMatch.full_name_overide:
                                fullName = personMatch.full_name_overide
                            elif personMatch.full_name_preferred:
                                #print(personMatch.full_name_preferred)
                                ##  Going to need to do a bit of replacing to remove some information that is just never in the formatted.
                                fullPreferredName = personMatch.full_name_preferred
                                fullPreferredName = fullPreferredName.replace('(LEGAL)','')
                                fullPreferredName = fullPreferredName.replace('(SHE, HER)','')
                                fullPreferredName = fullPreferredName.replace('(SHE HER)','')
                                preferedLastName, preferedFirstName = fullPreferredName.split(',')
                                preferedLastName = preferedLastName.strip()
                                preferedFirstName = preferedFirstName.strip()
                                preferedFirstName = preferedFirstName.split(" ")[0]
                                fullName = f"{preferedFirstName} {preferedLastName}"
                                #fullName = f"{preferedLastName}, {preferedFirstName}"
                            else:
                                fullName = f"{personMatch.first_name} {personMatch.last_name}"
                                #fullName = f"{personMatch.last_name}, {personMatch.first_name}"
                            if personMatch.is_attorney == 'YES':
                                #outputFile.write(f"{docID} has match {personMatch.first_name} {personMatch.last_name}* ({personMatch.work_email_address.split('@')[-1]})\n")
                                matchedMetadataValues.add(f"{fullName}* ({personMatch.work_email_address.split('@')[-1]})")
                            else:
                                #outputFile.write(f"{docID} has match {personMatch.first_name} {personMatch.last_name} ({personMatch.work_email_address.split('@')[-1]})\n")
                                matchedMetadataValues.add(f"{fullName} ({personMatch.work_email_address.split('@')[-1]})")
                else:
                    outputFile.write(f"{docID} contains a non-email item {nameItem}\n\n")
            missingFromFormatted = matchedMetadataValues - formattedAttorneyValues
            missingFromMeta = formattedAttorneyValues - matchedMetadataValues
            if missingFromFormatted:
                for missingItem in missingFromFormatted:
                    outputFile.write(f"{docID} has {missingItem} missing from the formatted field\n")
            if missingFromMeta:
                for missingItem in missingFromMeta:
                    outputFile.write(f"{docID} has {missingItem} missing from the metadata field\n")
            if missingFromFormatted:
                outputFile.write("\n")
            elif missingFromMeta:
                outputFile.write("\n")
    outputFile.close()


##    people_list = PeopleList()
##    people_list.add_person(Person(firstName = "Sally", lastName = "Smith", workEmailAddress = "fooBar@gmail.com", altWorkEmailAddress = ""))
##    people_list.add_person(Person(firstName = "Gary", lastName = "Cooper", workEmailAddress = "", altWorkEmailAddress = "spam.eggs@hotmail.com"))
##    people_list.add_person(Person(firstName = "", lastName = "", workEmailAddress = "noname@gmail.com", altWorkEmailAddress = ""))
##    people_list.add_person(Person(firstName = "Sally", lastName = "Smith", workEmailAddress = "eggs@outlook.com", altWorkEmailAddress = ""))
##    print("\nAll People:")
##    people_list.list_people()
##    print("\nSearching...")
##    result = people_list.search_by_email('fooBar@gmail.com')
##    print(result if result else "email not found.")
Revision:	846
Committed:	Tue Dec 10 17:22:55 2024 UTC (15 months, 2 weeks ago) by nino.borges
Content type:	text/x-python
File size:	23675 byte(s)
Log Message:	They changed a column name to 'comments', so updated dataclass. also added some warning messages when the consilio override ingestion cant find a matching email address in the MAL. I'm seeing many of these.