This script automatically fetch information from Paris hospitals patient information system by simulating GUI actions.

APHPscraping.py

"""This script automatically fetch information from Paris hospitals patient information system by simulating GUI actions.

See it in action there : https://www.youtube.com/watch?v=nGMSP_wBr-8 .

Although this is ad-hoc to the task at hand, we share this because some features are nice and worth replicating :
    - Logging allow for after-the-fact debugging. This is necessary because the script is to be launched by people who don't 
        have IT skills, and it automates a buggy software which randomly crashes.
    - There are some workarounds for pyautogui's limitations, see nonasciichars_workaround() and change_date().
    - Using csv files as input and output and rewriting them at every iteration allow for no data loss, no matter how critical the failure.
    - Trapping errors in try:...except: blocks allow to log failure states and correct them on the go, while continuing the tasks
        on entries that happen to work.
        
This is released under the WTF public licence.
"""
__author__ = 'Edouard Klein <edou -at- rdklein.fr>'

from pyautogui import locateOnScreen, center, click, hotkey, typewrite, press, moveTo, moveRel, alert, dragRel, position, onScreen
import pyautogui
pyautogui.PAUSE = .1
from time import sleep
import pyperclip
import datetime
import csv
import sys
import traceback

import logging
logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
                    datefmt='%m-%d %H:%M',
                    filename='./debuglog.txt',
                    filemode='w')
console = logging.StreamHandler()
console.setLevel(logging.INFO)
logging.getLogger('').addHandler(console)



def nonasciichars_workaround(text):
    """Workaround for pyautogui's inability to handle non american keyboard layouts'

    We copy the chars into the clipboard and then press control-v.
    """
    pyperclip.copy(text)
    hotkey('ctrl', 'v')


def info_sleep(seconds):
    """I'm sleeping and I log it"""
    logging.debug("Waiting for the computer to catch up ("+str(seconds)+" seconds)...")
    sleep(seconds)
    logging.debug("\tdone")


def click_on_image(filename, clicks=1, max_tries=10):
    """Click on the first occurrence on the image at filename on screen
    
    To avoid conservatively waiting before calling this in order to let slow Cristal Net refresh the screen,
    we provide an active waiting mechanism where we keep looking for the image until it is found.
    """
    logging.debug("Clicking on image "+filename+" (max_tries : "+str(max_tries)+") ...")
    nb_tries = 0
    while nb_tries < max_tries:
        try:
            logging.debug("\t try number "+str(nb_tries))
            click(center(locateOnScreen(filename, grayscale=True)), clicks=clicks)
        except TypeError:  # This is what is raised when locateOnScreen returns None
            nb_tries+=1
            pass
        else:
            logging.debug("\tdone.")
            return
    raise Exception("Image "+filename+" could not be found")


def change_date(date_text):
    """Change the date field to the given date

    The date field of crystal net does not react to typewrite() or press(), so we must use silly control actions like these
    """
    pyperclip.copy(date_text)
    pyautogui.rightClick()
    pyautogui.moveRel(20, 110)  # Now on Tout Sélectionner
    pyautogui.click()
    pyautogui.moveRel(-20, -110)  # Back on date field
    pyautogui.rightClick()
    pyautogui.moveRel(20, 90)  # Now on Supprimer
    pyautogui.click()
    pyautogui.moveRel(-20, -90)  # Back on date field
    pyautogui.rightClick()
    pyautogui.moveRel(20, 70)  # Now on Coller
    pyautogui.click()
    pyautogui.moveRel(-20, -70)  # Back on date field

last_read_date = "ERROR"


def get_str_date_at_point():
    """Get the date right of the mouse"""
    global last_read_date
    dragRel(50, 0, button='left')
    hotkey('ctrl', 'c')
    moveRel(-50, 0)
    answer = pyperclip.paste()
    logging.debug("Date : "+answer)
    last_read_date = answer
    return answer


