Slack Approval Notification

Introduction

This extension allows you to send a notification to a Slack DM when a invoice entered the workflow for approval. The notification in Slack contains a button to go directly to the approval task in Rossum.

Prerequisites

1. Access to the Slack Workspace

2. Create an Slack App on https://api.slack.com/apps/ (ask your workspace administrator for access).

3. Get the Bot User OAuth Token for API access to your workspace

Extension settings

1. Create an extension

2. Triggered event: Document status: changed

3. Queues: select your queues

4. Additional metadata: none

5. Runtime: python3.12

6. Select token owner

Configuration JSON

Configuration contains the email address of the slack user to sent the DM to. This is for demo purposes only, so you receive the notifications.

settings:

{
  "slack_user_email": "example@rossum.ai"
}

secrets:

{
  "slack_token": "copy_slack_token_here"
}

Python code

Copy following code in

import requests
import json
from typing import List, Optional


def rossum_hook_request_handler(payload: dict) -> dict:
    
    messages = []
    operations = []

    #general settings   
    slackuser = payload["settings"]["slack_user_email"]
    slacktoken = payload["secrets"]["slack_token"]
    base_url = payload["base_url"]
    rossum_token = payload["rossum_authorization_token"]
    
    #annotation data    
    status = payload["annotation"]["status"]
    annotation_id = payload["annotation"]["id"]
    annotation_content_url = payload["annotation"]["content"]

    # get annoation data
    annotation_content = get_annotation_content(annotation_content_url, rossum_token)
  
    # check if status is to_review, only then sent slack notification
    if status == "in_workflow":
        user = find_slackuser_by_email(slackuser, slacktoken)
        send_slack_message(user, slacktoken, annotation_id, base_url, annotation_content["content"])
        
    
    messages = [create_message("info", "slack notification sent")]
    return {"messages": messages, "operations": operations}


def find_slackuser_by_email(slackuser, slacktoken):
    
    url = "https://slack.com/api/users.lookupByEmail?email=" + slackuser
    header = {"Authorization": "Bearer " + slacktoken}
    response = requests.get(url, headers=header)

    return response.json()["user"]
    
    

def send_slack_message(user, slacktoken, annotation_id, base_url, annotation_content):
    
    userid = user["id"]
    realname = user["real_name"]
    
    # Extract invoice information
    invoicenumber = find_by_schema_id(annotation_content, "document_id")["content"]["value"]
    vendorname = find_by_schema_id(annotation_content, "sender_name")["content"]["value"]
    duedate = find_by_schema_id(annotation_content, "date_due")["content"]["value"]
    amount_total = find_by_schema_id(annotation_content, "amount_total")["content"]["value"]
    currency = find_by_schema_id(annotation_content, "currency")["content"]["value"].upper()

    # Get document URL for direct access
    document_url = base_url + "/requests/" + str(annotation_id)
  
    # Prepare Slack message
    blocks =  [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": "📋 Invoice Approval Required"
                }
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Assigned to:*\n{realname}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": "*Status:*\nPending Approval"
                    }
                ]
            },
            {
                "type": "divider"
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": "*Invoice Details:*"
                }
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Vendor:*\n{vendorname}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Invoice Number:*\n{invoicenumber}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Amount:*\n{amount_total} {currency}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Due Date:*\n{duedate}"
                    }
                ]
            },
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {
                            "type": "plain_text",
                            "text": "Review Invoice"
                        },
                        "style": "primary",
                        "url": document_url
                    }
                ]
            }
        ]
    

    url = "https://slack.com/api/chat.postMessage"
    header = {"Authorization": "Bearer " + slacktoken}
    formdata = {"channel": userid, "text":"An invoice requires your approval", "blocks": json.dumps(blocks)}
    
    response = requests.post(url, headers=header, data=formdata)
    return


    
def create_message(message_type, message_content, datapoint_id=None):
    return {
        "content": message_content,
        "type": message_type,
        "id": datapoint_id,
    }


def get_annotation_content(annotationurl: str, rossum_token: str) -> dict:
    url = annotationurl
    header = {"Authorization": "Bearer " + rossum_token}
    response = requests.get(url, headers=header)

    return response.json()
    
    

def find_by_schema_id(content, schema_id: str) -> Optional[dict]:
    """
    Return the first datapoint matching a schema id or None if no such element exists.
    :param content: annotation content tree (see https://api.elis.rossum.ai/docs/#annotation-data)
    :param schema_id: field's ID as defined in the extraction schema(see https://api.elis.rossum.ai/docs/#document-schema)
    :return: the datapoint matching the schema ID
    """
    for node in content:
        if node["schema_id"] == schema_id:
            return node
        elif "children" in node:
            # if the element with specified schema_id was found
            if found := find_by_schema_id(node["children"], schema_id):
                return found
    return None