expense-approval-bot/src/processor/app.py

import json
import os
import re
import urllib.parse
import urllib.request

import boto3

secrets_client = boto3.client("secretsmanager")
_cached_token: str | None = None

SLACK_BOT_TOKEN_SECRET_ARN = os.environ["SLACK_BOT_TOKEN_SECRET_ARN"]

STAGES = {
    "C0AQ2AWLNEN": {"next": "C0APLSGABAB", "label": "Processed"},
    "C0APLSGABAB": {"next": "C0AQ09CDJH4", "label": "Authorized"},
    "C0AQ09CDJH4": {"next": "C0APYUM1JFP", "label": "Matched"},
}

REACT_HINT_RE = re.compile(r"_React_ :white_check_mark: _to advance to \w+_")


def handler(event, context):
    if event.get("reaction") != "white_check_mark":
        print("Not white_check_mark reaction, skipping")
        return

    from_channel = event["item"]["channel"]
    message_ts = event["item"]["ts"]

    route = STAGES.get(from_channel)
    if not route:
        print(f"Channel {from_channel} not in STAGES, skipping")
        return

    to_channel = route["next"]
    label = route["label"]
    token = get_bot_token()

    history = slack_get("conversations.history", token, {
        "channel": from_channel,
        "latest": message_ts,
        "limit": 1,
        "inclusive": "true",
    })
    original_text = history["messages"][0]["text"]

    permalink_resp = slack_get("chat.getPermalink", token, {
        "channel": from_channel,
        "message_ts": message_ts,
    })
    permalink = permalink_resp["permalink"]

    text = REACT_HINT_RE.sub("", original_text).strip()
    next_stage = STAGES.get(to_channel)
    next_label = next_stage["label"] if next_stage else None
    react_line = f"\n\n_React_ :white_check_mark: _to advance to {next_label}_" if next_label else ""
    permalink_line = (
        f"\n\n📎 *Original Submission:* <{permalink}|View Original Message>"
        if to_channel == "C0APLSGABAB"
        else ""
    )
    full_text = f"{text}{permalink_line}{react_line}"

    slack_post("chat.postMessage", token, {
        "channel": to_channel,
        "text": full_text,
        "mrkdwn": True,
        "unfurl_links": False,
        "unfurl_media": False,
    })

    slack_post("chat.postMessage", token, {
        "channel": from_channel,
        "thread_ts": message_ts,
        "text": f"➡️ Advanced to {label}",
    })


def get_bot_token() -> str:
    global _cached_token
    if _cached_token:
        return _cached_token
    result = secrets_client.get_secret_value(SecretId=SLACK_BOT_TOKEN_SECRET_ARN)
    _cached_token = result["SecretString"]
    return _cached_token


def slack_get(method: str, token: str, params: dict) -> dict:
    url = f"https://slack.com/api/{method}?{urllib.parse.urlencode(params)}"
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
    with urllib.request.urlopen(req, timeout=10) as resp:
        data = json.loads(resp.read())
    if not data.get("ok"):
        raise RuntimeError(f"{method} failed: {data.get('error')}")
    return data


def slack_post(method: str, token: str, body: dict) -> dict:
    req = urllib.request.Request(
        f"https://slack.com/api/{method}",
        data=json.dumps(body).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json; charset=utf-8",
        },
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        data = json.loads(resp.read())
    if not data.get("ok"):
        raise RuntimeError(f"{method} failed: {data.get('error')}")
    return data