LINE <-> Slack

main.py

import html
import logging
import os
import re

from slack_bolt import App, Ack
from linebot import WebhookHandler, LineBotApi
from linebot.exceptions import InvalidSignatureError
from linebot.models import (
    MessageEvent,
    TextMessage,
    StickerMessage,
    ImageMessage,
    VideoMessage,
    TextSendMessage,
)
from slack_sdk import WebClient

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s.%(msecs)03d %(levelname)s %(pathname)s (%(lineno)s): %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)

slack_channel_id = os.environ["SLACK_CHANNEL_ID"]
slack_channel_ids_to_relay = os.environ.get("SLACK_FORWARD_CHANNEL_IDS").split(",")
slack_app = App(
    token=os.environ["SLACK_BOT_TOKEN"],
    signing_secret=os.environ["SLACK_SIGNING_SECRET"],
)
slack_team_id = slack_app.client.auth_test()["team_id"]
slack_app_token = os.environ.get("SLACK_APP_TOKEN")

slack_api = slack_app.client
line_bot_api = LineBotApi(os.environ.get("LINE_CHANNEL_ACCESS_TOKEN"))
line_push_message_to_id = os.environ.get("LINE_PUSH_MESSAGE_TO_ID")


# ---------------------------------
# Slack bot app
# ---------------------------------

from slack_bolt.adapter.flask import SlackRequestHandler
from slack_bolt.adapter.socket_mode import SocketModeHandler

slack_flask_handler = SlackRequestHandler(slack_app)

slack_app.action("open-line-app")(lambda ack: ack())


@slack_app.event("app_home_opened")
def update_home_tab(event: dict, client: WebClient):
    client.views_publish(
        user_id=event["user"],
        view={
            "type": "home",
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": ":line: と :slack: 間での転送を行うボットです。今後、機能が増えるかも？",
                    },
                    "accessory": {
                        "type": "button",
                        "url": "https://line.me/R/nv/chat",
                        "text": {"type": "plain_text", "text": ":line: を開く"},
                        "value": "button",
                        "action_id": "open-line-app",
                    },
                }
            ],
        },
    )


# TODO: メッセージショートカットで Slack メッセージを LINE に転送
@slack_app.shortcut("send-to-line")
def send_to_line(ack: Ack, logger: logging.Logger, shortcut: dict):
    logger.info(shortcut["message"])
    ack()


def slack_channel_deep_link(channel_id: str) -> str:
    return f"slack://channel?team={slack_team_id}&id={channel_id or slack_channel_id}"


def is_target_channel(event):
    channel_id = event.get("channel")
    return (channel_id == slack_channel_id or channel_id in slack_channel_ids_to_relay) and event.get("text") is not None


def get_user_name(client: WebClient, event: dict):
    name = client.users_info(user=event.get("user")).get("user", {}).get("profile", {}).get("display_name", "名無し")
    if len(name.strip()) == 0:
        # bot or workflow builder
        return "bot"
    return name


@slack_app.event(
    event={"type": "message", "subtype": None},
    matchers=[is_target_channel],
)
def send_to_line_from_channel(event: dict, client: WebClient):
    text, user_name, channel_id = (
        event.get("text"),
        get_user_name(client, event),
        event.get("channel"),
    )
    updated_text = ""
    current_url = ""
    processing_link = False
    skipping_link_label = False
    for c in text:
        if processing_link:
            if c == "|":
                skipping_link_label = True
                continue
            if c == ">":
                processing_link = False
                skipping_link_label = False
                updated_text += current_url
                current_url = ""
                continue
            if skipping_link_label:
                continue
            current_url += c
        else:
            if c == "<":
                processing_link = True
                continue
            updated_text += c
    updated_text = html.unescape(updated_text)
    message_text = f"""@{user_name} が Slack から連絡\n---\n{updated_text}\n\nSlack を開く: {slack_channel_deep_link(channel_id)}"""
    line_bot_api.push_message(to=line_push_message_to_id, messages=[TextSendMessage(text=message_text)])


@slack_app.event(
    event={"type": "message", "subtype": "file_share"},
    matchers=[is_target_channel],
)
def send_file_to_line_from_channel(event: dict, client: WebClient):
    text, user_name, channel_id = (
        event.get("text", ""),
        get_user_name(client, event),
        event.get("channel"),
    )
    if text == "":
        text = "(本文なし)"

    file_preview = ""
    if event.get("files") is not None:
        for file in event.get("files"):
            preview = f"ファイル名: {file.get('name') or file.get('title')} ({file.get('pretty_type', '')})\n\n"
            if file.get("preview_plain_text") is not None:
                text_preview = (
                    file.get("preview_plain_text", "").replace("\r\n", "\n").replace("\r", "").replace("\n\n", "\n")
                )
                preview += text_preview + " ...\n(※ 続きは Slack で確認してください)\n\n"
            else:
                preview += "(※ このファイルは表示できないので Slack で確認してください)\n\n"
            file_preview += preview
    message_text = (
        f"""@{user_name} が Slack から連絡\n---\n{text}\n\n"""
        f"""-----\n(添付ファイルあり)\n{file_preview}\n-----\n\n"""
        f"""Slack を開く: {slack_channel_deep_link(channel_id)}"""
    )
    line_bot_api.push_message(to=line_push_message_to_id, messages=[TextSendMessage(text=message_text)])


