AI 协助办公 |记一次用 GPT-4 写一个消息同步 App

news2024/11/26 4:27:10

GPT-4 最近风头正劲,作为 NebulaGraph 的研发人员的我自然是跟进新技术步伐。恰好,现在有一个将 Slack channel 消息同步到其他 IM 的需求,看看 GPT-4 能不能帮我完成这次的信息同步工具的代码编写工作。

本文记录了我同 GPT 主要的交互过程,中间删减了部分非主线对话。

在开始同 GPT-4 谈 App 需求之前,先让他它科普下什么是 Outgoing Webhook,和 Incoming Webhook,以及 Slack 相关的运行机制。

GPT-4:当我们在 Slack 中设置了 Outgoing Webhook 之后,当用户在 Slack 上进行某些特定操作时,会触发 Outgoing Webhook,并将相关信息发送到我们指定的 URL 上。这些信息可以包括文本、图片、文件等等。而 Incoming Webhook 则是另一个平台(如微软 Teams)上的一个 URL,当我们将信息发送到该 URL 时,信息将显示在目标平台上。

GPT-4 写的第一版程序

我:能不能写个程序用来接收 Slack 的 Outgoing Webhook,对接到微软 Teams 的 Incoming Webhook

GPT:可以写一个这样的程序。

以下是一个使用 Python 和 Flask 库的简单示例,可以接收 Slack 的 Outgoing Webhook 并将消息发送到 Microsoft Teams 的 Incoming Webhook:

import json
from flask import Flask, request
import requests

app = Flask(__name__)

# 替换为你的 Microsoft Teams Incoming WebHook URL
teams_webhook_url = 'https://example.com/teams/webhook/url'

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 从 Slack Outgoing Webhook 提取文本和用户名
    slack_data = request.form
    text = slack_data.get('text')
    username = slack_data.get('user_name')

    # 使用用户名和文本创建 Microsoft Teams Message
    teams_payload = {
        "@context": "https://schema.org/extensions",
        "@type": "MessageCard",
        "text": f"From {username}: {text}",
    }

    # 将数据发送到 Microsoft Teams Incoming Webhook
    requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})
    
    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

在运行此代码之前,请确保已经安装了 flaskrequests 库:

pip install flask
pip install requests

在此示例中,我们使用 Flask 创建了一个简单的 Web 服务器,监听 5000 端口,并定义一个处理 Slack Webhooks 的端点。将此端点 URL 配置到你的 Slack Outgoing Webhook 中。

在处理收到的 Webhook 时,我们提取请求中的文本和用户名。然后,创建一个 Microsoft Teams 的 Incoming Webhook 负载,并将其发送到你的 Microsoft Teams Incoming Webhook 的 URL。

请注意,这是一个简单示例,没有处理错误和异常情况。在实际生产环境中,需要对这个程序进行适当的错误处理,以提高健壮性。

增加需求

我:哦,其实 Slack 推荐用的是它的 Event API,能改改吗

GPT:当然可以。使用 Slack Event API 时,需要修改代码以处理不同类型的事件,并确保验证来自 Slack 的请求。

以下是一个使用 Python 和 Flask 库的示例,根据 Slack Event API 将消息发送到 Microsoft Teams 的 Incoming Webhook:

import json
from flask import Flask, request, make_response
import requests

app = Flask(__name__)

# 替换为你的 Microsoft Teams Incoming Webhook URL
teams_webhook_url = 'https://example.com/teams/webhook/url'

# 设置你的 Slack 签名密钥
slack_signing_secret = 'your_signing_secret'

# 用于验证 Slack 请求的实用函数
def verify_slack_request(request):
    from slack_sdk.signature import SignatureVerifier
    verifier = SignatureVerifier(secret=slack_signing_secret)
    return verifier.is_valid_request(request)

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 验证请求是否来自 Slack
    if not verify_slack_request(request):
        return make_response("Request not verified", 403)

    # 处理事件数据
    event_data = request.json
    event_type = event_data.get('type')

    # 在事件 API 设置期间处理 URL 验证
    if event_type == 'url_verification':
        return event_data["challenge"]

    # 将消息转发到 Teams
    if event_type == 'event_callback':
        event = event_data['event']
        if event['type'] == 'message':
            username = event.get('username', 'unknown_user')
            text = event.get('text', '')

            # 创建 Teams 消息
            teams_payload = {
                "@context": "https://schema.org/extensions",
                "@type": "MessageCard",
                "text": f"From {username}: {text}",
            }

            # 将数据发送到 Microsoft Teams Incoming Webhook
            requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})
    
    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

在运行此代码之前,请确保已经安装了 slack-sdkflaskrequests 库:

pip install slack-sdk
pip install flask
pip install requests

在你的 Slack 应用中设置好 Event API,订阅 message.channels 事件,并将 slack_signing_secret 替换为你的 Slack 应用签名密钥。

现在,我们对代码进行了修改,不再接收 Outgoing Webhook,而是处理 Slack Event API 发送的事件。在收到来自 Slack 的事件时,我们首先验证发送方,然后根据事件类型进行处理。在此示例中,我们关注 message.channels 事件,将消息发送到 Microsoft Teams 的 Incoming Webhook。

请注意,这仍然是一个简化的示例。你可能需要根据具体的业务需求,优化这个程序及其错误处理。

过了几天之后,我发现这段代码有些问题。于是展开了和 GPT-4 的第二次对话。

我:

import json
from flask import Flask, request, make_response
import requests

app = Flask(__name__)

# 替换为你的 Microsoft Teams Incoming Webhook URL
teams_webhook_url = 'https://example.com/teams/webhook/url'

# 设置你的 Slack 签名密钥
slack_signing_secret = 'your_signing_secret'

