2
\$\begingroup\$

I have inherited this nightmare Python, Selenium, and Pandas project. I am learning Python as I go; and have been cleaning up and refactoring the code where I see necessary - It was worst if you can imagine; anyway, if anyone is a Python/Pandas/Selenium enthusiast and has ideas to optimize it further beyond this point; I'd love to hear.

The goal of this project, was to grab excel files in a directory or 'hot folder'; iterate through the files, sequentially; take the excel data to pandas, then to fill out a 1-page 'data entry' style web form via selenium on a website and save.

  • It would loop and fill out this 1-web form as many times as there are data in the 'hot folder'/excel files to do so.
  • Currently, I have it working with a .zip folder; and saving the form after 1 entry.
  • Lastly, eventually; I will just shoot out a "stat" log file with the date appended with the data success rate from each file..

Here is my /config.py.

FILE_LOCATION = r"C:\Zip\2019.02.12 VCCS Docs.zip" #altering this to directory produces Permission Denied
UNZIP_LOCATION = r"C:\Zip\Pending"
VITAL_URL = 'http://boringDBwebsite:8080/ContactForm'
HEADLESS = False
PROCESSORS = 4
MAPPING_DOC = ".//map/mappingDoc.xlsx"

here is my /main.py

"""Start Point"""

from data.find_pending_records import FindPendingRecords
from vital.vital_entry import VitalEntry
import sys
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.action_chains import ActionChains
import pandas as pd

if __name__ == "__main__":
    try: 
        # for PENDING_RECORDS in FindPendingRecords().get_excel_data(): begin to loop through entire directory

            PENDING_RECORDS = FindPendingRecords().get_excel_data()
            # Do operations on PENDING_RECORDS

            # Reads excel to map data from excel to vital
            MAP_DATA = FindPendingRecords().get_mapping_data()

            # Configures Driver
            VITAL_ENTRY = VitalEntry()

            # Start chrome and navigate to vital website
            # driver = VITAL_ENTRY.instantiate_chrome()
            VITAL_ENTRY.instantiate_chrome()

            # Begin processing Records
            VITAL_ENTRY.process_records(PENDING_RECORDS, MAP_DATA)

            # Save Record
            # VITAL_ENTRY.save_contact(driver)

            print (PENDING_RECORDS)

            print("All done")

    except Exception as exc:
        # print(exc)
        raise

#Start Stat Log
old_stdout = sys.stdout
log_file = open("./logs/message.log","w")
sys.stdout = log_file
# print ("Write to File")
sys.stdout = old_stdout
log_file.close()

# df = pd.read_excel(UNZIP_LOCATION, sheetname="Care Coordinaton Contact Log 2.12.2019")
# print len(df)

here is my /vital_entry.py

"""Module for completing VITAL contact logs from Advance Health"""
import re
import sys
import os
from time import sleep
from itertools import repeat
from traceback import format_exception
from datetime import datetime, timedelta
from multiprocessing import cpu_count
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.action_chains import ActionChains
from selenium.common.exceptions import NoSuchElementException
from joblib import Parallel, delayed
from vital.vital_common import (RadioWhileException, DidNotProcessAll, wait_until,
                                page_change_until, wait_ready_state)
import config

