Rainbow Engine

IT技術を分かりやすく簡潔にまとめることによる学習の効率化、また日常の気付きを記録に残すことを目指します。

Python Slack

Slack AppのWorkflow Stepsの使い方

投稿日:2023年5月12日 更新日:

 

<目次>

(1) Slack AppのWorkflow Stepsの使い方
 (1-0) やりたいこと
 (1-1) STEP1:Slackボットの開発(所要時間:60分)
 (1-2) STEP2:Slack AppにWorkflow Stepsの設定を追加(所要時間:10分)
 (1-3) STEP3:Pythonプログラムに「/slack/events」が呼ばれた時の処理を追記(所要時間:15分)

(1) Slack AppのWorkflow Stepsの使い方

(1-0) やりたいこと

・Slackのワークフロービルダーで使用できるカスタムステップの作成
・PythonプログラムでSlackのワークフローから処理を起動する
(図100①)
ここの機能を使ってみる

(図100②)

目次にもどる

(1-1) STEP1:Slackボットの開発(所要時間:60分)

↓こちらの記事内の手順(STEP1~STEP7)を全て実施し、土台となるSlackボットを用意します。
この記事では、Slackで発生するイベント(例:チャンネル投稿)を検知し、動作するSlack App(ボット)の作成手順を紹介しています。

(1-2) STEP2:Slack AppにWorkflow Stepsの設定を追加(所要時間:10分)

(1-2-1) STEP2-1:Workflow Stepsの有効化

・①メニューの「Workflow Steps」を開き「Turn on workflow steps」をONにする
(図112①)


・②「Add Step」押下
(図112②)

・③必要情報を入力して「Save」を押下
(図112③)

(備考)

「Workflow Steps」を新規登録すると(図113①)のように、残り必要な設定をWarningで示してくれます。
(図113①)

「STEP1:Slackボットの開発(所要時間:60分)」を終えられている方は「Event Subscriptions」は済んでいるため、警告は1つしか出ない(Interactivityの警告)と思います。その場合、次の「STEP2-1:Eventの登録」は実施不要です。

(1-2-2) STEP2-2:Eventの登録

・①「Event Subscriptions」のリンクを押下
・②「Enable Events」をONにする
・③「Request URL」の入力
→STEP1で作成したAzure App ServiceのURLの末尾に「/slack/events/」を付加したURLをに指定。
→(例)https://slack-app-workflow-step.azurewebsites.net/slack/events
(図113②)

※もし「Your URL didn’t respond with the value of the challenge parameter.」エラーが出たら下記の記事に沿って対処する事で解消します。
・④「Save Changes」押下
(図113③)

(1-2-3) STEP2-3:Interactivityの有効化

・①Warningの「Interactivity」のリンクを押下 or メニューの「Interactivity & Shortcuts」を押下
(図114①)

・②「Interactivity」をONにする
(図114②)
・③「Request URL」の入力
→STEP1で作成したAzure App ServiceのURLの末尾に「/slack/interactive/」を付加したURLをに指定。
→(例)https://slack-app-workflow-step.azurewebsites.net/slack/interactive
(図114③)
・④「Install App」メニュー⇒「Reinstall to Workspace」を押下
(図114④)

・⑤「許可する」
(図114⑤)

(1-2-4) STEP2-4:Appの権限設定(Permission)

ワークフローステップを実行するために、workflow.steps:executeを付与します。
・①メニューから「OAuth & Permissions」を開く
・②「Bot Token Scopes」に「workflow.steps:execute」を追加して「Save Changes」
※それ以外の権限は、やらせたい処理に応じて必要な分だけ付与します。

目次にもどる

(1-3) STEP3:Pythonプログラムに「/slack/events」が呼ばれた時の処理を追記(所要時間:15分)

・⓪(参考)ワークフロー編集画面イメージ
→サンプルプログラムはこんな感じの画面を作り、ワークフロービルダーから編集や保存を可能にします。
→また利用者がこのステップを呼び出した時の処理も記述できます。
(図122)

・①コードを追記します。
(追記コード)
import requests
from slack_bolt import App
from slack_sdk import WebClient
#Flask
from flask import Flask, request, Response, jsonify
from slack_bolt.adapter.flask import SlackRequestHandler
#ソケットモード用
from slack_bolt.adapter.socket_mode import SocketModeHandler
from slack_bolt.workflows.step import WorkflowStep

# モードに応じて書き換え
BOT_USER_ID = "xxxx"
# Botトークン(Flask)
WEBAPPS_SLACK_TOKEN = "xoxb-xxxx"
WEBAPPS_SIGNING_SECRET = "xxx"

# Botトークン(ソケットモード)
SOCK_SLACK_BOT_TOKEN = "xoxb-xxx"
SOCK_SLACK_APP_TOKEN = "xapp-1-xxx"

