[PYTHON] Automatically access the flow in enebular and pull the trigger

My name is Niamugi and I am in charge of the second day of enebular Advent Calendar 2019. This time I would like to show you how to automatically access http://enebular.com/app and trigger the flow.

manner

Scraping

Trigger

The Node-RED flow created by enebular is finally deployed to some device or service. Even though there is a flow on enebular, it's a waste not to use it ... So, if the flow can be opened and executed automatically, can it be used? I thought, I tried it.

Try out

Write a Node-RED flow for testing

If you click the button of the inject node "start", it is a flow that just executes Cloud Functions of GCP with http request. enebular_node_red.PNG

Scraping http://enebular.com/app

We will investigate step by step from the place where the web page is opened. The program is written in python, which is easy to scrape.

log in

Open http://enebular.com/app and wait until you can enter your email address.
WebDriverWait(driver, 5).until(ec.element_to_be_clickable((By.NAME, "email")))

Once open, enter your email address and password and press Enter.

id = driver.find_element_by_name("email")
id.send_keys(un)
password = driver.find_element_by_name("password")
password.send_keys(pw)
password.send_keys(Keys.ENTER)

Select an asset

Wait for the asset you want to select to appear. This time it is "test scraping".
assetNm = "testscraping"
assetPath = '//span[@data-testid="' + assetNm +'"]'
WebDriverWait(driver, 5).until(ec.element_to_be_clickable((By.XPATH, assetPath)))

Click on the asset when it appears.

driver.find_element_by_xpath(assetPath).click()

Press the "edit" button in the Overview

Wait for the asset overview to appear. It is judged that the time when the Edit button on the right side can be clicked is displayed.
editBtn = '//button[@data-testid="btn-edit-flow"]'
WebDriverWait(driver, 5).until(ec.element_to_be_clickable((By.XPATH, editBtn)))

Click the Edit button when displayed.

driver.find_element_by_xpath(editBtn).click()

Switch tabs

Click the Edit button to display the flow in a separate tab. Therefore, you need to switch tabs.

handle_array = driver.window_handles
driver.switch_to.window(handle_array[1])

Wait for the flow to appear

** This is the biggest challenge. ** ** First, "Loading ..." is displayed.

Next, the frame for editing the flow is displayed. ** At this time, the flow tab is not displayed yet. ** ** After a while, the flow tabs (dummy and input this time) will be displayed.

Since the frame display and tab display timing are different, some ingenuity is required. The problem was solved by repeating loading of frames until the flow was displayed correctly (= tabs were displayed). (It's not good to wait with "while True" ... you should have a time limit ...)

flowNm = 'input'
frame = '//div[@id="iframeBlock"]/iframe'
flowPath = '//a[@title="' + flowNm + '"]'

# Loading...Wait for the end
WebDriverWait(driver, 30).until(ec.frame_to_be_available_and_switch_to_it((By.XPATH, frame)))

#Wait until the flow is fully displayed
while True:
    #Undo the frame
    driver.switch_to.default_content()
    #Switch frames
    iframe = driver.find_element_by_xpath(frame)
    driver.switch_to_frame(iframe)        

    try:
        #Exception if the flow is not displayed correctly
        WebDriverWait(driver, 1).until(ec.visibility_of_element_located((By.XPATH, flowPath)))
        break
    except:
        pass

Click the tab when it appears.

driver.find_element_by_xpath(flowPath).click()

Press the trigger

This time, click the button of the inject node "start" to execute the flow. It's a hassle to get to the node.
triggerNodeNm = 'start'
try:
    #Move to canvas
    workspace = driver.find_element_by_id('workspace')
    chart = workspace.find_element_by_id('chart')
    canvas = chart.find_element_by_class_name('innerCanvas')
    
    # class = "node nodegroup"Find
    nodes = canvas.find_elements_by_css_selector('.node.nodegroup')
    #Check if the node name matches the one you specified
    #If they match, press the trigger
    for node in nodes:
        nodeNm = node.find_element_by_tag_name('text').text
        if nodeNm == triggerNodeNm:
            node.find_element_by_class_name('node_button_button').click()
            break
        
except Exception as e:
    driver.quit()

After pressing the trigger, the screen is closed and the process ends.

driver.quit()

Result of trying

When you run the program ... The CloudFunctions function has been executed!

Summary

By scraping, I was able to utilize the flow of enebular. This method may be sufficient if the flow does not depend on the system or interface to which it is deployed. (There is a problem that it does not work if the tag or class name of the web page changes ...)

I hope it will give you some hints. See you soon.

Recommended Posts

Automatically access the flow in enebular and pull the trigger
Access the Twitter API in Python
Automatically update and confirm the school homepage
Find it in the procession and edit it
Access files in the same directory as the executable
12. Save the first column in col1.txt and the second column in col2.txt
Automatically create word and excel reports in python
About the difference between "==" and "is" in python
When the axis and label overlap in matplotlib
Macports easy_install automatically resolves and runs the version
Access the variables defined in the script from the REPL