# 用于验证 Slack 请求的实用函数
def verify_slack_request(request):
    from slack_sdk.signature import SignatureVerifier
    verifier = SignatureVerifier(secret=slack_signing_secret)
    return verifier.is_valid_request(request)

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 验证请求是否来自 Slack
    if not verify_slack_request(request):
        return make_response("Request not verified", 403)

    # 处理事件数据
    event_data = request.json
    event_type = event_data.get('type')

    # 在事件 API 设置期间处理 URL 验证
    if event_type == 'url_verification':
        return event_data["challenge"]

    # 将消息转发到 Teams
    if event_type == 'event_callback':
        event = event_data['event']
        if event['type'] == 'message':
            username = event.get('username', 'unknown_user')
            text = event.get('text', '')

            # 创建 Teams 消息
            teams_payload = {
                "@context": "https://schema.org/extensions",
                "@type": "MessageCard",
                "text": f"From {username}: {text}",
            }

            # 将数据发送到 Microsoft Teams Incoming Webhook
            requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})
    
    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

请修改上面这段代码,使它只转发 Alice 和 Bob 这两个用户发送的消息

GPT:在现有代码的基础上,可以添加一个条件检查,只有当用户是 Alice 或 Bob 时才将消息转发到 Microsoft Teams。

具体实现如下所示:

import json
from flask import Flask, request, make_response
import requests

app = Flask(__name__)

# 替换为你的 Microsoft Teams Incoming Webhook URL
teams_webhook_url = 'https://example.com/teams/webhook/url'

# 设置你的 Slack 签名密钥
slack_signing_secret = 'your_signing_secret'

# 用于验证 Slack 请求的实用函数
def verify_slack_request(request):
    from slack_sdk.signature import SignatureVerifier
    verifier = SignatureVerifier(secret=slack_signing_secret)
    return verifier.is_valid_request(request)

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 验证请求是否来自 Slack
    if not verify_slack_request(request):
        return make_response("Request not verified", 403)

    # 处理事件数据
    event_data = request.json
    event_type = event_data.get('type')

    # 在事件 API 设置期间处理 URL 验证
    if event_type == 'url_verification':
        return event_data["challenge"]

    # 将 Alice 或 Bob 的消息转发到 Teams
    if event_type == 'event_callback':
        event = event_data['event']
        if event['type'] == 'message':
            username = event.get('username', 'unknown_user')
            text = event.get('text', '')

            # 只将 Alice 或 Bob 的消息转发到 Teams
            if username in ('alice', 'bob'):
                # 创建 Microsoft Teams 消息
                teams_payload = {
                    "@context": "https://schema.org/extensions",
                    "@type": "MessageCard",
                    "text": f"From {username}: {text}",
                }

                # 将数据发送到 Microsoft Teams Incoming Webhook
                requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})
    
    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

在这个新代码中,我们添加了一个条件检查,只有当用户是 Alic 或 Bob 时才将消息转发到 Microsoft Teams。如果现有代码中包含其他名称,您可以将其添加到条件检查中进行过滤。

调试程序

我:这个 event 中并不包含 username 的信息,改一下代码

GPT:确实,在 Slack API Event 中,username 字段已经不再推荐使用,应改为使用 user 字段来获取用户的信息。

以下是修改后的代码:

import json
from flask import Flask, request, make_response
import requests

app = Flask(__name__)

# 替换为你的 Microsoft Teams Incoming Webhook URL
teams_webhook_url = 'https://example.com/teams/webhook/url'

# 设置你的 Slack 签名密钥
slack_signing_secret = 'your_signing_secret'

# 用于验证 Slack 请求的实用函数
def verify_slack_request(request):
    from slack_sdk.signature import SignatureVerifier
    verifier = SignatureVerifier(secret=slack_signing_secret)
    return verifier.is_valid_request(request)

# 从 Slack API 中获取用户信息的实用函数
def get_user_info(user_id):
    from slack_sdk import WebClient
    slack_bot_token = 'your_bot_token' # 替换为自己的 Bot Token
    client = WebClient(token=slack_bot_token)

    try:
        response = client.users_info(user=user_id)
        user_info = response['user']
        return user_info
    except:
        return None

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 验证请求是否来自 Slack
    if not verify_slack_request(request):
        return make_response("Request not verified", 403)

    # 处理事件数据
    event_data = request.json
    event_type = event_data.get('type')

    # 在事件 API 设置期间处理 URL 验证
    if event_type == 'url_verification':
        return event_data["challenge"]

    # 将 Alice 或 Bob 的消息转发到 Teams
    if event_type == 'event_callback':
        event = event_data['event']
        if event['type'] == 'message':
            user_id = event['user']
            user_info = get_user_info(user_id)

            # 只将 Alice 或 Bob 的消息转发到 Teams
            if user_info and user_info['name'] in ('alice', 'bob'):
                # 创建 Microsoft Teams 消息
                text = event.get('text', '')
                teams_payload = {
                    "@context": "https://schema.org/extensions",
                    "@type": "MessageCard",
                    "text": f"From {user_info['name']}: {text}",
                }

                # 将数据发送到 Microsoft Teams Incoming Webhook
                requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})
    
    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

在新代码中,我们使用 user 字段来获取用户的 ID,然后使用 get_user_info() 函数从 Slack API 中获取用户的完整信息,包括用户名。根据用户信息的用户名,我们将仅转发 Alice 和 Bob 发送的消息到 Microsoft Teams。

再次增加定制需求

我:再改一下,只转发来自 xx 这个组织的用户发送的消息

GPT:可以通过检查用户是否属于 xx Organization 来实现这一点。以下是相应的代码更改:

import json
from flask import Flask, request, make_response
import requests

app = Flask(__name__)

# 替换为你的 Microsoft Teams Incoming Webhook URL
teams_webhook_url = 'https://example.com/teams/webhook/url'

# 设置你的 Slack 签名密钥
slack_signing_secret = 'your_signing_secret'

