81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""打包完成后的钉钉通知。"""
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import time
|
||
from urllib.parse import quote
|
||
|
||
import httpx
|
||
|
||
|
||
class NotificationError(Exception):
|
||
"""通知发送失败。"""
|
||
|
||
|
||
def _signed_webhook_url(webhook_url: str, secret: str) -> str:
|
||
"""为启用了加签的钉钉机器人附加 timestamp 与 sign 参数。"""
|
||
if not secret:
|
||
return webhook_url
|
||
|
||
timestamp = str(round(time.time() * 1000))
|
||
signature = hmac.new(
|
||
secret.encode("utf-8"),
|
||
f"{timestamp}\n{secret}".encode("utf-8"),
|
||
hashlib.sha256,
|
||
).digest()
|
||
separator = "&" if "?" in webhook_url else "?"
|
||
return f"{webhook_url}{separator}timestamp={timestamp}&sign={quote(base64.b64encode(signature))}"
|
||
|
||
|
||
def build_dingtalk_payload(config_data: dict, download_url: str, qr_code_url: str = "") -> dict:
|
||
"""生成与 AutoPacking/upload_iap.py 一致的钉钉 Markdown 内容。"""
|
||
details = (
|
||
f"**环境：** {config_data.get('SERVER', '')}\n\n"
|
||
f"**版本：** {config_data.get('VERSION', '')}\n\n"
|
||
f"**APP名称：** {config_data.get('APPID_NAME', '')}\n\n"
|
||
f"**包名：** {config_data.get('BUNDLE_ID', '')}\n\n"
|
||
f"**App Guid：** {config_data.get('APPID', '')}"
|
||
)
|
||
text = f"## 【iOS】打包信息\n\n{details}\n\n**iOS 下载链接:** {download_url}\n"
|
||
if qr_code_url:
|
||
text += f"\n"
|
||
|
||
return {
|
||
"msgtype": "markdown",
|
||
"markdown": {"title": "iOS应用下载", "text": text},
|
||
"at": {"atMobiles": [], "isAtAll": False},
|
||
}
|
||
|
||
|
||
def send_dingtalk_notification(
|
||
dingtalk_config: dict,
|
||
config_data: dict,
|
||
download_url: str,
|
||
qr_code_url: str = "",
|
||
) -> bool:
|
||
"""发送钉钉通知；未启用通知时不发起网络请求。"""
|
||
if not dingtalk_config.get("enabled"):
|
||
return False
|
||
|
||
webhook_url = dingtalk_config.get("webhook_url", "").strip()
|
||
if not webhook_url:
|
||
raise NotificationError("钉钉通知已启用，但未配置 Webhook URL")
|
||
|
||
payload = build_dingtalk_payload(config_data, download_url, qr_code_url)
|
||
payload["at"]["atMobiles"] = dingtalk_config.get("at_mobiles", [])
|
||
response = httpx.post(
|
||
_signed_webhook_url(webhook_url, dingtalk_config.get("secret", "")),
|
||
json=payload,
|
||
timeout=15,
|
||
)
|
||
if response.status_code // 100 != 2:
|
||
raise NotificationError(f"钉钉通知发送失败 ({response.status_code})")
|
||
|
||
try:
|
||
result = response.json()
|
||
except ValueError as exc:
|
||
raise NotificationError("钉钉通知返回内容无效") from exc
|
||
if result.get("errcode", 0) != 0:
|
||
raise NotificationError(f"钉钉通知发送失败: {result.get('errmsg', '未知错误')}")
|
||
return True
|