forked from erp-dev/erp
117 lines
3.2 KiB
Python
117 lines
3.2 KiB
Python
"""
|
||
WeCom (企业微信) robot webhook helpers.
|
||
|
||
Only text/markdown messages are supported for now.
|
||
|
||
API:
|
||
POST https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx
|
||
Content-Type: application/json
|
||
{
|
||
"msgtype": "text",
|
||
"text": {"content": "hello world"}
|
||
}
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.parse import urlencode
|
||
from urllib.request import Request, urlopen
|
||
|
||
from django.conf import settings
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class WeComWebhookResponse:
|
||
errcode: int
|
||
errmsg: str
|
||
raw: dict[str, Any]
|
||
|
||
@property
|
||
def ok(self) -> bool:
|
||
return int(self.errcode) == 0
|
||
|
||
|
||
def send_wecom_webhook_message(
|
||
*,
|
||
content: str,
|
||
msgtype: str = "text",
|
||
key: str | None = None,
|
||
timeout_seconds: float = 10.0,
|
||
) -> WeComWebhookResponse:
|
||
"""
|
||
Send a message to WeCom robot webhook.
|
||
|
||
Supported msgtype:
|
||
- text: {"msgtype":"text","text":{"content": "..."}}
|
||
- markdown: {"msgtype":"markdown","markdown":{"content": "..."}}
|
||
"""
|
||
t = (msgtype or "text").strip().lower()
|
||
if t not in {"text", "markdown"}:
|
||
raise ValueError("msgtype 仅支持 text 或 markdown")
|
||
|
||
content = (content or "").strip()
|
||
if not content:
|
||
raise ValueError("content 不能为空")
|
||
|
||
base_url = getattr(settings, "WECOM_WEBHOOK_BASE_URL", "").strip()
|
||
if not base_url:
|
||
base_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
|
||
|
||
k = (key or getattr(settings, "WECOM_WEBHOOK_KEY", "") or "").strip()
|
||
if not k:
|
||
raise ValueError("WeCom webhook key 未配置（settings.WECOM_WEBHOOK_KEY）")
|
||
|
||
url = f"{base_url}?{urlencode({'key': k})}"
|
||
if t == "markdown":
|
||
payload = {"msgtype": "markdown", "markdown": {"content": content}}
|
||
else:
|
||
payload = {"msgtype": "text", "text": {"content": content}}
|
||
|
||
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
|
||
req = Request(
|
||
url=url,
|
||
data=data,
|
||
headers={"Content-Type": "application/json"},
|
||
method="POST",
|
||
)
|
||
|
||
try:
|
||
with urlopen(req, timeout=float(timeout_seconds)) as resp:
|
||
body = resp.read().decode("utf-8", errors="replace")
|
||
except HTTPError as exc:
|
||
body = ""
|
||
try:
|
||
body = exc.read().decode("utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
raise RuntimeError(f"WeCom webhook HTTPError: status={exc.code}, body={body}") from exc
|
||
except URLError as exc:
|
||
raise RuntimeError(f"WeCom webhook URLError: {exc}") from exc
|
||
|
||
try:
|
||
raw = json.loads(body) if body else {}
|
||
except Exception as exc:
|
||
raise RuntimeError(f"WeCom webhook 响应不是合法 JSON: {body}") from exc
|
||
|
||
errcode = int(raw.get("errcode") or 0)
|
||
errmsg = str(raw.get("errmsg") or "")
|
||
return WeComWebhookResponse(errcode=errcode, errmsg=errmsg, raw=raw)
|
||
|
||
|
||
def send_wecom_webhook_text(
|
||
*,
|
||
content: str,
|
||
key: str | None = None,
|
||
timeout_seconds: float = 10.0,
|
||
) -> WeComWebhookResponse:
|
||
"""
|
||
Backward-compatible wrapper for sending text.
|
||
"""
|
||
return send_wecom_webhook_message(content=content, msgtype="text", key=key, timeout_seconds=timeout_seconds)
|
||
|