class VitalEntry:
    """Vital Entry"""

    def __init__(self):
        self.assessment_type = "Care Coordination Contact Log (Custom) - VHP"
        self.processors = config.PROCESSORS

        assert self.processors > 0, 'Need number of processes to be greater than zero.'
        assert self.processors <= cpu_count(),\
            ('Need number of processes to be at most the max number of cores/threads '
             f'available. For this computer, it\'s {cpu_count()}.')

        self.user = os.environ['USERNAME']
        self.options = webdriver.ChromeOptions()

    def instantiate_chrome(self):
        """Create Chrome webdriver instance."""
        self.options.headless = config.HEADLESS

        if not self.options.headless:
            self.options.add_argument("--start-maximized")

        self.options.add_argument('--disable-infobars')
        self.options.add_argument('--disable-gpu')

        self.driver = webdriver.Chrome(options=self.options)
        self.driver.set_page_load_timeout(30)
        self.driver.implicitly_wait(15)
        self.driver.get(config.VITAL_URL)
        # return driver

    def close_chrome(self):
        """Used to close out of chrome"""
        self.driver.quit()
        print("Closing chrome")

    def process_records(self, records, map_data, completed=None, errors=None):
        """Code to execute after webdriver initialization."""
        series_not_null = False
        try:
            num_attempt = 0

            for record in records.itertuples():
                print(record)
                series_not_null = True
                mrn = record.MRN

                self.navigate_to_search(num_attempt)
                self.navigate_to_member(mrn)
                self.navigate_to_assessment()
                self.add_assessment(record, map_data)
                # self.save_contact(self.driver)

                self.driver.switch_to.parent_frame()
                sleep(.5)

                error_flag = self.close_member_tab(self.driver, mrn, error_flag)

        except Exception as exc:
            if series_not_null:
                errors = self.process_series_error(exc)

        return completed, errors

    def navigate_to_search(self, num_attempt):
        """Uses webdriver to navigate to Search tab and select Member Status=All"""

        if num_attempt == 0:
            page_change_until(self.driver, By.XPATH, './/*[text()="Search"]')
            wait_ready_state(self.driver)
        else:
            self.driver.switch_to.parent_frame()
            elem = wait_until(self.driver, By.XPATH, './/*[text()="Search"]')
            is_disp_n = 0
            while True:
                if elem.is_displayed():
                    break
                else:
                    self.driver.switch_to.parent_frame()
                    is_disp_n += 1
                    sleep(1)
                if is_disp_n == 30:
                    raise Exception('Could not find Search tab after 30 tries.')

        num_attempt += 1
        radio_locator = (By.XPATH, './/*[@type="RADIO"][@value="All"]')

        while True:
            break_while_timer = datetime.now()

            if datetime.now() - break_while_timer > timedelta(seconds=20):
                break_while = True
                break

            try:
                if wait_until(self.driver, *radio_locator).is_selected():
                    pass
                else:
                    wait_until(self.driver, *radio_locator).click()
                break
            except Exception:
                sleep(1)

    def navigate_to_member(self, mrn):
        """Finds member"""
        wait_until(self.driver, By.XPATH, './/*[@name="MemberIdItem_1"]').clear()
        wait_until(self.driver, By.XPATH, './/*[@name="MemberIdItem_1"]').send_keys(f'{mrn}'+Keys.ENTER)

        page_change_until(self.driver, By.XPATH, f'.//*[text()="{mrn}"]')

        wait_ready_state(self.driver)

    def navigate_to_assessment(self):
        """Navigates to the appropriate contact log"""
        self.driver.find_element_by_css_selector("div[eventproxy^='memberAssessment']").click() #clicks assessment icon
        element = self.driver.find_element_by_xpath(f"//div[contains(text(), '{self.assessment_type}')]")
        actions = ActionChains(self.driver)
        actions.move_to_element(element).perform()
        self.driver.find_element_by_xpath(f"//div[contains(text(), '{self.assessment_type}')]").click()
        self.driver.find_element_by_css_selector("div[eventproxy^='createSelectedAssessmentsButton']").click()


    def add_assessment(self, record, map_data):
        """Create contact log"""
        qna_frame = self.driver.find_element_by_css_selector("iframe[id^='iccc']")
        self.driver.switch_to.frame(qna_frame)

        pages = self.driver.find_element_by_css_selector("ul[class='nav nav-pills nav-stacked qna-tabs']")
        pages = pages.find_elements_by_css_selector("a")

        for page in pages:
            page.click()

            questions = self.driver.find_elements_by_css_selector("fieldset")
            questions = [question for question in questions if question.text not in  ("", " ", "NaT", "NaN", None)]

            for question in questions[1:]:
                self.q_text = question.find_element_by_css_selector("span[class='question-text ng-binding']").text
                questionType = map_data.loc[map_data['question_text'] == self.q_text, 'question_type'].item()

                answer = map_data.loc[map_data['question_text'] == self.q_text, 'map'].item()
                answer = getattr(record, answer)

                if answer not in ("", " ", "NaT", "NaN", None):
                # while answer != "" and answer != " " and answer != "NaT":

                    if questionType == 'checks':
                        self.choose_checks(question, answer)
                    else:
                        try:
                            if questionType == 'text':
                                self.driver.implicitly_wait(0)

                                (question.find_element_by_css_selector("textarea").send_keys(str(answer))
                                if 
                                    question.find_elements_by_css_selector("textarea")
                                else 
                                    question.find_element_by_css_selector("input").send_keys(answer))

                                self.driver.implicitly_wait(15)

                            elif questionType == 'date':
                                try:
                                    answer = answer.strftime('%m/%d/%Y')
                                    question.find_element_by_css_selector("input").send_keys(answer)
                                    page.click()

                                except Exception as e:
                                    raise Errors.RequiredDataError('Issues with Assessment Date -- {}'.format(e))
                            elif questionType == 'radio':
                                question.find_element_by_css_selector("input[value='{}']".format(answer)).click()
                        except:
                            continue                              
                        else:
                            # driver.find_element_by_css_selector("#publishButton").click()
                            pass                  

    def choose_checks(self, question, answer):
        checkboxes = question.find_elements_by_css_selector('label')
        for check in checkboxes:
            if check.text == answer:
                check.click()

    def close_member_tab(self, driver, mrn, error_flag):
        """Close member tab"""
        tab = driver.find_element_by_xpath(f'.//*[contains(@aria-label,"{mrn}")]')
        while True:
            try:
                tab = driver.find_element_by_xpath(
                    f'.//*[contains(@aria-label,"{mrn}")]')
            except NoSuchElementException:
                break
            except Exception:
                sleep(1)
                continue
            else:
                try:
                    tab.find_element_by_xpath('./div/table/tbody/tr/td/table/tbody/tr/td[2]').click()
                except Exception:
                    pass
                break

        if error_flag:
            error_flag = False
            num_tries = 1
            while num_tries <= 5:
                try:
                    xpath = ('.//*[starts-with(@eventproxy, "YesNoCan celDialog")]//*[text()="No"]')
                    driver.find_element_by_xpath(xpath).click()
                    break
                except Exception:
                    sleep(1)
                    num_tries += 1
                    continue

        return error_flag

    # def find_hidden_fields (self, driver, answer):
    #     driver.implicitly_wait(10)
    #     # driver.find_element_by_css_selector("").click()
    #     driver.find_element_by_xpath("//*[@id='qna_0']/fieldset[1]/fieldset[21]/div[2]/div/textarea").click()
    #     driver.find_element_by_xpath("//*[@id='qna_0']/fieldset[1]/fieldset[21]/div[2]/div/textarea").send_keys("test")  

    # def save_contact (self, driver):
    #     driver.implicitly_wait(15)
    #     driver.find_element_by_css_selector("#publishButton").click() 

    def process_series_error(self, errors):
        #stub
        print(errors)
        return errors