def init_up_to_search_link():
    """Place ourselves on a page that displays the search link"""
    # Let's see the desktop
    click_on_image('img/Desktop.bmp')
    # Launch IE.
    # click_on_image('img/IE.bmp',clicks=2) # This is slow (up to more than 30s)
    click((40, 319), clicks=2)
    # Go to Cristal Net start page
    info_sleep(3)
    logging.debug('Going to CristalNet start page...')
    hotkey('ctrl', 'l')
    nonasciichars_workaround('cnet-gh02.bbs.aphp.fr/cnet_cnp1bjn')
    press('enter')
    info_sleep(.5)
    nonasciichars_workaround(r'SECRET') # Login
    press('tab')
    typewrite('VERY_SECRET')  # Password
    press('enter')
    logging.debug('\tdone.')


def search_for_patient(number, birth_date):
    """Launch a search for blood results on the time window of the pregnancy for the patient"""
    # Going on the patient search page
    logging.debug("Going on the patient search page")
    #click_on_image('img/examens_soins.bmp')
    info_sleep(1)
    #click_on_image('img/recherche_avancee.bmp')
    moveTo(100,270)  # "Recherche Avancee" link
    click()
    logging.debug('done')
    # Filling search fields
    logging.debug("Filling and sending the form...")
    click_on_image('img/ipp.bmp')
    moveRel(0, 25)
    click() # Now on IPP
    typewrite(str(number))
    moveRel(450,40) # Move to first date field
    start_date = date_from_str(birth_date) - datetime.timedelta(days=31*9)
    start_date_string = start_date.strftime("%d/%m/%Y")
    change_date(start_date_string)
    moveRel(100,0) # Now on second date field
    end_date = date_from_str(birth_date) + datetime.timedelta(days=10)
    end_date_string = end_date.strftime("%d/%m/%Y")
    change_date(end_date_string)
    press('tab')
    press('tab')
    press('tab')
    press('tab')
    press('tab')
    press('tab') # Now on UM exécutante
    typewrite('h') # Puts us on Hématologie (LMR)
    press('enter') # Launch search
    click_on_image('img/result_list.bmp')  #Waiting for result page to appear
    logging.debug("\tdone")


we_scrolled = False


def move_to_previous_record():
    """Move pointer to previous record"""
    pos = list(position())
    pos[1] -= 50
    if pos[1] < 190:
        raise Exception("No previous record")
    moveRel(0, -50)
    # FIXME: Refactor these two function as one, that scroll up and down as necessary
    # FIXME: Maybe do an iterable of records
    # FIXME: This is probably not needed in practice, though.


def move_to_next_record():
    """Move pointer to next date, scroll once if necessary"""
    global we_scrolled
    pos = list(position())
    pos[1] += 50
    if onScreen(pos):
        moveRel(0,50)
    else:
        if not we_scrolled:
            moveTo(575, 190)
            click()
            press('down');press('down');press('down');press('down');press('down');press('down');press('down')
            we_scrolled = True
        else:
            raise Exception("Date not Found !")


def find_record_at(date):
    """Find the record at, or immediately after the provided date"""
    logging.debug("Looking for record at date "+date+"...")
    moveTo(575, 190)  # Left of first date
    while date_from_str(get_str_date_at_point()) < date_from_str(date):
        move_to_next_record()


def find_hemoglobine():
    # Now let's click on the "+" button to have the test results
    moveRel(130,0)
    click()
    click_on_image('img/OK.bmp')  # Click on the OK button of the useless error message that pops up
    info_sleep(1)  # sleeping 1 sec is probably shorter than searching for image uselessly
    try:
        click_on_image('img/results_header.bmp')  # Wait for the test result page to be displayed
    except Exception:
        close_results_page()
        raise Exception("Result page did not properly open or was not recognized")
    # Check if the results include hemoglobine
    try:
        click_on_image('img/hemoglobine.bmp', max_tries=1)
        moveRel(185, 0)
        dragRel(30, 0, button='left')
        hotkey('ctrl', 'c')
        logging.debug('Hemoglobine : '+pyperclip.paste())
        answer = pyperclip.paste()
    except Exception:
        logging.debug('PAS DINFOS SUR CETTE PAGE')
        answer = None
    finally:
        close_results_page()
    return answer


