enebular内のフローに自動でアクセスしてトリガーを引く

7708 ワード
enebular GoogleCloudPlatform Python node-red Python テキストリンク

enebular Advent Calendar 2019の2日目を担当します、ニアムギです。
今回はhttp://enebular.com/app に自動でアクセスしてフローのトリガーを引く方法を紹介したいと思います。

やり方

スクレイピングします。

きっかけ

enebularで作成したNode-REDのフローは、最終的に何かしらのデバイスやサービスにデプロイします。
ただせっかくenebular上にフローがあるのに、それを使わないのはもったいないなと…
そこで、自動でフローを開いて実行出来たら活用できるのでは？と思い、試してみました。

試してみる

テスト用のNode-REDフローを書く

injectノード"start"のボタンをクリックするとhttpリクエストでGCPのCloudFunctionsを実行するだけのフローです。

※フローは何でも良いので、自分なりに確認しやすいフローにしました。
※詳しく知りたい方はNode-RED+CloudFunctionsでPythonを動かすを参照下さい。

http://enebular.com/app をスクレイピングする

Webページを開くところから手順を踏みながら調べていきます。
プログラムはスクレイピングのしやすいpythonで書いています。

ログインする


http://enebular.com/app を開いて、メールアドレスが入力できるまで待ちます。

WebDriverWait(driver, 5).until(ec.element_to_be_clickable((By.NAME, "email")))

開いたらメールアドレスとパスワードを入力して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)

アセットを選択する


選択するアセットが表示されるまで待ちます。今回は"testscraping"です。

assetNm = "testscraping"
assetPath = '//span[@data-testid="' + assetNm +'"]'
WebDriverWait(driver, 5).until(ec.element_to_be_clickable((By.XPATH, assetPath)))

表示されたらアセットをクリックします。

driver.find_element_by_xpath(assetPath).click()

Overviewにある"edit"ボタンを押す


アセットのOverviewが表示されるまで待ちます。
右側にあるEditボタンをクリックできる時が表示された時と判断します。

editBtn = '//button[@data-testid="btn-edit-flow"]'
WebDriverWait(driver, 5).until(ec.element_to_be_clickable((By.XPATH, editBtn)))

表示されたらEditボタンをクリックします。

driver.find_element_by_xpath(editBtn).click()

タブを切り替える

Editボタンをクリックすると別タブにフローが表示されます。そのためタブを切り替える必要があります。

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

フローが表示されるのを待つ

ここが一番の難関です。

まず"Loading..."が表示されます。


次にフローを編集するフレームなどが表示されます。 この時はまだフローのタブが表示されていません。


しばらく経ってからフローのタブ(今回はdummyとinput)が表示されます。

フレームの表示とタブの表示タイミングが異なるため工夫が必要です。
フローが正しく表示される(＝タブが表示される)までフレームの読み込みを繰り返すことで解決しました。("while True"で待つのはよろしくないですね…時間制限を持たせるべきですね…)

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

# Loading... が終わるのを待つ
WebDriverWait(driver, 30).until(ec.frame_to_be_available_and_switch_to_it((By.XPATH, frame)))

# フローが完全に表示されるまで待つ
while True:
    # フレームを元に戻す
    driver.switch_to.default_content()
    # フレームを切り替える
    iframe = driver.find_element_by_xpath(frame)
    driver.switch_to_frame(iframe)        

    try:
        # フローが正しく表示されていない場合はExceptionになる
        WebDriverWait(driver, 1).until(ec.visibility_of_element_located((By.XPATH, flowPath)))
        break
    except:
        pass

表示されたらタブをクリックします。

driver.find_element_by_xpath(flowPath).click()

トリガーを押す


今回はinjectノード"start"のボタンをクリックさせてフローを実行させます。
ノードにたどり着くまでがなかなか面倒です。

triggerNodeNm = 'start'
try:
    # 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"を探す
    nodes = canvas.find_elements_by_css_selector('.node.nodegroup')
    # ノード名が指定したものと一致しているかチェックする
    # 一致している場合、トリガーを押す
    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()

トリガーを押せたのであとは画面を閉じて終了です。

driver.quit()

試した結果

プログラムを実行すると・・・

CloudFunctionsの関数が実行されました！

まとめ

スクレイピングすることでenebularのフローを活用できました。
デプロイ先のシステムやインターフェースに依存しないフローであれば、この方法でも良いかもしれません。
(Webページのタグやクラス名などが変わると動かなくなる問題はあります…)

何かのヒントになれば幸いです。ではでは。