@slack_app.event("message")
def just_ack():
    pass


# ---------------------------------
# LINE bot app
# ---------------------------------

line_handler = WebhookHandler(os.environ["LINE_CHANNEL_SECRET"])


def get_conversation(event: dict):
    conv = "場所不明"
    if event.source.type == "user":
        name = line_bot_api.get_profile(event.source.user_id).display_name
        conv = f"{name}とボットの会話"
    elif event.source.type == "room":
        room_id, user_id = event.source.room_id, event.source.user_id
        name = line_bot_api.get_room_member_profile(room_id, user_id).display_name
        conv = f"{name}を含む会話"
    elif event.source.type == "group":
        conv = line_bot_api.get_group_summary(event.source.group_id).group_name
    return conv


@line_handler.add(MessageEvent, message=TextMessage)
def handle_message(event: dict):
    logger.info(event)
    text = event.message.text
    messages = []
    if "住所" in text:
        message = f"僕の住所は「〒111-2222 東京都xxx」だよ！"
        messages.append(TextSendMessage(text=message))

    if len(messages) > 0:
        line_bot_api.reply_message(event.reply_token, messages)

    formatted_text = ">" + re.sub(r"\n", "\n>", text)
    slack_api.chat_postMessage(
        channel=slack_channel_id,
        text=text,
        blocks=[
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f":line: *{get_conversation(event)}* からの転送\n{formatted_text}",
                },
                "accessory": {
                    "type": "button",
                    "url": "https://line.me/R/nv/chat",
                    "text": {"type": "plain_text", "text": ":line: を開く"},
                    "value": "button",
                    "action_id": "open-line-app",
                },
            }
        ],
    )


@line_handler.add(MessageEvent, message=StickerMessage)
def handle_message(event: dict):
    info = "(" + " ".join(event.message.keywords[:5]) + ")" if event.message.keywords else ""
    slack_api.chat_postMessage(
        channel=slack_channel_id,
        text=f"LINE スタンプが送られました！ {info}",
        blocks=[
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f":line: *{get_conversation(event)}* からの転送\nスタンプが送られました！ {info}",
                },
                "accessory": {
                    "type": "button",
                    "url": "https://line.me/R/nv/chat",
                    "text": {"type": "plain_text", "text": ":line: を開く"},
                    "value": "button",
                    "action_id": "open-line-app",
                },
            }
        ],
    )


@line_handler.add(MessageEvent, message=ImageMessage)
def handle_message(event: dict):
    logger.info(event)
    message_content = line_bot_api.get_message_content(event.message.id)
    slack_api.files_upload(
        file=message_content.content,
        filename="line_uploaded_image.jpg",
        filetype="jpeg",
        channels=slack_channel_id,
        initial_comment=f":line: *{get_conversation(event)}* からの転送",
    )


@line_handler.add(MessageEvent, message=VideoMessage)
def handle_message(event: dict):
    logger.info(event)
    message_content = line_bot_api.get_message_content(event.message.id)
    slack_api.files_upload(
        file=message_content.content,
        filename="line_uploaded_video.mp4",
        filetype="mp4",
        channels=slack_channel_id,
        initial_comment=f":line: *{get_conversation(event)}* からの転送",
    )


# ---------------------------------
# Flask web app
# ---------------------------------

from flask import Flask, request, make_response

# Do not change this name - app
app = Flask(__name__)


from concurrent.futures.thread import ThreadPoolExecutor

line_handler_executor = ThreadPoolExecutor(1)


@app.route("/line/callback", methods=["POST"])
def callback():
    try:
        signature = request.headers.get("X-Line-Signature")
        body = request.get_data(as_text=True)
        logger.warning(f"Request body from LINE Messaging API: {body}")
        # async execution
        def run_line_handler():
            try:
                line_handler.handle(body, signature)
            except Exception as e:
                logger.exception(e)

        line_handler_executor.submit(run_line_handler)
        return "OK"

    except InvalidSignatureError:
        logger.warning("Invalid LINE request signature detected")
        return make_response("Invalid request", 400)
    except Exception:
        logger.exception("Failed to handle a request")
        return make_response("Something wrong", 500)


if slack_app_token is not None:
    slack_socket_mode_handler = SocketModeHandler(slack_app)
    slack_socket_mode_handler.connect()
else:

    @app.route("/slack/events", methods=["POST"])
    def slack_events():
        return slack_flask_handler.handle(request)


if __name__ == "__main__":
    app.run(port=3000)

# export LINE_CHANNEL_ACCESS_TOKEN=
# export LINE_CHANNEL_SECRET=
# export LINE_PUSH_MESSAGE_TO_ID=
# export SLACK_CHANNEL_ID=
# export SLACK_BOT_TOKEN=
# export SLACK_SIGNING_SECRET=
# FLASK_APP=main.py FLASK_ENV=development flask run -p 3000

Procfile

web: gunicorn --bind :$PORT --workers 1 --threads 10 --timeout 0 main:app

requirements.txt

slack-bolt>=1.14.0,<2
slack-sdk>=3.16.2,<4
line-bot-sdk>=2,<3
Flask>=2,<3
gunicorn>=20.1