here is my /find_pending_records.py

    """Module used to find records that need to be inserted into Vital"""
    from zipfile import ZipFile
    import math
    import pandas
    import config

    class FindPendingRecords:
        """Class used to find records that need to be inserted into Vital"""

        @classmethod
        def find_file(cls):
            """"Finds the excel file to process"""
            archive = ZipFile(config.FILE_LOCATION)
            for file in archive.filelist:
                if file.filename.__contains__('Care Coordinaton Contact Log '):
                    return archive.extract(file.filename, config.UNZIP_LOCATION)
            return FileNotFoundError

        # class FindPendingRecords: Begin solution for loop through entire directory
        #     @classmethod
        #     def find_file(cls):
        #         return ["file1", "file2", "file3"]

        # def __init__(self):
        #     self.files = self.find_file()

        # def get_excel_data(self):
        #     for excel_data in self.files:
        #         # process your excel_data
        #         yield excel_data

        # @classmethod
        # def find_file(cls):
        #      all_files = list()
        #      """"Finds the excel file to process"""
        #      archive = ZipFile(config.FILE_LOCATION)
        #      for file in archive.filelist:
        #         if file.filename.__contains__('Care Coordinaton Contact Log '):
        #              all_files.append(archive.extract(file.filename, config.UNZIP_LOCATION))
        #      return all_files    

        def get_excel_data(self):
            """Places excel data into pandas dataframe"""
            excel_data = pandas.read_excel(self.find_file())

            columns = pandas.DataFrame(columns=excel_data.columns.tolist())
            excel_data = pandas.concat([excel_data, columns])

            excel_data.columns = excel_data.columns.str.strip()
            excel_data.columns = excel_data.columns.str.replace("/", "_")
            excel_data.columns = excel_data.columns.str.replace(" ", "_")

            num_valid_records = 0

            for row in excel_data.itertuples():
                mrn = row.MRN

                if mrn in ("", " ", None) or math.isnan(mrn):
                    print(f"Invalid record: {row}")
                    excel_data = excel_data.drop(excel_data.index[row.Index])
                    # continue
                else:
                    num_valid_records += 1

            print(f"Processing #{num_valid_records} records")

            return self.clean_data_frame(excel_data)

        def clean_data_frame(self, data_frame):
            """Cleans up dataframes"""
            for col in data_frame.columns:
                if "date" in col.lower():
                    data_frame[col] = pandas.to_datetime(data_frame[col],
                                                         errors='coerce', infer_datetime_format=True)
                    data_frame[col] = data_frame[col].dt.date
            data_frame['MRN'] = data_frame['MRN'].astype(int).astype(str)

            return data_frame

        def get_mapping_data(self):
            map_data = pandas.read_excel(config.MAPPING_DOC, sheet_name='main')
            columns = pandas.DataFrame(columns=map_data.columns.tolist())
            return pandas.concat([map_data, columns])

