Blame for src/dingtalk.py

"""
钉钉自定义机器人：加签并 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)