def close_results_page():
    """Close the page"""
    try:
        moveTo(center(locateOnScreen('img/three_buttons.bmp', grayscale=True)))
    except TypeError:
        raise Exception("CRITICAL: Was unable to close result window !!!")
    moveRel(23, 0)
    click()


def date_from_str(str_date):
    """Return a datetime object from a string DD/MM/YYYY"""
    day, month, year = str_date.split('/')
    return datetime.datetime(int(year), int(month), int(day))


def hemoglobine_and_dates(number, birth_date):
    """Return [date_before, hemoglobine_before, date_after, hemoglobine_after]"""
    search_for_patient(number, birth_date)
    find_record_at(birth_date)
    move_to_previous_record()  # Now at birth date let's go back one notch
    get_str_date_at_point()  # As a side effect, it will store the date in last_read_date
    pos = position()
    hem = find_hemoglobine()
    while not hem:
        moveTo(pos)
        move_to_previous_record()
        pos = position()
        get_str_date_at_point()  # Side effect of setting last_read_date
        hem = find_hemoglobine()
    answer = [last_read_date, hem]
    find_record_at(birth_date)  # Now let's find the record after
    while not date_from_str(get_str_date_at_point()) > date_from_str(birth_date):
        move_to_next_record()
    pos = position()
    hem = find_hemoglobine()
    while not hem:
        moveTo(pos)
        move_to_next_record()
        pos = position()
        get_str_date_at_point()  # Side effect of setting last_read_date
        hem = find_hemoglobine()
    answer += [last_read_date, hem]
    return answer


if __name__ == "__main__":
    output_rows = {}
    error_rows = {}
    with open('20141218_error.csv', 'r') as error_file:
        reader = csv.DictReader(error_file, delimiter=',')
        for row in reader:
            error_rows["#"+row["number"]+" "+row["date"]] = row
    with open('20141218_output.csv', 'r') as output_file:
        reader = csv.DictReader(output_file, delimiter=',')
        for row in reader:
            output_rows["#"+row["number"]+" "+row["date"]] = row
    #init_up_to_search_link() #We'll do it by hand, it's more reliable.
    info_sleep(3) #Leave me time to change app after I launch the program
    with open('20141218_entree_programme.csv', 'r') as input_file:
        reader = csv.DictReader(input_file, delimiter=';')
        for row in reader:
            row_key = "#"+row["number"]+" "+row["date"]
            try:
                output_rows[row_key]
                logging.info(row_key+" has already been successfully processed. Not doing it again !")
                continue
            except KeyError:
                try:
                    error_rows[row_key]
                    logging.info(row_key+" produced an error last time we tried. Not trying again !")
                    continue
                except KeyError:
                    pass
            logging.info("Processing "+row_key+"...")
            try:
                ans = hemoglobine_and_dates(int(row['number']), row['date'])
                row['date_before'] = ans[0]
                row['hemo_before'] = ans[1]
                row['date_after'] = ans[2]
                row['hemo_after'] = ans[3]
                output_rows[row_key] = row
                with open('20141218_output.csv', 'w') as output_file:
                    writer = csv.DictWriter(output_file, ['number', 'date', 'date_before', 'hemo_before', 'date_after', 'hemo_after'])
                    writer.writeheader()
                    for x in output_rows.values():
                        writer.writerow(x)
                logging.info("... "+str(ans)+" -> done :)")
            except Exception as err:
                for frame in traceback.extract_tb(sys.exc_info()[2]):
                    fname,lineno,fn,text = frame
                row['error_message'] = "Line {}: {}".format(lineno,str(err))
                error_rows[row_key] = row
                with open('20141218_error.csv', 'w') as error_file:
                    writer = csv.DictWriter(error_file, ['number', 'date', 'error_message'])
                    writer.writeheader()
                    for x in error_rows.values():
                        writer.writerow(x)
                logging.info('\t... there was an error :"'+row['error_message']+'", trying another one')