关于nebula:AI-协助办公-｜记一次用-GPT4-写一个消息同步-App

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)

在运行此代码之前，请确保曾经装置了 flask 和 requests 库：

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-sdk、flask 和 requests 库：

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_token 和 dingtalk_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_TOKEN 和 SLACK_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。