# 用于验证 Slack 请求的实用函数
def verify_slack_request(request):
    from slack_sdk.signature import SignatureVerifier
    verifier = SignatureVerifier(secret=slack_signing_secret)
    return verifier.is_valid_request(request)

# 从 Slack API 获取用户信息的实用函数
def get_user_info(user_id):
    from slack_sdk import WebClient
    slack_bot_token = 'your_bot_token' # 替换为自己的 Bot Token
    client = WebClient(token=slack_bot_token)

    try:
        response = client.users_info(user=user_id)
        user_info = response['user']
        return user_info
    except:
        return None

# 检测用户是否在 xx 组织
def is_user_in_xorg_organization(user_info):
    user_email = user_info.get('profile', {}).get('email', '')
    return user_email.endswith('@xorg.com') if user_email else False

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 验证请求是否来自 Slack
    if not verify_slack_request(request):
        return make_response("Request not verified", 403)

    # 处理事件数据
    event_data = request.json
    event_type = event_data.get('type')

    # 在事件 API 设置期间处理 URL 验证
    if event_type == 'url_verification':
        return event_data["challenge"]

    # 转发 xx 组织里的消息到 Teams
    if event_type == 'event_callback':
        event = event_data['event']
        if event['type'] == 'message':
            user_id = event['user']
            user_info = get_user_info(user_id)

            # 只转发 xx 组织里的消息
            if user_info and is_user_in_xorg_organization(user_info):
                # 创建 Teams 消息
                text = event.get('text', '')
                teams_payload = {
                    "@context": "https://schema.org/extensions",
                    "@type": "MessageCard",
                    "text": f"From {user_info['name']}: {text}",
                }

                # 将数据发送到 Microsoft Teams Incoming Webhook
                requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})
    
    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

在新代码中,我们引入了一个名为 is_user_in_xorg_organization() 的新函数,它提取用户的电子邮件地址并检查是否以@xorg.com结尾。使用此函数,我们可以检查用户是否属于 xx 组织。如果为真,则我们将消息转发到 Microsoft Teams,否则我们将其忽略。

我:这里有一个 secret 和一个 token 都是 hardcode 的,改成读取系统的环境变量吧

GPT:是的,为了更安全地使用这些凭据,我们可以存储它们作为操作系统环境变量,然后在应用程序中使用它们。这样我们就不必明文存储这些凭据。

以下是相应的代码更改:

import json
from flask import Flask, request, make_response
import requests
import os

app = Flask(__name__)

# 从环境变量中读取 Microsoft Teams Incoming Webhook URL
teams_webhook_url = os.environ.get('TEAMS_WEBHOOK_URL')

# 从环境变量中读取 Slack bot token
slack_bot_token = os.environ.get('SLACK_BOT_TOKEN')

# 从环境变量中读取 Slack 签名密钥
slack_signing_secret = os.environ.get('SLACK_SIGNING_SECRET')

# 用于验证 Slack 请求的实用函数
def verify_slack_request(request):
    from slack_sdk.signature import SignatureVerifier
    verifier = SignatureVerifier(secret=slack_signing_secret)
    return verifier.is_valid_request(request)

# 从 Slack API 获取用户信息的实用函数
def get_user_info(user_id):
    from slack_sdk import WebClient
    client = WebClient(token=slack_bot_token)

    try:
        response = client.users_info(user=user_id)
        user_info = response['user']
        return user_info
    except:
        return None

# 判断用户是否在 xx 组织
def is_user_in_xorg_organization(user_info):
    user_email = user_info.get('profile', {}).get('email', '')
    return user_email.endswith('@xorg.com') if user_email else False

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 验证请求是否来自 Slack
    if not verify_slack_request(request):
        return make_response("Request not verified", 403)

    # 处理事件数据
    event_data = request.json
    event_type = event_data.get('type')

    # 在事件 API 设置期间处理 URL 验证
    if event_type == 'url_verification':
        return event_data["challenge"]

    # 转发 xx 组织里的消息到 Teams
    if event_type == 'event_callback':
        event = event_data['event']
        if event['type'] == 'message':
            user_id = event['user']
            user_info = get_user_info(user_id)

            # 只转发 xx 组织里的消息
            if user_info and is_user_in_xorg_organization(user_info):
                # 创建 Teams 消息
                text = event.get('text', '')
                teams_payload = {
                    "@context": "https://schema.org/extensions",
                    "@type": "MessageCard",
                    "text": f"From {user_info['name']}: {text}",
                }

                # 将数据发送到 Microsoft Teams Incoming Webhook
                requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})
    
    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

在这里,我们使用 os.environ.get() 从操作系统环境变量中读取凭据,而不直接将其硬编码到代码中。这使得更容易管理凭据,并且代码更加安全。

我:我还想把这个消息同时转发到钉钉

GPT:好的,你需要安装 dingtalkchatbot 模块来使用钉钉机器人 API。在安装之前,请确保已经在钉钉管理后台中创建了自定义机器人,并且已经获得了 Webhook URL。

以下是修改后的代码:

import json
from flask import Flask, request, make_response
import requests
import os
from dingtalkchatbot.chatbot import DingtalkChatbot

app = Flask(__name__)

# 从环境变量中读取 Microsoft Teams Incoming Webhook URL
teams_webhook_url = os.environ.get('TEAMS_WEBHOOK_URL')

# 从环境变量中读取 Slack bot token
slack_bot_token = os.environ.get('SLACK_BOT_TOKEN')

# 从环境变量中读取 Slack 签名密钥
slack_signing_secret = os.environ.get('SLACK_SIGNING_SECRET')

# 从环境变量中读取钉钉机器人访问 token
dingtalk_access_token = os.environ.get('DINGTALK_ACCESS_TOKEN')

# 从环境变量中读取钉钉机器人密钥
dingtalk_secret = os.environ.get('DINGTALK_SECRET')