Lastly, and less important; /vital_common.py

"""
Base functions and classes for contact log completion
"""
from datetime import datetime, date, timedelta
from inspect import getmembers, isclass, getclasstree
import re
from time import sleep
from contextlib import contextmanager
import pandas as pd
import numpy as np
from selenium import webdriver, common
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys 

def recurse_tree(class_tree, recurse=0):
    """Parse through subclass tree and rank by layer level."""
    out_list = []

    for sub_tree in class_tree:
        if isinstance(sub_tree, list):
            out_list.extend(recurse_tree(sub_tree, recurse=recurse+1))
        elif isinstance(sub_tree, tuple):
            out_list.append((sub_tree[1][0], recurse))

    return out_list

class RadioWhileException(Exception):
    """Exception for missing radio button in while loop."""
    def __str__(self):
        return 'Never found radio button for "all" on Search tab.'


class DidNotProcessAll(Exception):
    """Exception for not processing all records."""
    def __str__(self):
        return "Didn't process all records"


class NoCloseButton(Exception):
    """Exception when close button doesn't exist for tab"""
    def __str__(self):
        return "Close button doesn't exist on tab"


class ContactLogError(Exception):
    """Error in processing the contact log."""
    def __init__(self, add_msg):
        self.add_msg = add_msg

    def __str__(self):
        return 'Error in contact log completion: {}.'.format(self.add_msg)


def full_unravel(lol):
    """Take list-like objects with list-like entries
    and create a single list"""
    items = []
    for lobj in lol:
        if isinstance(lobj, tuple(list.__subclasses__() + [list])):
            items.extend(full_unravel(lobj))
        else:
            items.append(lobj)
    return items


def subsubclasses(cls):
    """Find subclasses and subclasses of subclasses."""
    subc = cls.__subclasses__()
    ssc = [x.__subclasses__() if hasattr(x, '__subclasses__')
           else x for x in datetime.__subclasses__()]
    return full_unravel(ssc) + subc #+ [cls]


def send_enter(self):
    """WebElement method shortcut for send_keys(Keys.ENTER)."""
    self.send_keys(Keys.ENTER)


webdriver.remote.webelement.WebElement.send_enter = send_enter


def wait_until(driver, by_strat, value, tries=None):
    """Wrapper for WebDriverWait.until method using the
    presence_of_element_located expected condition. defaults to 6 tries at 5
    seconds per try."""
    if isinstance(tries, type(None)):
        tries = 6
    count = 0
    while count < tries:
        try:
            return WebDriverWait(driver, 5).until(EC.presence_of_element_located((by_strat, value)))
        except common.exceptions.TimeoutException:
            sleep(1)
            count += 1
    raise common.exceptions.TimeoutException


def wait_until_not(driver, by_strat, value):
    """Similar to above, excpet for until_not method"""
    return WebDriverWait(driver, 20).until_not(EC.presence_of_element_located((by_strat, value)))


@contextmanager
def page_change(driver):
    """Function to find innerHTML from 'html' tag prior to performing a
    click that causes a page load."""
    old_html = driver.find_element_by_tag_name('html').get_attribute('innerHTML')
    yield

    def new_html():
        """Grab new HTML code"""
        return driver.find_element_by_tag_name('html').get_attribute('innerHTML')

    while True:
        if old_html != new_html():
            break
        else:
            sleep(1)


