THENG
/
a-cloud-all
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

添加webhook

This commit is contained in:
孙小云 2026-01-27 14:17:44 +08:00
parent e037e0e11c
commit 6af0532185
3 changed files with 316 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.devops/config.yaml
View File

@ -22,6 +22,12 @@ logging:
file: .devops/logs/devops.log
max_size: 10485760 # 10MB
# 钉钉通知配置
dingtalk:
enabled: true # 是否启用钉钉通知
access_token: ed3533e05cf13e090c098436ee6cd52b2adfa2d85b5b2b9da1ae2bccdaecb8f3
secret: SEC66372694e16e7e931f53aefb4b847b7fb6c42350a10f0f27fbf4151785353261
# 基础设施服务配置（只部署一次）
infrastructure:
- name: ruoyi-mysql

68
.devops/monitor.py
View File

@ -20,6 +20,7 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from scripts.log import Logger
from scripts import docker, maven, npm
from scripts.init import mysql, redis, nacos
from scripts.dingtalk import DingTalkNotifier
class GitMonitor:
@ -33,11 +34,13 @@ class GitMonitor:
self.global_branch = 'main'
self.project_root = None
self.runtime_path = None
self.dingtalk_notifier = None
# 初始化
self._print_startup_banner()
self._load_config()
self._init_paths()
self._init_dingtalk()
def _print_startup_banner(self):
"""打印启动横幅"""
@ -89,6 +92,24 @@ class GitMonitor:
Logger.error(f"路径初始化失败: {e}")
sys.exit(1)
def _init_dingtalk(self):
"""初始化钉钉通知器"""
try:
dingtalk_config = self.config.get('dingtalk', {})
if dingtalk_config.get('enabled', False):
access_token = dingtalk_config.get('access_token')
secret = dingtalk_config.get('secret')
if access_token:
self.dingtalk_notifier = DingTalkNotifier(access_token, secret)
Logger.info("✓ 钉钉通知已启用")
else:
Logger.warning("钉钉通知已启用但未配置 access_token")
else:
Logger.info("钉钉通知未启用")
except Exception as e:
Logger.warning(f"钉钉通知初始化失败: {e}")
def get_remote_commit(self, repo_url, branch):
"""获取远程仓库的最新提交 hash"""
try:
@ -263,11 +284,22 @@ class GitMonitor:
def deploy(self, repo_config):
"""执行部署流程"""
repo_path = self.runtime_path / 'a-cloud-all'
repo_name = repo_config['name']
commit_hash = self.last_commits.get(repo_name, 'unknown')
start_time = time.time()
Logger.separator()
Logger.info(f"开始部署: {repo_config['name']}")
Logger.info(f"开始部署: {repo_name}")
Logger.separator()
# 发送构建开始通知
if self.dingtalk_notifier:
self.dingtalk_notifier.send_build_start(
repo_name=repo_name,
branch=self.global_branch,
commit_hash=commit_hash
)
try:
# 1. 更新主仓库和子模块
if not self.update_main_repo():
@ -303,12 +335,46 @@ class GitMonitor:
service_name = repo_config['docker_service']
if not docker.run_docker_compose(compose_dir, service_name):
# 发送构建失败通知
if self.dingtalk_notifier:
duration = time.time() - start_time
self.dingtalk_notifier.send_build_failure(
repo_name=repo_name,
branch=self.global_branch,
commit_hash=commit_hash,
error_msg="Docker 部署失败"
)
return False
# 计算构建耗时
duration = time.time() - start_time
# 发送构建成功通知
if self.dingtalk_notifier:
self.dingtalk_notifier.send_build_success(
repo_name=repo_name,
branch=self.global_branch,
commit_hash=commit_hash,
duration=duration
)
Logger.info(f"部署完成: {repo_config['name']}")
return True
except Exception as e:
# 计算构建耗时
duration = time.time() - start_time
# 发送构建失败通知
if self.dingtalk_notifier:
self.dingtalk_notifier.send_build_failure(
repo_name=repo_name,
branch=self.global_branch,
commit_hash=commit_hash,
error_msg=str(e),
at_all=True
)
Logger.error(f"部署异常: {e}")
return False

243
.devops/scripts/dingtalk.py Normal file
View File

