shenleiandClaude Sonnet 5
dae14a8a10
钉钉机器人启用了自定义关键词校验。9ace2c8 将 App_Store 标题改为 「【iOS】App_Store 包信息」，丢掉了「打包」二字，导致消息被钉钉拒收 （errmsg: 关键词不匹配），任务日志出现「钉钉通知未发送」。 标题改回「【iOS】App_Store 打包信息」，重新包含关键词；测试同步更新。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014FETBaqMgRqx3kdpzccKg1
113 lines
4.0 KiB
Python
113 lines
4.0 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 _normalize_mobiles(at_mobiles) -> list:
|
||
"""把手机号配置统一成去重、去空的字符串列表，兼容列表或逗号/换行分隔的字符串。"""
|
||
if not at_mobiles:
|
||
return []
|
||
if isinstance(at_mobiles, str):
|
||
raw = at_mobiles.replace("\n", ",").replace("，", ",").replace(" ", ",").split(",")
|
||
else:
|
||
raw = at_mobiles
|
||
seen = []
|
||
for item in raw:
|
||
mobile = str(item).strip()
|
||
if mobile and mobile not in seen:
|
||
seen.append(mobile)
|
||
return seen
|
||
|
||
|
||
def build_dingtalk_payload(
|
||
config_data: dict,
|
||
download_url: str,
|
||
qr_code_url: str = "",
|
||
at_mobiles=None,
|
||
) -> dict:
|
||
"""生成与 AutoPacking/upload_iap.py 一致的钉钉 Markdown 内容。"""
|
||
mobiles = _normalize_mobiles(at_mobiles)
|
||
# App_Store 包与 Ad Hoc 用不同标题，App_Store 只有 IPA 下载地址、无二维码。
|
||
# 标题必须保留「打包」二字：钉钉机器人启用了自定义关键词，缺少关键词会被拒收。
|
||
heading = "【iOS】App_Store 打包信息" if config_data.get("BUILD_TYPE") == "App_Store" else "【iOS】打包信息"
|
||
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"## {heading}\n\n{details}\n\n**iOS 下载链接:** {download_url}\n"
|
||
if qr_code_url:
|
||
text += f"\n"
|
||
# 钉钉 markdown 消息必须在正文内出现 @手机号 才会真正 @ 到人，
|
||
# 仅靠 at.atMobiles 不生效。
|
||
if mobiles:
|
||
text += "\n\n" + " ".join(f"@{m}" for m in mobiles)
|
||
|
||
return {
|
||
"msgtype": "markdown",
|
||
"markdown": {"title": "iOS应用下载", "text": text},
|
||
"at": {"atMobiles": mobiles, "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")
|
||
|
||
# Ad_Hoc 与 App_Store 使用各自的 @ 名单。
|
||
mobiles_key = "at_mobiles_app_store" if config_data.get("BUILD_TYPE") == "App_Store" else "at_mobiles"
|
||
payload = build_dingtalk_payload(
|
||
config_data, download_url, qr_code_url, dingtalk_config.get(mobiles_key, [])
|
||
)
|
||
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
|