##############################################
# ステップの定義
##############################################
#### ポイント① ####
# slack_boltのAppインスタンス生成 ⇒イベント受付のために必要
def app_mode_change(i_name):
    if i_name == "__main__":
        # ソケットモード=ローカル実行
        return App(token=SOCK_SLACK_BOT_TOKEN)
    else:
        # Flask=WebAPサーバ実行
        return App(token=WEBAPPS_SLACK_TOKEN, signing_secret=WEBAPPS_SIGNING_SECRET)
# Boltアプリの起動
s_app = app_mode_change(__name__)
# Flaskクラスのインスタンス生成
app = Flask(__name__)
handler_flask, handler_socket = None,None

#### ポイント② ####
#Workflow StepのコールバックIDを指定して「builder」メソッドを実行
new_pleasanter_record = WorkflowStep.builder("new_pleasanter_record")

#ソケットーモードの場合のハンドラ設定
if __name__ == "__main__":
    handler_socket = SocketModeHandler(app=s_app, app_token=SOCK_SLACK_APP_TOKEN, trace_enabled=True)
#Flaskでのハンドラー設定
else:
    handler_flask = SlackRequestHandler(s_app)

#### ポイント③ ####
# Event Subscriptionで登録した「/slack/events」宛てのリクエストを受け付ける
# →イベント受付のために必要
@app.route("/slack/events", methods=["POST"])
def slack_events():
    # challenge用
    # json = request.json
    # d = {'challenge' : json["challenge"]}
    # return jsonify(d)
    return handler_flask.handle(request)

# Interactivity & Shortcutsで登録した「/slack/interactive」宛てのリクエストを受け付ける
@app.route("/slack/interactive", methods=["POST"])
def slack_interactive():
    return handler_flask.handle(request)

#### ポイント④ ####
# ワークフロービルダーから、当該ステップを追加して内容を編集する時の処理を記述するメソッド
@new_pleasanter_record.edit
def edit(ack, step, configure):
    ack()
    
    # TODO:initial_xxxは空白で無い場合に取得する&initial_valueにセットするよう、条件分岐を入れないと、空欄の時の初回がエラーになってしまう
    initial_request_depratment= step["inputs"]["task_requester_department"]["value"]
    initial_name = step["inputs"]["task_name"]["value"]
    initial_description = step["inputs"]["task_description"]["value"]
    blocks = [
        # 依頼部署
        {
            "type": "input",
            "block_id": "task_requester_department_input",
            "element": {
                "type": "plain_text_input",
                "action_id": "requester_department",
                "placeholder": {"type": "plain_text", "text":"依頼部署"},
                "initial_value": initial_request_depratment
            },
            "label": {"type": "plain_text", "text":"Requester Department"},
        },
        # 依頼者
        {
            "type": "input",
            "block_id": "task_requester_input",
            "element": {
                "type": "plain_text_input",
                "action_id": "requester",
                "placeholder": {"type": "plain_text", "text":"依頼部署"},
                #"initial_value": initial_request
            },
            "label": {"type": "plain_text", "text":"Requester Name"},
        },
        # タイトル
        {
            "type": "input",
            "block_id": "task_name_input",
            "element": {
                "type": "plain_text_input",
                "action_id": "name",
                "placeholder": {"type": "plain_text", "text":"チケットのタイトルを追加"},
                "initial_value": initial_name
            },
            "label": {"type": "plain_text", "text":"Task name"},
        },
        # 内容
        {
            "type": "input",
            "block_id": "task_description_input",
            "element": {
                "type": "plain_text_input",
                "action_id": "description",
                "placeholder": {"type": "plain_text", "text":"チケットの説明を追加"},
                "initial_value": initial_description
            },
            "label": {"type": "plain_text", "text":"Task description"},
        },
    ]
    configure(blocks=blocks)

# ワークフロービルダーから、当該ステップを追加して内容を保存する時の処理を記述するメソッド
@new_pleasanter_record.save
def save(ack, view, update):
    ack()
    # valuesでビュー全体を取り出し、それを元に各項目を取り出し  
    values = view["state"]["values"]
    task_requester_department = values["task_requester_department_input"]["requester_department"]
    task_requester = values["task_requester_input"]["requester"]
    task_name = values["task_name_input"]["name"]
    task_description = values["task_description_input"]["description"]

    # ユーザーが入力したテキストボックスの値を取り出し
    inputs = {
        "task_requester_department": {"value": task_requester_department["value"]},
        "task_requester": {"value": task_requester["value"]},
        "task_name": {"value": task_name["value"]},
        "task_description": {"value": task_description["value"]}
    }
    # 出力を後続のステップに渡す
    outputs = [
        # 依頼部署
        {
            "type": "text",
            "name": "task_requester_department",
            "label":"Requester Department",
        },
        # 依頼者
        {
            "type": "text",
            "name": "task_requester",
            "label":"Requester Name",
        },
        # タイトル
        {
            "type": "text",
            "name": "task_name",
            "label":"Task name",
        },
        # 内容
        {
            "type": "text",
            "name": "task_description",
            "label":"Task description",
        }
    ]
    print("========SAVE INPUTS======="+str(inputs))
    print("========SAVE OUTPUTS======="+str(outputs))
    update(inputs=inputs, outputs=outputs)