@ -0,0 +1,243 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
钉钉 Webhook 通知模块
用于发送构建和部署通知到钉钉群
"""
import time
import hmac
import hashlib
import base64
import urllib.parse
import urllib.request
import json
from datetime import datetime
class DingTalkNotifier:
"""钉钉通知器"""
def __init__(self, access_token, secret=None):
"""
初始化钉钉通知器
Args:
access_token: 钉钉机器人的 access_token
secret: 钉钉机器人的加签密钥（可选）
"""
self.access_token = access_token
self.secret = secret
self.base_url = "https://oapi.dingtalk.com/robot/send"
def _generate_sign(self, timestamp):
"""
生成钉钉加签
Args:
timestamp: 当前时间戳（毫秒）
Returns:
签名字符串
"""
if not self.secret:
return None
string_to_sign = f'{timestamp}\n{self.secret}'
string_to_sign_enc = string_to_sign.encode('utf-8')
secret_enc = self.secret.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 sign
def _build_url(self):
"""
构建完整的 webhook URL
Returns:
完整的 URL 字符串
"""
url = f"{self.base_url}?access_token={self.access_token}"
if self.secret:
timestamp = str(round(time.time() * 1000))
sign = self._generate_sign(timestamp)
url += f"&timestamp={timestamp}&sign={sign}"
return url
def send_text(self, content, at_mobiles=None, at_all=False):
"""
发送文本消息
Args:
content: 消息内容
at_mobiles: @的手机号列表
at_all: 是否@所有人
Returns:
发送结果 (True/False)
"""
data = {
"msgtype": "text",
"text": {
"content": content
},
"at": {
"atMobiles": at_mobiles or [],
"isAtAll": at_all
}
}
return self._send_request(data)
def send_build_success(self, repo_name, branch, commit_hash, duration):
"""
发送构建成功通知
Args:
repo_name: 仓库名称
branch: 分支名称
commit_hash: 提交哈希
duration: 构建耗时（秒）
Returns:
发送结果 (True/False)
"""
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
title = f"✅ 构建成功 - {repo_name}"
text = f"""### ✅ 构建成功
**仓库**: {repo_name}
**分支**: {branch}
**提交**: {commit_hash[:8]}
**耗时**: {duration:.1f} 秒
**时间**: {now}
---
构建和部署已完成！
"""
return self.send_markdown(title, text)
def send_build_failure(self, repo_name, branch, commit_hash, error_msg, at_all=False):
"""
发送构建失败通知
Args:
repo_name: 仓库名称
branch: 分支名称
commit_hash: 提交哈希
error_msg: 错误信息
at_all: 是否@所有人
Returns:
发送结果 (True/False)
"""
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
title = f"❌ 构建失败 - {repo_name}"
# 限制错误信息长度
error_display = error_msg[:500] if len(error_msg) > 500 else error_msg
text = f"""### ❌ 构建失败
**仓库**: {repo_name}
**分支**: {branch}
**提交**: {commit_hash[:8]}
**时间**: {now}
**错误信息**:
```
{error_display}
```
---
请检查日志并修复问题！
"""
return self.send_markdown(title, text, at_all=at_all)
def send_build_start(self, repo_name, branch, commit_hash):
"""
发送构建开始通知
Args:
repo_name: 仓库名称
branch: 分支名称
commit_hash: 提交哈希
Returns:
发送结果 (True/False)
"""
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
title = f"🚀 构建开始 - {repo_name}"
text = f"""### 🚀 构建开始
**仓库**: {repo_name}
**分支**: {branch}
**提交**: {commit_hash[:8]}
**时间**: {now}
---
构建任务已启动，请稍候...
"""
return self.send_markdown(title, text)
def _send_request(self, data):
"""
发送 HTTP 请求到钉钉 webhook
Args:
data: 请求数据字典
Returns:
发送结果 (True/False)
"""
try:
url = self._build_url()
headers = {'Content-Type': 'application/json'}
json_data = json.dumps(data).encode('utf-8')
req = urllib.request.Request(url, data=json_data, headers=headers)
response = urllib.request.urlopen(req, timeout=10)
result = json.loads(response.read().decode('utf-8'))
if result.get('errcode') == 0:
return True
else:
print(f"钉钉通知发送失败: {result.get('errmsg')}")
return False
except Exception as e:
print(f"钉钉通知发送异常: {e}")
return False
def send_markdown(self, title, text, at_mobiles=None, at_all=False):
"""
发送 Markdown 消息
Args:
title: 消息标题
text: Markdown 格式的消息内容
at_mobiles: @的手机号列表
at_all: 是否@所有人
Returns:
发送结果 (True/False)
"""
data = {
"msgtype": "markdown",
"markdown": {
"title": title,
"text": text
},
"at": {
"atMobiles": at_mobiles or [],
"isAtAll": at_all
}
}
return self._send_request(data)