307 lines
9.4 KiB
Python
307 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
飞书机器人通知器
|
||
发送代码质量扫描报告到飞书
|
||
"""
|
||
import json
|
||
import time
|
||
import hashlib
|
||
import hmac
|
||
import base64
|
||
import logging
|
||
import requests
|
||
from typing import Dict, Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class FeishuNotifier:
|
||
"""飞书机器人通知器"""
|
||
|
||
def __init__(self, config: Dict[str, Any]):
|
||
"""
|
||
初始化飞书通知器
|
||
|
||
Args:
|
||
config: 飞书配置
|
||
"""
|
||
self.config = config
|
||
self.webhook_url = config.get('webhook_url', '')
|
||
self.secret = config.get('secret', '')
|
||
|
||
if not self.webhook_url:
|
||
logger.warning('飞书 Webhook URL 未配置')
|
||
|
||
def send_report(self, report: Dict[str, Any]) -> bool:
|
||
"""
|
||
发送扫描报告到飞书
|
||
|
||
Args:
|
||
report: 报告数据
|
||
|
||
Returns:
|
||
是否发送成功
|
||
"""
|
||
if not self.webhook_url:
|
||
logger.error('飞书 Webhook URL 未配置')
|
||
return False
|
||
|
||
try:
|
||
# 构建消息内容
|
||
message = self._build_message(report)
|
||
|
||
# 如果配置了签名，则使用签名验证
|
||
if self.secret:
|
||
timestamp, sign = self._generate_sign()
|
||
payload = {
|
||
"timestamp": timestamp,
|
||
"sign": sign,
|
||
"msg_type": "interactive",
|
||
"card": message
|
||
}
|
||
else:
|
||
payload = {
|
||
"msg_type": "interactive",
|
||
"card": message
|
||
}
|
||
|
||
# 发送请求
|
||
headers = {'Content-Type': 'application/json'}
|
||
response = requests.post(
|
||
self.webhook_url,
|
||
headers=headers,
|
||
data=json.dumps(payload).encode('utf-8'),
|
||
timeout=30
|
||
)
|
||
|
||
# 解析响应
|
||
result = response.json()
|
||
|
||
if result.get('code') == 0:
|
||
logger.info('飞书消息发送成功')
|
||
return True
|
||
else:
|
||
logger.error(f'飞书消息发送失败: {result.get("msg")}')
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f'发送飞书通知失败: {str(e)}', exc_info=True)
|
||
return False
|
||
|
||
def _generate_sign(self) -> tuple:
|
||
"""
|
||
生成飞书签名
|
||
|
||
Returns:
|
||
(timestamp, sign) 元组
|
||
"""
|
||
# 当前时间戳（秒）
|
||
timestamp = str(int(time.time()))
|
||
|
||
# 拼接字符串
|
||
string_to_sign = '{}\n{}'.format(timestamp, self.secret)
|
||
|
||
# 使用 HmacSHA256 计算签名
|
||
hmac_code = hmac.new(
|
||
string_to_sign.encode('utf-8'),
|
||
digestmod=hashlib.sha256
|
||
).digest()
|
||
|
||
# 进行 Base64 编码
|
||
sign = base64.b64encode(hmac_code).decode('utf-8')
|
||
|
||
return timestamp, sign
|
||
|
||
def _build_message(self, report: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
构建飞书卡片消息
|
||
|
||
Args:
|
||
report: 报告数据
|
||
|
||
Returns:
|
||
飞书卡片消息结构
|
||
"""
|
||
# 根据状态选择颜色
|
||
status = report.get('status', 'pass')
|
||
if status == 'pass':
|
||
theme_color = 'green'
|
||
status_icon = '✅'
|
||
elif status == 'fail':
|
||
theme_color = 'red'
|
||
status_icon = '❌'
|
||
else:
|
||
theme_color = 'orange'
|
||
status_icon = '⚠️'
|
||
|
||
# 构建问题摘要
|
||
total_issues = report.get('total_issues', 0)
|
||
total_errors = report.get('total_errors', 0)
|
||
total_warnings = report.get('total_warnings', 0)
|
||
|
||
# 获取扫描结果详情
|
||
scan_details = []
|
||
for scanner_name, result in report.get('scan_results', {}).items():
|
||
tool_name = result.get('tool', scanner_name)
|
||
summary = result.get('summary', {})
|
||
files_scanned = result.get('files_scanned', 0)
|
||
total = summary.get('total', 0)
|
||
|
||
if total > 0:
|
||
detail_text = f"{tool_name}: 扫描 {files_scanned} 个文件，发现 {total} 个问题"
|
||
else:
|
||
detail_text = f"{tool_name}: 扫描 {files_scanned} 个文件，无问题"
|
||
|
||
scan_details.append(detail_text)
|
||
|
||
# 检查是否为 PR 扫描
|
||
pr_url = report.get('pr_url')
|
||
target_branch = report.get('target_branch')
|
||
|
||
# 构建基本信息文本
|
||
if pr_url and target_branch:
|
||
# PR 扫描
|
||
title = f"{status_icon} PR 代码质量扫描报告"
|
||
basic_info = (f"**仓库:** `{report.get('repo_name', 'unknown')}`\n"
|
||
f"**源分支:** `{report.get('branch', 'unknown')}` → **目标分支:** `{target_branch}`\n"
|
||
f"**PR链接:** [查看PR]({pr_url})\n"
|
||
f"**提交:** `{report.get('commit_id', 'unknown')}`\n"
|
||
f"**提交者:** {report.get('author', 'unknown')}")
|
||
else:
|
||
# Push 扫描
|
||
title = f"{status_icon} 代码质量扫描报告"
|
||
basic_info = (f"**仓库:** `{report.get('repo_name', 'unknown')}`\n"
|
||
f"**分支:** `{report.get('branch', 'unknown')}`\n"
|
||
f"**提交:** `{report.get('commit_id', 'unknown')}`\n"
|
||
f"**提交者:** {report.get('author', 'unknown')}")
|
||
|
||
# 构建卡片消息
|
||
card = {
|
||
"header": {
|
||
"title": {
|
||
"tag": "plain_text",
|
||
"content": title
|
||
},
|
||
"template": theme_color
|
||
},
|
||
"elements": [
|
||
{
|
||
"tag": "div",
|
||
"text": {
|
||
"tag": "lark_md",
|
||
"content": basic_info
|
||
}
|
||
},
|
||
{
|
||
"tag": "div",
|
||
"text": {
|
||
"tag": "lark_md",
|
||
"content": f"**扫描状态:** {report.get('status_text', 'unknown')}\n"
|
||
f"📊 总问题: {total_issues} | "
|
||
f"🔴 错误: {total_errors} | "
|
||
f"🟡 警告: {total_warnings}"
|
||
}
|
||
}
|
||
]
|
||
}
|
||
|
||
# 添加扫描详情
|
||
if scan_details:
|
||
card["elements"].append({
|
||
"tag": "div",
|
||
"text": {
|
||
"tag": "lark_md",
|
||
"content": "**扫描详情:**\n" + "\n".join([f"- {d}" for d in scan_details])
|
||
}
|
||
})
|
||
|
||
# 添加主要问题列表（最多显示5个）
|
||
all_issues = []
|
||
for scanner_name, result in report.get('scan_results', {}).items():
|
||
for issue in result.get('issues', [])[:3]: # 每个扫描器最多显示3个
|
||
all_issues.append(issue)
|
||
|
||
if all_issues:
|
||
issues_text = "**主要问题:**\n"
|
||
for i, issue in enumerate(all_issues[:5], 1):
|
||
severity = issue.get('severity', 'Unknown')
|
||
severity_emoji = {
|
||
'HIGH': '🔴',
|
||
'MEDIUM': '🟡',
|
||
'LOW': '🔵',
|
||
'ERROR': '🔴',
|
||
'WARNING': '🟡'
|
||
}.get(severity.upper(), '⚪')
|
||
|
||
file_path = issue.get('file', 'unknown')
|
||
line_num = issue.get('line', 0)
|
||
message = issue.get('message', 'No message')[:50]
|
||
|
||
issues_text += f"{i}. {severity_emoji} `{file_path}:{line_num}` - {message}\n"
|
||
|
||
card["elements"].append({
|
||
"tag": "div",
|
||
"text": {
|
||
"tag": "lark_md",
|
||
"content": issues_text
|
||
}
|
||
})
|
||
|
||
# 添加时间戳
|
||
card["elements"].append({
|
||
"tag": "div",
|
||
"text": {
|
||
"tag": "lark_md",
|
||
"content": f"🕐 扫描时间: {report.get('timestamp', '')}"
|
||
}
|
||
})
|
||
|
||
return card
|
||
|
||
def send_simple_message(self, title: str, content: str) -> bool:
|
||
"""
|
||
发送简单文本消息
|
||
|
||
Args:
|
||
title: 标题
|
||
content: 内容
|
||
|
||
Returns:
|
||
是否发送成功
|
||
"""
|
||
if not self.webhook_url:
|
||
logger.error('飞书 Webhook URL 未配置')
|
||
return False
|
||
|
||
try:
|
||
# 构建消息
|
||
payload = {
|
||
"msg_type": "text",
|
||
"content": {
|
||
"text": f"{title}\n{content}"
|
||
}
|
||
}
|
||
|
||
# 发送请求
|
||
headers = {'Content-Type': 'application/json'}
|
||
response = requests.post(
|
||
self.webhook_url,
|
||
headers=headers,
|
||
data=json.dumps(payload).encode('utf-8'),
|
||
timeout=30
|
||
)
|
||
|
||
result = response.json()
|
||
|
||
if result.get('code') == 0:
|
||
logger.info('飞书消息发送成功')
|
||
return True
|
||
else:
|
||
logger.error(f'飞书消息发送失败: {result.get("msg")}')
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f'发送飞书通知失败: {str(e)}')
|
||
return False
|