fix: 上传二维码并发送 Ad Hoc 通知

backend/services/notification.py

@@ -0,0 +1,80 @@
+"""打包完成后的钉钉通知。"""
+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![image]({qr_code_url})"
+
+    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