# 从环境变量中读取钉钉机器人密钥
def verify_slack_request(request):
    from slack_sdk.signature import SignatureVerifier
    verifier = SignatureVerifier(secret=slack_signing_secret)
    return verifier.is_valid_request(request)

# 从 Slack API 获取用户信息的实用函数
def get_user_info(user_id):
    from slack_sdk import WebClient
    client = WebClient(token=slack_bot_token)

    try:
        response = client.users_info(user=user_id)
        user_info = response['user']
        return user_info
    except:
        return None

# 判断用户是否在 xx 组织
def is_user_in_xorg_organization(user_info):
    user_email = user_info.get('profile', {}).get('email', '')
    return user_email.endswith('@xorg.com') if user_email else False

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 验证请求是否来自 Slack
    if not verify_slack_request(request):
        return make_response("Request not verified", 403)

    # 处理事件数据
    event_data = request.json
    event_type = event_data.get('type')

    # 在事件 API 设置期间处理 URL 验证
    if event_type == 'url_verification':
        return event_data["challenge"]

    # 转发 xx 组织里的消息到 Teams 和钉钉
    if event_type == 'event_callback':
        event = event_data['event']
        if event['type'] == 'message':
            user_id = event['user']
            user_info = get_user_info(user_id)

            # 只转发 xx 组织里的消息
            if user_info and is_user_in_xorg_organization(user_info):
                # 创建 Teams 消息
                text = event.get('text', '')
                teams_payload = {
                    "@context": "https://schema.org/extensions",
                    "@type": "MessageCard",
                    "text": f"From {user_info['name']}: {text}",
                }

                # 将数据发送到 Microsoft Teams Incoming Webhook
                requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})

                # 创建钉钉消息
                dingtalk_robot = DingtalkChatbot(dingtalk_access_token, dingtalk_secret)
                dingtalk_payload = f"From {user_info['name']}: {text}"
                dingtalk_robot.send_text(msg=dingtalk_payload, is_at_all=False)

    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

在新代码中,我们添加了一个钉钉机器人发送消息的新模块,并在 slack_to_teams() 函数中添加了一些代码来将消息转发到钉钉。

请注意,您需要将 dingtalk_access_tokendingtalk_secret 设置为您自己的值,以便使用自己的钉钉机器人。并且,您需要安装 dingtalkchatbot 模块来使用此代码。通过在命令行中输入以下命令,安装此模块:

pip install dingtalkchatbot

我:这个很不错,我还希望支持多个 organization 的用户

GPT:好的,那么现在我们需要做的就是将检查用户是否在 xx 组织内的函数扩展为支持多个组织。

我们可以将所有要支持的组织电子邮件地址作为列表传递给 is_user_in_organization() 函数,这样函数将检查此邮箱地址是否以任一组织的标准结尾。

以下是代码更新后的版本:

import json
from flask import Flask, request, make_response
import requests
import os
from dingtalkchatbot.chatbot import DingtalkChatbot

app = Flask(__name__)

# 从环境变量中读取 Microsoft Teams Incoming Webhook URL
teams_webhook_url = os.environ.get('TEAMS_WEBHOOK_URL')

# 从环境变量中读取 Slack bot token
slack_bot_token = os.environ.get('SLACK_BOT_TOKEN')

# 从环境变量中读取 Slack 签名密钥
slack_signing_secret = os.environ.get('SLACK_SIGNING_SECRET')

# 从环境变量中读取钉钉访问的 token
dingtalk_access_token = os.environ.get('DINGTALK_ACCESS_TOKEN')

# 从环境变量中读取钉钉密钥
dingtalk_secret = os.environ.get('DINGTALK_SECRET')

# 包含组织及其域名的列表
organizations = ["@example.com", "@org.com", "@edu.com"]

# 用于验证 Slack 请求的实用函数
def verify_slack_request(request):
    from slack_sdk.signature import SignatureVerifier
    verifier = SignatureVerifier(secret=slack_signing_secret)
    return verifier.is_valid_request(request)

# 从 Slack API 获取用户信息的实用函数
def get_user_info(user_id):
    from slack_sdk import WebClient
    client = WebClient(token=slack_bot_token)

    try:
        response = client.users_info(user=user_id)
        user_info = response['user']
        return user_info
    except:
        return None

# 判断用户是否在指定的组织内
def is_user_in_organization(user_info):
    user_email = user_info.get('profile', {}).get('email', '')
    for org in organizations:
        if user_email.endswith(org):
            return True
    return False

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 验证请求是否来自 Slack
    if not verify_slack_request(request):
        return make_response("Request not verified", 403)

    # 处理事件数据
    event_data = request.json
    event_type = event_data.get('type')

    # 在事件 API 设置期间处理 URL 验证
    if event_type == 'url_verification':
        return event_data["challenge"]

    # 转发指定组织里的消息到 Teams 和钉钉
    if event_type == 'event_callback':
        event = event_data['event']
        if event['type'] == 'message':
            user_id = event['user']
            user_info = get_user_info(user_id)

            # 只转发特定组织内的消息
            if user_info and is_user_in_organization(user_info):
                # 创建 Teams 消息
                text = event.get('text', '')
                teams_payload = {
                    "@context": "https://schema.org/extensions",
                    "@type": "MessageCard",
                    "text": f"From {user_info['name']}: {text}",
                }

                # 将数据发送到 Microsoft Teams Incoming Webhook
                requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})

                # 创建钉钉消息
                dingtalk_robot = DingtalkChatbot(dingtalk_access_token, dingtalk_secret)
                dingtalk_payload = f"From {user_info['name']}: {text}"
                dingtalk_robot.send_text(msg=dingtalk_payload, is_at_all=False)

    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

在这里,我们已经将多个组织存储在 organizations 列表中,并将其传递给 is_user_in_organization() 函数作为一个额外参数。函数将检查电子邮件地址是否以任何组织的标准结尾。

