289 lines
8.7 KiB
Python
289 lines
8.7 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)
|
|||
|
|
|
|||
|
|
# 构建卡片消息
|
|||
|
|
card = {
|
|||
|
|
"header": {
|
|||
|
|
"title": {
|
|||
|
|
"tag": "plain_text",
|
|||
|
|
"content": f"{status_icon} 代码质量扫描报告"
|
|||
|
|
},
|
|||
|
|
"template": theme_color
|
|||
|
|
},
|
|||
|
|
"elements": [
|
|||
|
|
{
|
|||
|
|
"tag": "div",
|
|||
|
|
"text": {
|
|||
|
|
"tag": "lark_md",
|
|||
|
|
"content": 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')}"
|
|||
|
|
}
|
|||
|
|
},
|
|||
|
|
{
|
|||
|
|
"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
|