fix: 认证系统、登录兼容性、运行时配置、钉钉通知及代码清理
- 实现 JWT 认证中间件,保护配置管理和任务删除接口 - 修复 ConfigView 登录按钮(Vue 3 inject 替代 $root) - 页面刷新时通过 /api/auth/me 校验 token 有效性 - max_concurrent_builds 修改后运行时即时生效 - 实现钉钉 webhook 通知(构建成功/失败自动推送) - 删除未使用的 useWebSocket composable - log_streamer 使用 LOG_QUEUE_MAX_SIZE 配置常量
This commit is contained in:
@@ -0,0 +1,93 @@
|
||||
"""钉钉通知服务"""
|
||||
import asyncio
|
||||
import hashlib
|
||||
import hmac
|
||||
import base64
|
||||
import time
|
||||
import urllib.parse
|
||||
import json
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _sign(secret: str) -> tuple[str, str]:
|
||||
"""生成钉钉加签参数"""
|
||||
timestamp = str(round(time.time() * 1000))
|
||||
string_to_sign = f"{timestamp}\n{secret}"
|
||||
hmac_code = hmac.new(
|
||||
secret.encode("utf-8"),
|
||||
string_to_sign.encode("utf-8"),
|
||||
digestmod=hashlib.sha256,
|
||||
).digest()
|
||||
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
|
||||
return timestamp, sign
|
||||
|
||||
|
||||
async def send_dingtalk(webhook_url: str, title: str, text: str, secret: str = ""):
|
||||
"""发送钉钉 Markdown 通知"""
|
||||
import httpx
|
||||
|
||||
url = webhook_url
|
||||
if secret:
|
||||
timestamp, sign = _sign(secret)
|
||||
url = f"{url}×tamp={timestamp}&sign={sign}"
|
||||
|
||||
payload = {
|
||||
"msgtype": "markdown",
|
||||
"markdown": {"title": title, "text": text},
|
||||
}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
resp = await client.post(url, json=payload)
|
||||
data = resp.json()
|
||||
if data.get("errcode") != 0:
|
||||
logger.warning(f"钉钉通知发送失败: {data}")
|
||||
except Exception as e:
|
||||
logger.warning(f"钉钉通知异常: {e}")
|
||||
|
||||
|
||||
async def notify_build_result(task, dingtalk_config: dict):
|
||||
"""发送打包结果通知"""
|
||||
if not dingtalk_config.get("enabled") or not dingtalk_config.get("webhook_url"):
|
||||
return
|
||||
|
||||
status = task.status
|
||||
app_name = task.app_name
|
||||
build_type = task.build_type
|
||||
scheme_name = task.scheme_name
|
||||
|
||||
if status == "completed":
|
||||
title = f"✅ {app_name} 打包成功"
|
||||
lines = [
|
||||
f"### ✅ {app_name} 打包成功",
|
||||
f"- **应用**: {app_name}",
|
||||
f"- **类型**: {build_type}",
|
||||
f"- **Scheme**: {scheme_name}",
|
||||
f"- **分支**: {task.branch}",
|
||||
]
|
||||
if task.oss_url:
|
||||
lines.append(f"- **下载链接**: [点击下载]({task.oss_url})")
|
||||
text = "\n".join(lines)
|
||||
elif status == "failed":
|
||||
title = f"❌ {app_name} 打包失败"
|
||||
error = task.error_message or "未知错误"
|
||||
lines = [
|
||||
f"### ❌ {app_name} 打包失败",
|
||||
f"- **应用**: {app_name}",
|
||||
f"- **类型**: {build_type}",
|
||||
f"- **Scheme**: {scheme_name}",
|
||||
f"- **分支**: {task.branch}",
|
||||
f"- **错误**: {error}",
|
||||
]
|
||||
text = "\n".join(lines)
|
||||
else:
|
||||
return
|
||||
|
||||
await send_dingtalk(
|
||||
dingtalk_config["webhook_url"],
|
||||
title,
|
||||
text,
|
||||
dingtalk_config.get("secret", ""),
|
||||
)
|
||||
Reference in New Issue
Block a user