我:event = event_data['event'] 之后打印一条含时间信息的 log 到 console

GPT:好的,那么我们可以在事件处理程序中增加打印日志信息的代码。为了打印时间信息,我们需要导入 Python 内置的 time 模块。

请尝试以下代码更新:

import json
import time
from flask import Flask, request, make_response
import requests
import os
from dingtalkchatbot.chatbot import DingtalkChatbot

app = Flask(__name__)

# 从环境变量中读取 Microsoft Teams Incoming Webhook URL
teams_webhook_url = os.environ.get('TEAMS_WEBHOOK_URL')

# 从环境变量中读取 Slack bot token
slack_bot_token = os.environ.get('SLACK_BOT_TOKEN')

# 从环境变量中读取 Slack 签名密钥
slack_signing_secret = os.environ.get('SLACK_SIGNING_SECRET')

# 从环境变量中读取钉钉访问的 token
dingtalk_access_token = os.environ.get('DINGTALK_ACCESS_TOKEN')

# 从环境变量中读取钉钉密钥
dingtalk_secret = os.environ.get('DINGTALK_SECRET')

# 包含组织及其域名的列表
organizations = ["@example.com", "@org.com", "@edu.com"]

# 用于验证 Slack 请求的实用函数
def verify_slack_request(request):
    from slack_sdk.signature import SignatureVerifier
    verifier = SignatureVerifier(secret=slack_signing_secret)
    return verifier.is_valid_request(request)

# 从 Slack API 获取用户信息的实用函数
def get_user_info(user_id):
    from slack_sdk import WebClient
    client = WebClient(token=slack_bot_token)

    try:
        response = client.users_info(user=user_id)
        user_info = response['user']
        return user_info
    except:
        return None

# 判断用户是否在指定的组织内
def is_user_in_organization(user_info):
    user_email = user_info.get('profile', {}).get('email', '')
    for org in organizations:
        if user_email.endswith(org):
            return True
    return False

@app.route('/slack-to-teams', methods=['POST'])
def slack_to_teams():
    # 验证请求是否来自 Slack
    if not verify_slack_request(request):
        return make_response("Request not verified", 403)

    # 处理事件数据
    event_data = request.json
    event_type = event_data.get('type')

    # 在事件 API 设置期间处理 URL 验证
    if event_type == 'url_verification':
        return event_data["challenge"]

    # 转发指定组织里的消息到 Teams 和钉钉
    if event_type == 'event_callback':
        event = event_data['event']
        if event['type'] == 'message':
            user_id = event['user']
            user_info = get_user_info(user_id)

            # 只转发特定组织内的消息
            if user_info and is_user_in_organization(user_info):
                # 记录事件
                message = event.get('text', '')
                timestamp = event.get('ts')
                username = user_info.get('name', 'unknown_user')
                print(f'[INFO] Message "{message}" from {username} at {time.ctime(float(timestamp))}')

                # 创建 Teams 消息
                teams_payload = {
                    "@context": "https://schema.org/extensions",
                    "@type": "MessageCard",
                    "text": f"From {user_info['name']}: {message}",
                }

                # 将数据发送到 Microsoft Teams Incoming Webhook
                requests.post(teams_webhook_url, json.dumps(teams_payload), headers={'Content-Type': 'application/json'})

                # 创建钉钉消息
                dingtalk_robot = DingtalkChatbot(dingtalk_access_token, dingtalk_secret)
                dingtalk_payload = f"From {user_info['name']}: {message}"
                dingtalk_robot.send_text(msg=dingtalk_payload, is_at_all=False)

    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

在这个版本的代码中,我们添加了新行,记录事件的处理时间和消息内容,这将帮助您更好地跟踪代码的执行。

我:我的钉钉机器人支持 Webhook,怎么修改代码

GPT:如果您的钉钉机器人支持 Webhook,则可以将以下代码添加到 Flask 应用程序中的相应路由处理程序中,以将消息发送到钉钉机器人:

import requests

def send_dingtalk_message(message):
    webhook_url = 'https://oapi.dingtalk.com/robot/send?access_token=YOUR_ACCESS_TOKEN'
    headers = {'Content-Type': 'application/json'}
    data = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }
    response = requests.post(webhook_url, headers=headers, json=data)
    return response.json()

请确保将 YOUR_ACCESS_TOKEN 替换为您的钉钉机器人的实际访问令牌。然后,您可以在 forward_message 函数中添加以下代码,以使用上面定义的 send_dingtalk_message 函数将消息转发到钉钉机器人:

elif organization_name == 'dingtalk':
    send_dingtalk_message(message)

这将检查传递的组织名称是否是 dingtalk,并在机器人支持 Webhook 时发送消息到钉钉机器人。

请注意,上述代码仅是示例,您需要根据详细的钉钉机器人文档和需求来处理,以实现功能了通知机器人钉钉。

我:

def get_user_info(user_id):
    from slack_sdk import WebClient
    client = WebClient(token=slack_bot_token)

    try:
        response = client.users_info(user=user_id)
        user_info = response['user']
        return user_info
    except Exception as e:
        print(f"Error getting user info for user {user_id}: {e}")
        return None

这段代码取不到用户的 email

GPT:是的,这段代码只能获取到 Slack 用户的一些基本信息,并不能获取用户的电子邮件地址。如果您需要获取用户的电子邮件地址,您需要在 Slack 应用设置中申请相应的权限,并使用权限范围来请求访问用户数据。

在 Slack 应用设置中,您需要申请 "users:read.email" 权限,以便在通过 Web API 请求用户信息时获取带有电子邮件地址的用户数据。然后,您可以在通过 WebClient 获取用户信息时设置相应参数来请求包含电子邮件地址的用户数据。修改后的示例代码如下:

