Spaces:
Runtime error
Runtime error
| """ | |
| 钉钉自定义机器人：加签并 POST 发送 Markdown 消息。 | |
| """ | |
| import base64 | |
| import hashlib | |
| import hmac | |
| import os | |
| import time | |
| import urllib.parse | |
| from pathlib import Path | |
| import requests | |
| from dotenv import load_dotenv | |
| # 从项目根目录加载 .env | |
| _project_root = Path(__file__).resolve().parent.parent | |
| load_dotenv(_project_root / ".env") | |
| 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(base64.b64encode(hmac_code)) | |
| return timestamp, sign | |
| def build_signed_url(webhook: str, secret: str) -> str: | |
| """在 Webhook URL 后追加 timestamp 与 sign 参数。""" | |
| timestamp, sign = _sign(secret) | |
| sep = "&" if "?" in webhook else "?" | |
| return f"{webhook}{sep}timestamp={timestamp}&sign={sign}" | |
| def send_markdown(title: str, text: str) -> tuple[bool, str]: | |
| """ | |
| 发送钉钉 Markdown 消息。 | |
| 返回 (成功与否, 错误信息)。 | |
| """ | |
| webhook = os.environ.get("DINGTALK_WEBHOOK", "").strip() | |
| secret = os.environ.get("DINGTALK_SECRET", "").strip() | |
| if not webhook: | |
| return False, "未配置 DINGTALK_WEBHOOK" | |
| url = build_signed_url(webhook, secret) if secret else webhook | |
| body = { | |
| "msgtype": "markdown", | |
| "markdown": { | |
| "title": title, | |
| "text": text, | |
| }, | |
| } | |
| try: | |
| r = requests.post(url, json=body, headers={"Content-Type": "application/json"}, timeout=10) | |
| data = r.json() | |
| if r.status_code != 200 or data.get("errcode") != 0: | |
| return False, data.get("errmsg", r.text) or str(r.status_code) | |
| return True, "" | |
| except Exception as e: | |
| return False, str(e) | |