def page_change_until(driver, by_strat, value, tries=None):
    """Uses page_change wrapper in context and performs actual click"""
    with page_change(driver):
        wait_until(driver, by_strat, value, tries).click()


def wait_ready_state(driver):
    """Pauses code execution until page has loaded."""
    script = 'return document.readyState == "complete";'

    def func():
        """Execute script"""
        return driver.execute_script(script)
    while True:
        if func():
            break
        else:
            sleep(1)


def space_replace(text):
    """Replace spaces"""
    text = re.sub(r'\s', ' ', text)
    return re.sub(r' {2,}', ' ', text)

class NullType(type):
    """Make a Null type"""
    def __repr__(cls):
        return 'Null'

    def __str__(cls):
        return 'Null'


class Null(object, metaclass=NullType):
    """Make a null object of NullType"""
    def __repr__(self):
        return 'Null'

    def __str__(self):
        return 'Null'


def convert_to_string(obj):
    """Converts various data types to proper string format which
    will then be passed as SQL strings."""
    if pd.isnull(obj):
        return Null

    if hasattr(obj, 'tolist'):
        obj = obj.tolist()

    if isinstance(obj, float):
        if obj.is_integer():
            return str(int(obj))
        return str(obj)
    elif isinstance(obj, int):
        return str(obj)

    wrap = "'{}'"
    obj = str(obj).replace("'", "''")
    return wrap.format(obj)


def convert_to_cte(frame):
    """Converts dataframe rows to SELECT UNION SELECT format to then be used
    in a common table expression"""
    frame = frame.copy()
    frame.columns = list(map(lambda x: re.sub(r'[^\w]', '', str(x)),frame.columns))
    frame = frame.applymap(convert_to_string)
    selects = []
    for idx, series in frame.iterrows():
        select = "SELECT "
        series_items = list(series.to_dict().items())
        series_items = list(map(lambda x: '{1} as {0}'.format(*x),series_items))
        selects.append(select + ', '.join(series_items))
    return '\nUNION\n'.join(selects)


def excel_float_to_date(flt):
    """Convert Excel float values to datetime format"""
    if isinstance(flt, (datetime, date, np.datetime64)):
        return flt

    assert isinstance(flt, (float, int)), 'Value must by float or int object'

    if pd.isnull(flt):
        return pd.NaT

    days = int(np.trunc(flt))
    flt -= days
    flt *= 24*60*60
    seconds = int(np.round(flt, 0))
    base = datetime(1899, 12, 31, 0, 0)

    return base + timedelta(days, seconds)


def assess_type_map(key):
    """Dummy for type_map's use remaining_logs."""
    atm = {'CA': 'Comprehensive Assessment', 'HRA': 'HRA',
           'IBS': 'IBS (MLTSS)', 'TA': 'Triggering Assessment',
           'PA': 'Comprehensive Assessment'}

    return atm[key]


def category_icp(key):
    """Dummy for type_map's use remaining_logs."""
    cat = {'init': 'Initial', 
           'reass': 'Reassessment',
           'upd': 'Update'}

    key = re.sub(r'\s', '', key.lower())
    key_match = re.compile(r'init|reass|upd', re.IGNORECASE)
    matched = key_match.search(key)

    if matched:
        return cat[matched[0]]

    return None


def category_ict(key):
    """Dummy for type_map's use remaining_logs."""
    cat = {'ICT Summary': 'ICT Summary'}
    return cat[key]


def type_map(log, key):
    """Used for column setup in remaining_logs"""
    types = {'Assess': assess_type_map, 'ICP': category_icp, 'ICT': category_ict}
    return types[log](key)


def split_on_col_values(frame, col):
    """Splits DataFrame object on a columns unique values."""
    values = frame[col].unique()
    return {str(val): frame[frame[col] == val] for val in values}


def col_count(frame, col, **kwargs):
    """Get counts for unique items in DataFrame."""
    return frame[col].reset_index().groupby(col, **kwargs).count()

Any thoughts at all, or help; much appreciated. You may see some of my comments in the code above which were some recent/thoughts attempts of mine; such as for the 'iterating through multiple files' functionality.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.