def get_user_info(user_id):
    from slack_sdk import WebClient
    client = WebClient(token=slack_bot_token)

    try:
        response = client.users_info(user=user_id, include_locale=True, include_email=True)
        user_info = response['user']
        return user_info
    except Exception as e:
        print(f"Error getting user info for user {user_id}: {e}")
        return None

在此代码中,我们将 "include_email" 参数设置为 "True",以便从 Slack API 获取用户电子邮件地址。然后使用 "users_info" 方法来获取用户信息。具体如何设置权限管理和如何通过 SDK 获取详细信息,请参阅 Slack API 文档。

我:看上去 Slack Event 对同一个条消息发送了多次,如何避免

GPT:确保你已经在你的应用代码中启用了事件架构,在你的请求处理程序中使用了 acknowledge() 方法,以正确响应事件挂起流。当你使用 acknowledge() 方法来响应事件时,Slack API 会自动将事件流挂起状态水印标识为 response_metadata.next。到下一个事件传递时,这个水印将根据成功处理的事件数量进行更新。

如果你的应用已经配置了事件订阅,你还可以跟踪最近一次处理每种事件类型的时间,并在你的应用代码中使用此信息来过滤多余的事件,保证只处理一次。通常情况下,过滤非常简单,因为重复事件的时间戳几乎相同。

代码示例如下所示,运行此代码之前,你需要先将 SLACK_APP_TOKENSLACK_BOT_TOKEN 的值替换为你自己的应用令牌和机器人令牌。

import os
from slack_bolt import App, Ack

app = App(
    token=os.environ.get("SLACK_BOT_TOKEN"),
    signing_secret=os.environ.get("SLACK_APP_SECRET")
)

last_timestamp = {}

@app.event("message")
def handle_message(event, say, logger):
    # 检查此消息是否已被处理
    message_ts = float(event["event_ts"])
    last_ts = last_timestamp.get(event["channel"])
    if last_ts and message_ts <= last_ts:
        logger.info(f"Ignoring duplicate event {event['event_ts']} for channel {event['channel']}")
        return Ack()
    last_timestamp[event["channel"]] = message_ts

    # 处理事件
    logger.info(f"Received message {event['text']} in channel {event['channel']}")
    say("Hi there! I received your message.")
    return Ack()

在上面的示例代码中,我们跟踪每个频道中最近处理的事件的时间戳,并检查随后收到的事件是否已经处理过。如果是,我们将简单地响应事件挂起,告诉 Slack 不需要重新发送此事件的更多副本。如果不是,则处理事件并更新时间戳。这可以有效地防止对相同事件的多次响应。

我:

# Forward message from users in specified organizations to Teams and DingTalk
if event_type == 'event_callback':
    event = event_data['event']
    if event['type'] == 'message':
        user_id = event['user']
        user_info = get_user_info(user_id)
        # Only forward messages from users in specified organizations
        if user_info and is_user_in_organization(user_info):
            # Log the event
            message = event.get('text', '')
            timestamp = event.get('ts')
            username = user_info.get('name', 'unknown_user')
            print(f'[INFO] Message "{message}" from {username} at {time.ctime(float(timestamp))}')

这段代码会重复接收到 Slack 消息,怎么改

GPT:这段代码中的事件处理程序确实可能会多次处理同一条消息事件,这是由于多个订阅了该事件类型的应用程序同时接收到该消息事件所导致的。

为避免重复消息事件的处理,你可以考虑添加更详细的逻辑,例如跟踪已经处理过的事件,以确保同一事件不会被处理多次。以下是一个概念示例,你可以参考此示例来解决你的问题:

event_ids = set()
def handle_message_event(event_data):
    global event_ids
    event_id = event_data.get('event_id')
    if event_id is not None and event_id not in event_ids:
        event_type = event_data.get('event_type')
        event = event_data.get('event')
        if event_type == 'message':
            user_id = event['user']
            user_info = get_user_info(user_id)
            # 只转发指定组织内的消息
            if user_info and is_user_in_organization(user_info):
                # 记录事件
                message = event.get('text', '')
                timestamp = event.get('ts')
                username = user_info.get('name', 'unknown_user')
                print(f'[INFO] Message "{message}" from {username} at {time.ctime(float(timestamp))}')
                # 将事件 ID 添加到集合中,以避免重复处理
                event_ids.add(event_id)

在此示例中,我们使用了一个全局的 event_ids 集合来跟踪已经处理过的事件。当处理消息事件时,我们首先检查该事件是否已经存在于 event_ids 集合中,如果没有,则继续处理该事件并将其 event_id 添加到 event_ids 集合中。这样可以确保每个消息事件只被处理一次。

我:这样还不能阻止 Slack 继续发送 event,而只是不处理

GPT:你是对的,这样只能防止同一个事件被重复处理,但仍然会收到其他事件。如果你想完全停止接收事件,你需要进行以下几个步骤:

  1. 从你的 Slack 应用程序中删除事件订阅,这可以防止 Slack 继续将事件发送给你的应用程序。
  2. 如果你使用的是 slack-bolt 库,则可以通过停止 app.start() 方法来退出应用程序。这将使应用程序停止运行,同时不再处理任何事件。

以下是一个示例代码,展示如何使用 slack-bolt 库在处理完一条消息事件后停止应用程序运行:

from slack_bolt import App
from flask import Flask, request

app = App(token="YOUR_BOT_TOKEN")

@app.event("message")
def handle_message_event(payload):
    # 对消息事件执行某些操作
    # ...
    # 完成处理一个消息事件后,停止应用程序。
    app.stop()

if __name__ == "__main__":
    app.start(port=int(os.environ.get("PORT", 3000)))

在这个示例代码中,我们从 Slack 获取一个或多个消息事件,当处理完一个消息事件后,调用 app.stop() 方法来停止应用程序的运行。如果你想要一直处理消息事件而不停止应用程序,请忽略上述例子中的 app.stop() 方法。

