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 jsonfrom flask import Flask, requestimport requests
app = Flask(__name__)
# 替换为你的 Microsoft Teams Incoming WebHook URLteams_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 flaskpip 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 jsonfrom flask import Flask, request, make_responseimport requests
app = Flask(__name__)
# 替换为你的 Microsoft Teams Incoming Webhook URLteams_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-sdkpip install flaskpip 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 jsonfrom flask import Flask, request, make_responseimport requests
app = Flask(__name__)
# 替换为你的 Microsoft Teams Incoming Webhook URLteams_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 jsonfrom flask import Flask, request, make_responseimport requests
app = Flask(__name__)
# 替换为你的 Microsoft Teams Incoming Webhook URLteams_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 jsonfrom flask import Flask, request, make_responseimport requests
app = Flask(__name__)
# 替换为你的 Microsoft Teams Incoming Webhook URLteams_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 jsonfrom flask import Flask, request, make_responseimport requests
app = Flask(__name__)
# 替换为你的 Microsoft Teams Incoming Webhook URLteams_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 jsonfrom flask import Flask, request, make_responseimport requestsimport os
app = Flask(__name__)
# 从环境变量中读取 Microsoft Teams Incoming Webhook URLteams_webhook_url = os.environ.get('TEAMS_WEBHOOK_URL')
# 从环境变量中读取 Slack bot tokenslack_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 jsonfrom flask import Flask, request, make_responseimport requestsimport osfrom dingtalkchatbot.chatbot import DingtalkChatbot
app = Flask(__name__)
# 从环境变量中读取 Microsoft Teams Incoming Webhook URLteams_webhook_url = os.environ.get('TEAMS_WEBHOOK_URL')
# 从环境变量中读取 Slack bot tokenslack_bot_token = os.environ.get('SLACK_BOT_TOKEN')
# 从环境变量中读取 Slack 签名密钥slack_signing_secret = os.environ.get('SLACK_SIGNING_SECRET')
# 从环境变量中读取钉钉机器人访问 tokendingtalk_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 jsonfrom flask import Flask, request, make_responseimport requestsimport osfrom dingtalkchatbot.chatbot import DingtalkChatbot
app = Flask(__name__)
# 从环境变量中读取 Microsoft Teams Incoming Webhook URLteams_webhook_url = os.environ.get('TEAMS_WEBHOOK_URL')
# 从环境变量中读取 Slack bot tokenslack_bot_token = os.environ.get('SLACK_BOT_TOKEN')
# 从环境变量中读取 Slack 签名密钥slack_signing_secret = os.environ.get('SLACK_SIGNING_SECRET')
# 从环境变量中读取钉钉访问的 tokendingtalk_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 jsonimport timefrom flask import Flask, request, make_responseimport requestsimport osfrom dingtalkchatbot.chatbot import DingtalkChatbot
app = Flask(__name__)
# 从环境变量中读取 Microsoft Teams Incoming Webhook URLteams_webhook_url = os.environ.get('TEAMS_WEBHOOK_URL')
# 从环境变量中读取 Slack bot tokenslack_bot_token = os.environ.get('SLACK_BOT_TOKEN')
# 从环境变量中读取 Slack 签名密钥slack_signing_secret = os.environ.get('SLACK_SIGNING_SECRET')
# 从环境变量中读取钉钉访问的 tokendingtalk_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 osfrom 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 DingTalkif 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：你是对的，这样只能防止同一个事件被重复处理，但仍然会收到其他事件。如果你想完全停止接收事件，你需要进行以下几个步骤：
从你的 Slack 应用程序中删除事件订阅，这可以防止 Slack 继续将事件发送给你的应用程序。
如果你使用的是 slack-bolt 库，则可以通过停止 app.start() 方法来退出应用程序。这将使应用程序停止运行，同时不再处理任何事件。
以下是一个示例代码，展示如何使用 slack-bolt 库在处理完一条消息事件后停止应用程序运行：
from slack_bolt import Appfrom 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 的一些基本步骤：
访问 Slack API Tester：打开您的网络浏览器并打开以下网址：https://api.slack.com/apps/{app_id}/api-tester，其中，{app_id} 是您的 Slack 应用程序的 ID，可以在您的应用程序设置页面中找到。
选择要测试的 API 方法：使用左侧的菜单选择要测试的 Slack API 方法。例如，如果您要测试 "chat.postMessage" 方法，请单击左侧菜单中的此方法。
设置 API 参数：根据所选 Slack API 方法的要求，输入必需的参数和选项。根据 API 要求，您可能需要输入文本消息、目标频道、用户 ID 等参数。
发送请求：单击 "Send" 按钮以发送 API 请求。您将看到 API 响应、状态代码和响应体（如果有）。
检查 API 响应：查看 API 响应体以确定 API 是否正常工作。如果响应体显示消息已成功发送或其他成功操作，则说明 API 已成功工作。如果出现错误或错误响应，可以在下一步中调试操作。
调试和调整：如果 API 未正常工作或收到错误响应，您可以在此工具中进行调试，并根据 API 文档中提供的指导调整操作。
您可以使用 Slack API Tester 来测试各种 Slack API 方法，例如发送消息、查询用户或列表、创建频道或更新工作区设置等。这是一个简单、易于使用且非常实用的工具，可帮助您进行快速且方便的 Slack API 测试和调试。
我：SLACK_BOT_TOKEN 如何生成？
GPT：要获得一个 Slack bot token，您需要按照以下步骤操作：
前往 Slack API，单击 "Get started for free" 进行注册/登录。
创建一个新的 Workspace 或使用现有的 Workspace。
转到"Create an App"并为您的 bot 创建一个新的 app。
单击 "Add features and functionality" > "Bots" > "Add a Bot User"。设置 bot 的显示名称和默认用户名。
在 "Install App" 页面中，单击 "Install App to Workspace" 并授权它访问您的 Slack Workspace。
安装完成后，您将获得一个 Bot token，可以在 "App Home" 页面中找到它。
在获得了您的 Slack bot token 后，您需要将其存储在环境变量中，这样您的应用程序才能使用它。在 Linux 或 macOS 中，您可以使用命令行设置环境变量。在 Windows 中，您可以使用控制面板或 Powershell。
评论