# エンドユーザーが、当該ステップを実行する時に実行する処理を記述するメソッド
@new_pleasanter_record.execute
def execute(step, complete, fail):
    # 実行時にユーザーが入力した値を取り出し
    inputs = step["inputs"]

    # すべての処理が成功した場合
    outputs = {
        "task_requester_department": inputs["task_requester_department"]["value"],
        "task_requester": inputs["task_requester"]["value"],
        "task_name": inputs["task_name"]["value"],
        "task_description": inputs["task_description"]["value"],
    }

    print("========EXECUTE INPUTS======="+str(inputs))
    print("========EXECUTE OUTPUTS======="+str(outputs))

    try:
        ############## カスタム処理START(ご自身で実行したい処理をここに記述) ##############
        # (例) Pleasanterに明細を起票
        # →PleasanterのAPIをPythonからコールするサンプルをご紹介(★URL)
        ############## カスタム処理END(ご自身で実行したい処理をここに記述) ##############
        # ワークフローステップの完了
        complete(outputs=outputs)
    except:
        # 失敗した処理がある場合
        error = {"message":"Just testing step failure!"}
        # ワークフローステップの失敗
        fail(error=error)

# WorkflowStep の新しいインスタンスを作成する
ws = WorkflowStep(
    callback_id="new_pleasanter_record",
    edit=edit,
    save=save,
    execute=execute,
)
# ワークフローステップを渡してリスナーを設定する
s_app.step(ws)

# __name__はPythonにおいて特別な意味を持つ変数です。
# 具体的にはスクリプトの名前を値として保持します。
# この記述により、Flaskがmainモジュールとして実行された時のみ起動する事を保証します。
# (それ以外の、例えば他モジュールから呼ばれた時などは起動しない)
if __name__ == '__main__':
    EXEC_MODE = "FLASK_WEB_API"
    # Slack ソケットモード実行
    if EXEC_MODE == "SLACK_SOCKET_MODE":
        handler_socket.start()
    # Flask Web/APサーバ 実行
    elif EXEC_MODE == "FLASK_WEB_API":
        # Flaskアプリの起動
        # →Webサーバが起動して、所定のURLからアクセス可能になります。
        # →hostはFlaskが起動するサーバを指定しています(今回はローカル端末)
        # →portは起動するポートを指定しています(デフォルト5000)
        app.run(port=8000, debug=True)
・②requirement.txtも作成
(例)
requests
Flask
slack_bolt
slack_sdk
slack
redis
azure-functions
pyodbc
pandas
(図115)


・③Azure App Serviceへ再度リリースします。
(図122)

これでワークフロービルダー上からこのステップが利用可能になりました。また、ワークフローの利用者から呼ばれた時には「def execute(step, complete, fail):」に記述した処理が実行されます。
(参考)

目次にもどる

Adsense審査用広告コード


Adsense審査用広告コード


-Python, Slack
-

執筆者:


comment

メールアドレスが公開されることはありません。 が付いている欄は必須項目です

関連記事

PythonのPandas使用時に発生した「UnicodeEncodeError: ‘ascii’ codec can’t encode characters~」エラーの対処方法について

(0)目次&概説 (1) エラー対応1:UnicodeEncodeError  (1-1) 発生状況・エラーメッセージ   (1-1-1) エラーメッセージ   (1-1-2) エラーとなったソース …

Python3.5&Anaconda4.2のインストール手順(CentOS6)とHelloWorldの方法

(0)目次&概説 (1) インストール資源の入手  (1-1) ディストリビューション  (1-2) GitHubのpyenvのリポジトリのコピーを取得する (2) インストールの実施  (2-1) …

Pythonでcsvのカラム名・テーブルのレコードを取得する方法

今回PythonのPandasライブラリを用いてcsvファイルを読み込み、カラム名やテーブルのレコードを取得する方法をご紹介します。 <目次> (1) Pythonでcsvのカラム名・テーブルのレコー …

PythonにOpenCVを導入して画像を読込み&表示&拡大縮小してみる

(0)目次&概説 (1) OpenCVとは (2) OpenCVのインストール  (2-1) インストール資源の入手  (2-2) インストール時の注意事項  (2-3) インストールの実行 (3) …

CondaHTTPError: HTTP 000 CONNECTION FAILED for url XXXXエラーの対処方法

  <目次> (1) CondaHTTPError: HTTP 000 CONNECTION FAILED for url XXXXエラーの対処方法  (1-1) 事象概要とエラーメッセージ …

  • English (United States)
  • 日本語
Top