至此,就完成了一个转发特定 Slack 信息到 Teams 和钉钉的工具开发。

后续

目前来说,GPT 最后完成的代码初步可用,有一些 Python 环境问题需要用 sudo 来运行。

完整代码:https://github.com/xiajingchun/slack_integration

在添加 Slack App 时,有一个注意点是需要订阅 messages.groups 方能接收 private channel 例如 slack-connect 的消息。

支线任务:Slack 的开发者工具

我:Slack API Tester 怎么使用

GPT:Slack API Tester 是 Slack 开发工具包提供的一个简单的在线工具,用于测试 Slack API 功能。使用 Slack API Tester,您可以通过简单的 Web 界面轻松测试各种 Slack API 操作,而无需编写任何代码。

以下是在 Slack API Tester 中使用 API 的一些基本步骤:

  1. 访问 Slack API Tester:打开您的网络浏览器并打开以下网址:https://api.slack.com/apps/{app_id}/api-tester,其中,{app_id} 是您的 Slack 应用程序的 ID,可以在您的应用程序设置页面中找到。
  2. 选择要测试的 API 方法:使用左侧的菜单选择要测试的 Slack API 方法。例如,如果您要测试 "chat.postMessage" 方法,请单击左侧菜单中的此方法。
  3. 设置 API 参数:根据所选 Slack API 方法的要求,输入必需的参数和选项。根据 API 要求,您可能需要输入文本消息、目标频道、用户 ID 等参数。
  4. 发送请求:单击 "Send" 按钮以发送 API 请求。您将看到 API 响应、状态代码和响应体(如果有)。
  5. 检查 API 响应:查看 API 响应体以确定 API 是否正常工作。如果响应体显示消息已成功发送或其他成功操作,则说明 API 已成功工作。如果出现错误或错误响应,可以在下一步中调试操作。
  6. 调试和调整:如果 API 未正常工作或收到错误响应,您可以在此工具中进行调试,并根据 API 文档中提供的指导调整操作。

您可以使用 Slack API Tester 来测试各种 Slack API 方法,例如发送消息、查询用户或列表、创建频道或更新工作区设置等。这是一个简单、易于使用且非常实用的工具,可帮助您进行快速且方便的 Slack API 测试和调试。

我:SLACK_BOT_TOKEN 如何生成

GPT:要获得一个 Slack bot token,您需要按照以下步骤操作:

  1. 前往 Slack API,单击 “Get started for free” 进行注册/登录。
  2. 创建一个新的 Workspace 或使用现有的 Workspace。
  3. 转到"Create an App"并为您的 bot 创建一个新的 app。
  4. 单击 “Add features and functionality” > “Bots” > “Add a Bot User”。设置 bot 的显示名称和默认用户名。
  5. 在 “Install App” 页面中,单击 “Install App to Workspace” 并授权它访问您的 Slack Workspace。
  6. 安装完成后,您将获得一个 Bot token,可以在 “App Home” 页面中找到它。

在获得了您的 Slack bot token 后,您需要将其存储在环境变量中,这样您的应用程序才能使用它。在 Linux 或 macOS 中,您可以使用命令行设置环境变量。在 Windows 中,您可以使用控制面板或 Powershell。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/616182.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

不定积分练习

不定积分练习 在看视频的时候遇到了一道比较有趣的题&#xff0c;在这里给大家分享一下。 题目 计算 ∫ ( 1 x − 1 x ) e x 1 x d x \int(1x-\dfrac 1x)e^{x\frac 1x}dx ∫(1x−x1​)exx1​dx 解&#xff1a; \qquad 原式 ∫ e x 1 x d x ∫ x ( 1 − 1 x 2 ) e x 1…

ESP Certificate Bundle 分享

基本概念 数字签名&#xff1a;是一种将相当于现实世界中的盖章、签字的功能在计算机世界中进行实现的技术。使用数字签名可以识别篡改和伪装&#xff0c;还可以防止否认。 证书&#xff1a;要开车得先靠驾照&#xff0c;驾照上面记有本人的照片、姓名、出生日期等个人信息&a…

【大唐杯学习超快速入门】5G智能网络优化

这里写目录标题 学习--实验背景上下行速率VOLTE掉话率优化时延优化 接入保持特性提升无线接通率切换成功率 附录 数据分析其他几种选项差不多 学习–实验背景 观看视频进行学习&#xff0c;理解该仿真的内涵 使用人工智能代替人工分析&#xff0c;对5G网络进行优化 上下行速率…

Centos Linux 操作系统中配置Gitlab服务器

基本准备 安装常用的工具包 linux根据操作系统的不同&#xff0c;有不同的安装工具&#xff0c;如&#xff0c; 操作系统 格式 工具 Debian .deb apt, apt-cache, apt-get, dpkg Ubuntu .deb apt, apt-cache, apt-get, dpkg CentOS .rpm yum Fedora .rpm dnf …

企业数智底座白皮书:详解数智平台必备的六大能力

在以“升级企业数智化底座”为主题的2023用友BIP技术大会上&#xff0c;用友联合全球权威咨询机构IDC共同发布《建设数字中国 升级数智底座——企业数智化底座白皮书》&#xff0c;在这本数智平台白皮书里详细解读了企业数智平台应该具备的六大基础能力。 当前企业数智化能力进…

一文剖析mlock锁原理

一般用户空间关联的物理页面是按需通过缺页异常的方式分配和调页&#xff0c;当系统物理内存不足时页面回收算法会回收一些最近很少使用的页面&#xff0c;但是有时候我们需要锁住一些物理页面防止其被回收&#xff08;如时间有严格要求的应用&#xff09;&#xff0c;Linux中提…

市面上的智能电表都有多少安的?适用于那些场景?

