与沟通工具集成: 构建结果通知与团队协作
2025/8/30大约 12 分钟
在现代软件开发团队中，及时有效的沟通是确保项目成功的关键因素。通过将CI/CD平台与团队常用的沟通工具（如Slack、钉钉、Microsoft Teams等）集成，团队能够实时获得构建、测试、部署等关键环节的状态通知，促进团队成员之间的协作，并快速响应和解决问题。本文将深入探讨如何实现CI/CD平台与主流沟通工具的深度集成。
沟通工具集成的价值
沟通工具集成不仅能够提升团队的响应速度，还能增强工作流程的透明度。
1. 实时状态通知
通过集成实现关键事件的实时通知：
- 构建状态通知：及时告知团队成员构建成功或失败
- 部署状态通知：通知部署进度和结果
- 测试结果通知：分享自动化测试的结果
- 安全告警通知：及时通报安全扫描发现的问题
2. 上下文丰富的信息
提供包含丰富上下文信息的通知：
- 详细链接：直接链接到相关页面（构建详情、部署日志等）
- 环境信息：明确标识操作涉及的环境
- 责任人信息：显示触发操作的人员
- 时间戳：精确记录事件发生时间
3. 促进团队协作
通过沟通工具促进团队协作：
- 快速响应：团队成员可以快速响应通知中的问题
- 知识共享：分享最佳实践和解决方案
- 决策支持：提供数据支持的决策依据
- 流程透明：增强工作流程的可见性
Slack集成实践
Slack作为广泛使用的团队沟通工具，与其集成能够实现丰富的通知功能。
1. Webhook配置
Slack支持通过Incoming Webhooks接收外部通知：
Webhook配置示例
# Slack集成配置
slack_integration:
enabled: true
webhooks:
build_notifications:
url: "${SLACK_BUILD_WEBHOOK_URL}"
channel: "#ci-cd-notifications"
username: "CI/CD Bot"
icon_emoji: ":robot_face:"
deployment_notifications:
url: "${SLACK_DEPLOYMENT_WEBHOOK_URL}"
channel: "#deployments"
username: "Deployment Bot"
icon_emoji: ":rocket:"
alerts:
url: "${SLACK_ALERT_WEBHOOK_URL}"
channel: "#alerts"
username: "Alert Bot"
icon_emoji: ":rotating_light:"
message_templates:
build_success:
color: "good"
emoji: ":white_check_mark:"
build_failure:
color: "danger"
emoji: ":x:"
deployment_success:
color: "good"
emoji: ":rocket:"
deployment_failure:
color: "danger"
emoji: ":boom:"消息发送工具
#!/usr/bin/env python3
"""
Slack集成工具
实现与Slack的深度集成
"""
import requests
import json
import logging
from typing import Dict, List, Optional
from datetime import datetime
class SlackIntegration:
def __init__(self, config: Dict):
self.config = config
self.webhooks = config.get('webhooks', {})
self.message_templates = config.get('message_templates', {})
self.logger = logging.getLogger(__name__)
def send_build_notification(self, build_info: Dict) -> bool:
"""发送构建通知"""
try:
webhook_config = self.webhooks.get('build_notifications')
if not webhook_config:
self.logger.error("Build notifications webhook not configured")
return False
# 构建消息内容
message = self._build_build_message(build_info)
# 发送消息
return self._send_message(webhook_config['url'], message)
except Exception as e:
self.logger.error(f"Failed to send build notification: {e}")
return False
def send_deployment_notification(self, deployment_info: Dict) -> bool:
"""发送部署通知"""
try:
webhook_config = self.webhooks.get('deployment_notifications')
if not webhook_config:
self.logger.error("Deployment notifications webhook not configured")
return False
# 构建消息内容
message = self._build_deployment_message(deployment_info)
# 发送消息
return self._send_message(webhook_config['url'], message)
except Exception as e:
self.logger.error(f"Failed to send deployment notification: {e}")
return False
def send_alert_notification(self, alert_info: Dict) -> bool:
"""发送告警通知"""
try:
webhook_config = self.webhooks.get('alerts')
if not webhook_config:
self.logger.error("Alerts webhook not configured")
return False
# 构建消息内容
message = self._build_alert_message(alert_info)
# 发送消息
return self._send_message(webhook_config['url'], message)
except Exception as e:
self.logger.error(f"Failed to send alert notification: {e}")
return False
def _build_build_message(self, build_info: Dict) -> Dict:
"""构建构建通知消息"""
status = build_info.get('status', 'unknown')
template = self.message_templates.get(f'build_{status}', {})
status_emoji = template.get('emoji', ':question:')
status_color = template.get('color', '#cccccc')
message = {
'channel': self.webhooks.get('build_notifications', {}).get('channel', '#general'),
'username': self.webhooks.get('build_notifications', {}).get('username', 'CI/CD Bot'),
'icon_emoji': self.webhooks.get('build_notifications', {}).get('icon_emoji', ':robot_face:'),
'attachments': [
{
'color': status_color,
'blocks': [
{
'type': 'header',
'text': {
'type': 'plain_text',
'text': f'{status_emoji} 构建通知 - {build_info.get("project_name", "Unknown")}'
}
},
{
'type': 'section',
'fields': [
{
'type': 'mrkdwn',
'text': f'*构建ID:*\n{build_info.get("build_id", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*状态:*\n{status.capitalize()}'
},
{
'type': 'mrkdwn',
'text': f'*触发者:*\n{build_info.get("triggered_by", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*分支:*\n{build_info.get("branch", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*开始时间:*\n{build_info.get("start_time", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*结束时间:*\n{build_info.get("end_time", "Unknown")}'
}
]
},
{
'type': 'actions',
'elements': [
{
'type': 'button',
'text': {
'type': 'plain_text',
'text': '查看构建详情'
},
'url': build_info.get('build_url', '#')
}
]
}
]
}
]
}
return message
def _build_deployment_message(self, deployment_info: Dict) -> Dict:
"""构建部署通知消息"""
status = deployment_info.get('status', 'unknown')
template = self.message_templates.get(f'deployment_{status}', {})
status_emoji = template.get('emoji', ':question:')
status_color = template.get('color', '#cccccc')
message = {
'channel': self.webhooks.get('deployment_notifications', {}).get('channel', '#deployments'),
'username': self.webhooks.get('deployment_notifications', {}).get('username', 'Deployment Bot'),
'icon_emoji': self.webhooks.get('deployment_notifications', {}).get('icon_emoji', ':rocket:'),
'attachments': [
{
'color': status_color,
'blocks': [
{
'type': 'header',
'text': {
'type': 'plain_text',
'text': f'{status_emoji} 部署通知 - {deployment_info.get("service_name", "Unknown")}'
}
},
{
'type': 'section',
'fields': [
{
'type': 'mrkdwn',
'text': f'*环境:*\n{deployment_info.get("environment", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*版本:*\n{deployment_info.get("version", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*部署者:*\n{deployment_info.get("deployed_by", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*状态:*\n{status.capitalize()}'
},
{
'type': 'mrkdwn',
'text': f'*开始时间:*\n{deployment_info.get("start_time", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*结束时间:*\n{deployment_info.get("end_time", "Unknown")}'
}
]
},
{
'type': 'actions',
'elements': [
{
'type': 'button',
'text': {
'type': 'plain_text',
'text': '查看部署详情'
},
'url': deployment_info.get('deployment_url', '#')
},
{
'type': 'button',
'text': {
'type': 'plain_text',
'text': '查看服务'
},
'url': deployment_info.get('service_url', '#')
}
]
}
]
}
]
}
return message
def _build_alert_message(self, alert_info: Dict) -> Dict:
"""构建告警通知消息"""
severity = alert_info.get('severity', 'info')
severity_emoji = {
'critical': ':rotating_light:',
'warning': ':warning:',
'info': ':information_source:'
}.get(severity, ':information_source:')
severity_color = {
'critical': 'danger',
'warning': 'warning',
'info': 'good'
}.get(severity, '#cccccc')
message = {
'channel': self.webhooks.get('alerts', {}).get('channel', '#alerts'),
'username': self.webhooks.get('alerts', {}).get('username', 'Alert Bot'),
'icon_emoji': self.webhooks.get('alerts', {}).get('icon_emoji', ':rotating_light:'),
'attachments': [
{
'color': severity_color,
'blocks': [
{
'type': 'header',
'text': {
'type': 'plain_text',
'text': f'{severity_emoji} {alert_info.get("alert_name", "Unknown Alert")}'
}
},
{
'type': 'section',
'text': {
'type': 'mrkdwn',
'text': alert_info.get('description', 'No description provided')
}
},
{
'type': 'section',
'fields': [
{
'type': 'mrkdwn',
'text': f'*严重级别:*\n{severity.capitalize()}'
},
{
'type': 'mrkdwn',
'text': f'*触发时间:*\n{alert_info.get("timestamp", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*服务:*\n{alert_info.get("service", "Unknown")}'
},
{
'type': 'mrkdwn',
'text': f'*环境:*\n{alert_info.get("environment", "Unknown")}'
}
]
},
{
'type': 'actions',
'elements': [
{
'type': 'button',
'text': {
'type': 'plain_text',
'text': '查看告警详情'
},
'url': alert_info.get('alert_url', '#')
},
{
'type': 'button',
'text': {
'type': 'plain_text',
'text': '查看监控面板'
},
'url': alert_info.get('dashboard_url', '#')
}
]
}
]
}
]
}
return message
def _send_message(self, webhook_url: str, message: Dict) -> bool:
"""发送消息到Slack"""
try:
response = requests.post(webhook_url, json=message, timeout=30)
response.raise_for_status()
self.logger.info("Message sent to Slack successfully")
return True
except Exception as e:
self.logger.error(f"Failed to send message to Slack: {e}")
return False2. 交互式消息
利用Slack的交互功能提升用户体验：
按钮交互实现
#!/usr/bin/env python3
"""
Slack交互式消息处理工具
处理Slack按钮点击等交互事件
"""
import json
import hmac
import hashlib
import logging
from typing import Dict, Any
from flask import Flask, request, jsonify
class SlackInteractionHandler:
def __init__(self, signing_secret: str):
self.signing_secret = signing_secret
self.logger = logging.getLogger(__name__)
self.app = Flask(__name__)
self._setup_routes()
def _setup_routes(self):
"""设置路由"""
self.app.add_url_rule(
'/slack/interaction',
'slack_interaction',
self._handle_interaction,
methods=['POST']
)
def _handle_interaction(self):
"""处理交互事件"""
try:
# 验证请求签名
if not self._verify_signature(request):
return jsonify({'error': 'Invalid signature'}), 401
# 解析请求数据
payload = json.loads(request.form.get('payload', '{}'))
# 处理不同类型的交互
if payload.get('type') == 'block_actions':
return self._handle_block_actions(payload)
elif payload.get('type') == 'view_submission':
return self._handle_view_submission(payload)
else:
self.logger.warning(f"Unknown interaction type: {payload.get('type')}")
return jsonify({'status': 'ok'})
except Exception as e:
self.logger.error(f"Failed to handle Slack interaction: {e}")
return jsonify({'error': 'Internal server error'}), 500
def _verify_signature(self, req) -> bool:
"""验证请求签名"""
timestamp = req.headers.get('X-Slack-Request-Timestamp')
signature = req.headers.get('X-Slack-Signature')
body = req.get_data().decode('utf-8')
# 检查时间戳（防止重放攻击）
import time
if abs(time.time() - int(timestamp)) > 60 * 5: # 5分钟内有效
return False
# 构建签名字符串
sig_basestring = f"v0:{timestamp}:{body}"
my_signature = 'v0=' + hmac.new(
self.signing_secret.encode(),
sig_basestring.encode(),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(my_signature, signature)
def _handle_block_actions(self, payload: Dict) -> Any:
"""处理块操作"""
actions = payload.get('actions', [])
for action in actions:
action_id = action.get('action_id')
block_id = action.get('block_id')
if action_id == 'view_build_details':
return self._handle_view_build_details(action, payload)
elif action_id == 'retry_build':
return self._handle_retry_build(action, payload)
elif action_id == 'rollback_deployment':
return self._handle_rollback_deployment(action, payload)
else:
self.logger.warning(f"Unknown action ID: {action_id}")
return jsonify({'status': 'ok'})
def _handle_view_build_details(self, action: Dict, payload: Dict) -> Any:
"""处理查看构建详情操作"""
build_id = action.get('value')
user_id = payload.get('user', {}).get('id')
self.logger.info(f"User {user_id} requested build details for {build_id}")
# 这里可以实现具体的构建详情展示逻辑
# 例如：打开一个模态对话框显示详细信息
return jsonify({
'response_action': 'open_view',
'view': {
'type': 'modal',
'title': {
'type': 'plain_text',
'text': '构建详情'
},
'blocks': [
{
'type': 'section',
'text': {
'type': 'mrkdwn',
'text': f'正在加载构建 {build_id} 的详细信息...'
}
}
]
}
})
def _handle_retry_build(self, action: Dict, payload: Dict) -> Any:
"""处理重试构建操作"""
build_id = action.get('value')
user_id = payload.get('user', {}).get('id')
self.logger.info(f"User {user_id} requested to retry build {build_id}")
# 这里可以实现具体的重试构建逻辑
# 例如：调用CI/CD平台API重新触发构建
return jsonify({
'response_action': 'update',
'blocks': [
{
'type': 'section',
'text': {
'type': 'mrkdwn',
'text': f':hourglass_flowing_sand: 正在重试构建 {build_id}...'
}
}
]
})
def _handle_rollback_deployment(self, action: Dict, payload: Dict) -> Any:
"""处理回滚部署操作"""
deployment_id = action.get('value')
user_id = payload.get('user', {}).get('id')
self.logger.info(f"User {user_id} requested to rollback deployment {deployment_id}")
# 这里可以实现具体的回滚部署逻辑
# 例如：调用部署平台API触发回滚
return jsonify({
'response_action': 'update',
'blocks': [
{
'type': 'section',
'text': {
'type': 'mrkdwn',
'text': f':hourglass_flowing_sand: 正在回滚部署 {deployment_id}...'
}
}
]
})
def run(self, host: str = '0.0.0.0', port: int = 5000):
"""运行服务"""
self.app.run(host=host, port=port, debug=False)钉钉集成实践
针对国内用户，钉钉作为广泛使用的沟通工具，与其集成同样重要。
1. 机器人配置
钉钉支持通过自定义机器人接收外部通知：
机器人配置示例
# 钉钉集成配置
dingtalk_integration:
enabled: true
robots:
build_notifications:
webhook_url: "${DINGTALK_BUILD_WEBHOOK_URL}"
secret: "${DINGTALK_BUILD_SECRET}"
deployment_notifications:
webhook_url: "${DINGTALK_DEPLOYMENT_WEBHOOK_URL}"
secret: "${DINGTALK_DEPLOYMENT_SECRET}"
alerts:
webhook_url: "${DINGTALK_ALERT_WEBHOOK_URL}"
secret: "${DINGTALK_ALERT_SECRET}"消息发送工具
#!/usr/bin/env python3
"""
钉钉集成工具
实现与钉钉的深度集成
"""
import requests
import json
import hmac
import hashlib
import base64
import urllib.parse
import time
import logging
from typing import Dict, Any
class DingtalkIntegration:
def __init__(self, config: Dict):
self.config = config
self.robots = config.get('robots', {})
self.logger = logging.getLogger(__name__)
def _generate_sign(self, secret: str) -> tuple:
"""生成签名"""
if not secret:
return "", ""
timestamp = str(round(time.time() * 1000))
secret_enc = secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
return timestamp, sign
def send_message(self, robot_name: str, message: Dict[str, Any]) -> bool:
"""发送消息"""
try:
robot_config = self.robots.get(robot_name)
if not robot_config:
self.logger.error(f"Robot {robot_name} not configured")
return False
# 构建请求URL
webhook_url = robot_config['webhook_url']
secret = robot_config.get('secret')
if secret:
timestamp, sign = self._generate_sign(secret)
webhook_url = f"{webhook_url}×tamp={timestamp}&sign={sign}"
# 发送请求
response = requests.post(webhook_url, json=message, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('errcode') != 0:
self.logger.error(f"Failed to send Dingtalk message: {result}")
return False
self.logger.info(f"Message sent to Dingtalk robot {robot_name} successfully")
return True
except Exception as e:
self.logger.error(f"Failed to send Dingtalk message: {e}")
return False
def send_build_notification(self, build_info: Dict) -> bool:
"""发送构建通知"""
status = build_info.get('status', 'unknown')
status_emoji = {
'success': '✅',
'failed': '❌',
'running': '⏳',
'cancelled': '🚫'
}.get(status, '❓')
message = {
"msgtype": "markdown",
"markdown": {
"title": f"构建通知 - {build_info.get('project_name', 'Unknown')}",
"text": f"""## {status_emoji} 构建通知\n\n**项目名称**: {build_info.get('project_name', 'Unknown')}\n\n**构建ID**: {build_info.get('build_id', 'Unknown')}\n\n**状态**: {status.capitalize()}\n\n**触发者**: {build_info.get('triggered_by', 'Unknown')}\n\n**分支**: {build_info.get('branch', 'Unknown')}\n\n**开始时间**: {build_info.get('start_time', 'Unknown')}\n\n**结束时间**: {build_info.get('end_time', 'Unknown')}\n\n[查看构建详情]({build_info.get('build_url', '#')})"""
}
}
return self.send_message('build_notifications', message)
def send_deployment_notification(self, deployment_info: Dict) -> bool:
"""发送部署通知"""
status = deployment_info.get('status', 'unknown')
status_emoji = {
'success': '🚀',
'failed': '💥',
'running': '⏳',
'cancelled': '🚫'
}.get(status, '❓')
message = {
"msgtype": "markdown",
"markdown": {
"title": f"部署通知 - {deployment_info.get('service_name', 'Unknown')}",
"text": f"""## {status_emoji} 部署通知\n\n**服务名称**: {deployment_info.get('service_name', 'Unknown')}\n\n**环境**: {deployment_info.get('environment', 'Unknown')}\n\n**版本**: {deployment_info.get('version', 'Unknown')}\n\n**部署者**: {deployment_info.get('deployed_by', 'Unknown')}\n\n**状态**: {status.capitalize()}\n\n**开始时间**: {deployment_info.get('start_time', 'Unknown')}\n\n**结束时间**: {deployment_info.get('end_time', 'Unknown')}\n\n[查看部署详情]({deployment_info.get('deployment_url', '#')}) | [查看服务]({deployment_info.get('service_url', '#')})"""
}
}
return self.send_message('deployment_notifications', message)
def send_alert_notification(self, alert_info: Dict) -> bool:
"""发送告警通知"""
severity = alert_info.get('severity', 'info')
severity_emoji = {
'critical': '🚨',
'warning': '⚠️',
'info': 'ℹ️'
}.get(severity, 'ℹ️')
message = {
"msgtype": "markdown",
"markdown": {
"title": f"{severity_emoji} {alert_info.get('alert_name', 'Unknown Alert')}",
"text": f"""## {severity_emoji} {alert_info.get('alert_name', 'Unknown Alert')}\n\n{alert_info.get('description', 'No description provided')}\n\n**严重级别**: {severity.capitalize()}\n\n**触发时间**: {alert_info.get('timestamp', 'Unknown')}\n\n**服务**: {alert_info.get('service', 'Unknown')}\n\n**环境**: {alert_info.get('environment', 'Unknown')}\n\n[查看告警详情]({alert_info.get('alert_url', '#')}) | [查看监控面板]({alert_info.get('dashboard_url', '#')})"""
}
}
return self.send_message('alerts', message)2. 企业内部应用集成
对于更复杂的集成需求，可以开发钉钉企业内部应用：
应用配置
#!/usr/bin/env python3
"""
钉钉企业应用集成工具
实现与钉钉企业内部应用的深度集成
"""
import requests
import json
import logging
from typing import Dict, List, Optional
from datetime import datetime
class DingtalkAppIntegration:
def __init__(self, app_key: str, app_secret: str):
self.app_key = app_key
self.app_secret = app_secret
self.access_token = None
self.token_expires_at = 0
self.logger = logging.getLogger(__name__)
def _get_access_token(self) -> str:
"""获取访问令牌"""
# 检查令牌是否过期
if self.access_token and datetime.now().timestamp() < self.token_expires_at:
return self.access_token
try:
url = "https://oapi.dingtalk.com/gettoken"
params = {
'appkey': self.app_key,
'appsecret': self.app_secret
}
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('errcode') != 0:
raise Exception(f"Failed to get access token: {result}")
self.access_token = result['access_token']
# 令牌有效期通常为2小时，提前10分钟刷新
self.token_expires_at = datetime.now().timestamp() + result['expires_in'] - 600
return self.access_token
except Exception as e:
self.logger.error(f"Failed to get access token: {e}")
raise
def send_work_message(self, user_ids: List[str], message: Dict) -> bool:
"""发送工作通知消息"""
try:
access_token = self._get_access_token()
url = "https://oapi.dingtalk.com/topapi/message/corpconversation/asyncsend_v2"
params = {
'access_token': access_token
}
data = {
'userid_list': ','.join(user_ids),
'agent_id': self._get_agent_id(),
'msg': message
}
response = requests.post(url, params=params, json=data, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('errcode') != 0:
self.logger.error(f"Failed to send work message: {result}")
return False
self.logger.info("Work message sent successfully")
return True
except Exception as e:
self.logger.error(f"Failed to send work message: {e}")
return False
def _get_agent_id(self) -> str:
"""获取应用代理ID"""
# 这里应该从配置或通过API获取
return "123456789" # 示例ID
def create_chat_group(self, name: str, owner: str, user_ids: List[str]) -> Optional[str]:
"""创建群聊"""
try:
access_token = self._get_access_token()
url = "https://oapi.dingtalk.com/chat/create"
params = {
'access_token': access_token
}
data = {
'name': name,
'owner': owner,
'useridlist': user_ids
}
response = requests.post(url, params=params, json=data, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('errcode') != 0:
self.logger.error(f"Failed to create chat group: {result}")
return None
return result.get('chatid')
except Exception as e:
self.logger.error(f"Failed to create chat group: {e}")
return None
def send_chat_message(self, chat_id: str, message: Dict) -> bool:
"""发送群聊消息"""
try:
access_token = self._get_access_token()
url = "https://oapi.dingtalk.com/chat/send"
params = {
'access_token': access_token
}
data = {
'chatid': chat_id,
'msg': message
}
response = requests.post(url, params=params, json=data, timeout=30)
response.raise_for_status()
result = response.json()
if result.get('errcode') != 0:
self.logger.error(f"Failed to send chat message: {result}")
return False
self.logger.info("Chat message sent successfully")
return True
except Exception as e:
self.logger.error(f"Failed to send chat message: {e}")
return FalseMicrosoft Teams集成实践
Microsoft Teams作为企业级沟通平台，也支持丰富的集成能力。
1. Incoming Webhook配置
Teams支持通过Incoming Webhooks接收外部通知：
Webhook配置示例
# Microsoft Teams集成配置
teams_integration:
enabled: true
webhooks:
build_notifications:
url: "${TEAMS_BUILD_WEBHOOK_URL}"
deployment_notifications:
url: "${TEAMS_DEPLOYMENT_WEBHOOK_URL}"
alerts:
url: "${TEAMS_ALERT_WEBHOOK_URL}"消息发送工具
#!/usr/bin/env python3
"""
Microsoft Teams集成工具
实现与Microsoft Teams的深度集成
"""
import requests
import json
import logging
from typing import Dict, Any
class TeamsIntegration:
def __init__(self, config: Dict):
self.config = config
self.webhooks = config.get('webhooks', {})
self.logger = logging.getLogger(__name__)
def send_message(self, webhook_name: str, message: Dict[str, Any]) -> bool:
"""发送消息到Teams"""
try:
webhook_url = self.webhooks.get(webhook_name)
if not webhook_url:
self.logger.error(f"Webhook {webhook_name} not configured")
return False
response = requests.post(webhook_url, json=message, timeout=30)
response.raise_for_status()
self.logger.info(f"Message sent to Teams webhook {webhook_name} successfully")
return True
except Exception as e:
self.logger.error(f"Failed to send message to Teams: {e}")
return False
def send_build_notification(self, build_info: Dict) -> bool:
"""发送构建通知"""
status = build_info.get('status', 'unknown')
status_emoji = {
'success': '✅',
'failed': '❌',
'running': '⏳',
'cancelled': '🚫'
}.get(status, '❓')
message = {
"@type": "MessageCard",
"@context": "http://schema.org/extensions",
"themeColor": "0076D7",
"summary": f"Build {status} - {build_info.get('project_name', 'Unknown')}",
"sections": [
{
"activityTitle": f"{status_emoji} 构建通知",
"activitySubtitle": f"项目: {build_info.get('project_name', 'Unknown')}",
"facts": [
{
"name": "构建ID",
"value": build_info.get('build_id', 'Unknown')
},
{
"name": "状态",
"value": status.capitalize()
},
{
"name": "触发者",
"value": build_info.get('triggered_by', 'Unknown')
},
{
"name": "分支",
"value": build_info.get('branch', 'Unknown')
},
{
"name": "开始时间",
"value": build_info.get('start_time', 'Unknown')
},
{
"name": "结束时间",
"value": build_info.get('end_time', 'Unknown')
}
],
"markdown": True
}
],
"potentialAction": [
{
"@type": "OpenUri",
"name": "查看构建详情",
"targets": [
{
"os": "default",
"uri": build_info.get('build_url', '#')
}
]
}
]
}
return self.send_message('build_notifications', message)
def send_deployment_notification(self, deployment_info: Dict) -> bool:
"""发送部署通知"""
status = deployment_info.get('status', 'unknown')
status_color = {
'success': '28a745',
'failed': 'dc3545',
'running': 'ffc107',
'cancelled': '6c757d'
}.get(status, '0076D7')
status_emoji = {
'success': '🚀',
'failed': '💥',
'running': '⏳',
'cancelled': '🚫'
}.get(status, '❓')
message = {
"@type": "MessageCard",
"@context": "http://schema.org/extensions",
"themeColor": status_color,
"summary": f"Deployment {status} - {deployment_info.get('service_name', 'Unknown')}",
"sections": [
{
"activityTitle": f"{status_emoji} 部署通知",
"activitySubtitle": f"服务: {deployment_info.get('service_name', 'Unknown')}",
"facts": [
{
"name": "环境",
"value": deployment_info.get('environment', 'Unknown')
},
{
"name": "版本",
"value": deployment_info.get('version', 'Unknown')
},
{
"name": "部署者",
"value": deployment_info.get('deployed_by', 'Unknown')
},
{
"name": "状态",
"value": status.capitalize()
},
{
"name": "开始时间",
"value": deployment_info.get('start_time', 'Unknown')
},
{
"name": "结束时间",
"value": deployment_info.get('end_time', 'Unknown')
}
],
"markdown": True
}
],
"potentialAction": [
{
"@type": "OpenUri",
"name": "查看部署详情",
"targets": [
{
"os": "default",
"uri": deployment_info.get('deployment_url', '#')
}
]
},
{
"@type": "OpenUri",
"name": "查看服务",
"targets": [
{
"os": "default",
"uri": deployment_info.get('service_url', '#')
}
]
}
]
}
return self.send_message('deployment_notifications', message)通过与沟通工具的深度集成，CI/CD平台能够实现及时的状态通知和团队协作，提升团队的响应速度和工作效率。关键是要根据团队使用的具体沟通工具选择合适的集成方案，并设计丰富的消息格式和交互功能，以提供最佳的用户体验。