随着智能化技术的快速发展&#xff0c;智能电表已经成为了家庭和公寓物业配电系统的必备设备之一。智能电表的功率等级通常为10A或15A&#xff0c;具体取决于电表型号和用途。那么&#xff0c;这些功率等级的智能电表适用于哪些场景呢&#xff1f; 首先&#xff0c;10A功率等级…

【实验练习】基于自注意力机制Vision Transformer模型实现人脸朝向识别 (Python实现) 内容原创

题目 人脸识别是一个复杂的模式识别问题&#xff0c;人脸识别是人脸应用研究中非常重要的一步。由于人脸形状不规则、光线和背景条件多样&#xff0c;导致人脸检测精度受限。实际应用中&#xff0c;大量图像和视频源中人脸的位置、朝向、朝向角度都不是固定的&#xff0c;极大…

内网渗透—隧道技术

内网渗透—隧道技术 1. 隧道技术介绍1.1. 内网—隧道技术1.2. 常见的隧道协议1.3. 前置条件1.4. 判断内网的连通性 2. 网络层隧道技术2.1. ICMP隧道技术2.1.1. 常见工具2.1.2. Pingtunnel基础演示2.1.2.1. 下载服务端2.1.2.2. 下载客户端2.1.2.3. 设置CS连接2.1.2.4. 连接测试 …

功能上新|对比分析、Batches数量、函数释义Tips

本篇是继功能上新&#xff5c;内存篇、GPU篇之后&#xff0c;为大家展示更多关于提升浏览UWA GOT Online Overview报告体验的优化项&#xff0c;包括Overview报告的对比分析、Batches数量、函数释义Tips等。这些功能可以让你更快上手对报告的理解&#xff0c;亦或者更好地融入在…

.Net Core 6 WebApi 项目搭建(一、简单搭建)

前言 对于后端开发者最耻辱的是什么&#xff0c;是只会增删改查&#xff0c;只会CV&#xff0c;只会业务代码。没错&#xff0c;我就是被钉在耻辱柱上的一员&#xff0c;3年开发经验&#xff0c;不会搭建框架&#xff0c;只会写业务代码&#xff0c;丢人丢人啊&#xff0c;所以…

【XR】One More Thing:Vision Pro ,7年磨一剑,2023WWDC苹果发布Vision MR

One More Thing&#xff1a;Vision Pro &#xff0c;7年磨一剑&#xff0c;2023WWDC苹果发布Vision MR 1. 苹果MR Vision Pro&#xff1a;1. 专利布局&#xff1a;苹果表示在开发过程中申请了5000多项专利。2. 专属感知计算芯片3. 显示屏系统方面4. 续航方面5. Vision MR 的新框…

MATLAB安装配置MinGW-w64 C++编译器

文章目录 前言一、Mingw安装1、安装教程2、验证 二、MATLAB安装配置MinGW总结 #pic_center 前言 只是为方便学习&#xff0c;不做其他用途 一、Mingw安装 在网上找到的安装一直报错&#xff1a;The file has been downloaded incorrectly 1、安装教程 建议参考博客Mingw快捷安…

C++内存序、屏障和原子操作

文章目录 一、原子类型二、原子操作函数三、内存序1&#xff09;happens-before和synchronizes-with语义2&#xff09;内存序模式 四、标准库函数五、栅栏&#xff08;Barrier&#xff09; 一、原子类型 标准原子类型的备选名和与其相关的 std::atomic<> 特化类&#xf…

探索低代码的新形态(D2C、ChatGPT)

前言 低代码平台的出现&#xff0c;是互联网快速发展的背景下&#xff0c;满足产品快速迭代的实际需求。现在国内外都已经拥有非常多优秀的开源项目&#xff08;如&#xff1a;lowcode-engine&#xff09;和成熟的商业产品&#xff08;如&#xff1a;Mendix 、PowerPlatform&a…

Orillusion次时代 WebGPU 引擎

Orillusion 次时代 WebGPU 引擎 官网: https://www.orillusion.com/ 教程: https://www.orillusion.com/guide/ Orillusion 引擎是一款完全支持 WebGPU 标准的轻量级渲染引擎。基于最新的 Web 图形API标准&#xff0c;我们做了大量的探索和尝试&#xff0c;实现了很多曾经在 We…

python接口自动化(三)--如何设计接口测试用例(详解)

在开始接口测试之前&#xff0c;我们来想一下&#xff0c;如何进行接口测试的准备工作。或者说&#xff0c;接口测试的流程是什么&#xff1f;有些人就很好奇&#xff0c;接口测试要流程干嘛&#xff1f;不就是拿着接口文档直接利用接口 测试工具测试嘛。其实&#xff0c;如果…

【正点原子STM32连载】 第二十八章 低功耗实验摘自【正点原子】STM32F103 战舰开发指南V1.2

1&#xff09;实验平台&#xff1a;正点原子stm32f103战舰开发板V4 2&#xff09;平台购买地址&#xff1a;https://detail.tmall.com/item.htm?id609294757420 3&#xff09;全套实验源码手册视频下载地址&#xff1a; http://www.openedv.com/thread-340252-1-1.html 第二十…

计算机中数据的表示:定点数、浮点数

文章目录 1 概述2 定点数2.1 表示方法2.2 取值范围2.3 运算方法 3 浮点数3.1 表示方法3.2 运算方法 4 扩展4.1 等比数列前 n 项和公式 1 概述 #mermaid-svg-EXDrkn8G91FsDdps {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#merm…

给 a 标签设置 display:inline-block 之后 a 整体下沉

今天给一个a设置宽高&#xff0c;前提是添加了display:inline-block&#xff1b;然后发下a没有与父元素div顶部对齐&#xff0c;反而下沉了。试了好多办法都没成功&#xff0c;然后在网上找的教程。 原因 1、问题就是出在了display:inline-block;语句上&#xff0c;行内块元…