Metadata-Version: 2.4
Name: zy-dingtalk-chat-bot
Version: 1.1.1
Summary: 钉钉机器人工具包，封装 HTTP 回调签名校验、回复体构造、单聊、群聊及固定 Webhook 主动推送。
Author: ZY
License: ISC
Project-URL: Homepage, https://github.com/
Keywords: dingtalk,robot,bot,chatbot
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: ISC License (ISCL)
Classifier: Operating System :: OS Independent
Classifier: Topic :: Communications :: Chat
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: requests>=2.28
Provides-Extra: server
Requires-Dist: flask>=2.3; extra == "server"
Requires-Dist: python-dotenv>=1.0; extra == "server"

# zy-dingtalk-chat-bot (Python)

钉钉机器人工具包，封装 HTTP 回调签名校验、同步回复消息体、单聊发送、群聊发送，以及固定群机器人 Webhook 主动推送。

这是 [npm 包 dingtalk-chat-bot](https://www.npmjs.com/package/dingtalk-chat-bot) 的 Python 版本，API 风格保持一致。

## 支持范围

这个包当前支持 **HTTP 回调方式**，不支持 Stream 长连接方式。

钉钉机器人常见有两类接入：

- HTTP 回调：钉钉把用户消息 POST 到你的公网服务。
- Stream 长连接：你的程序主动连接钉钉并保持长连接。

本包适合这些场景：

- 用户私聊机器人后，自动回复或异步发送单聊消息。
- 群里 @ 机器人后，回复当前群聊。
- 监控告警、定时日报等任务，通过固定群机器人 Webhook 主动推送到群。

本包暂不包含：

- Stream 长连接收消息。
- 业务指令解析。
- 数据库、任务队列、权限系统。

## 安装

```bash
pip install zy-dingtalk-chat-bot
```

如果你需要本地运行示例 Flask 服务：

```bash
pip install "zy-dingtalk-chat-bot[server]"
```

## 创建实例

```python
import os
from dingtalk_chat_bot import create_dingtalk_bot

bot = create_dingtalk_bot(
    app_key=os.environ["DINGTALK_APP_KEY"],
    app_secret=os.environ["DINGTALK_APP_SECRET"],
    robot_code=os.environ.get("DINGTALK_ROBOT_CODE"),
)
```

配置说明：

- `app_key`：钉钉应用的 AppKey。发送单聊消息时需要。
- `app_secret`：钉钉应用的 AppSecret。校验 HTTP 回调签名、获取 accessToken 时需要。
- `robot_code`：机器人编码。通常可以填 AppKey；如果钉钉后台单独展示 robotCode，就填 robotCode。
- `webhook_secret`：可选，自定义机器人「加签」安全策略的密钥（SEC 开头）。设置后 `send_webhook_message` 会自动按官方算法拼 `timestamp` 和 `sign`。详见下方「自定义机器人加签」一节。
- `http_client`：可选，自定义 `requests.Session`。
- `signature_expires_in`：可选，HTTP 回调签名时间窗口，单位毫秒，默认 1 小时。
- `request_timeout`：可选，HTTP 请求超时时间，单位秒，默认 10。

如果你只使用固定群机器人 Webhook 推送，可以不传 `app_key`、`app_secret`：

```python
bot = create_dingtalk_bot()

bot.send_webhook_message(os.environ["DINGTALK_WEBHOOK_URL"], "服务正常")
```

## 环境变量

仓库提供了 `.env.example` 作为模板。本地运行示例服务时，可以自己创建 `.env`：

```bash
cp .env.example .env
```

示例：

```env
PORT=3000
DINGTALK_APP_KEY=
DINGTALK_APP_SECRET=
DINGTALK_ROBOT_CODE=
DINGTALK_WEBHOOK_URL=
```

`.env` 不应该提交到 Git，也不会进入发布包。

## HTTP 回调签名校验

当钉钉把消息 POST 到你的服务时，可以这样校验签名（以 Flask 为例）：

```python
from flask import Flask, jsonify, request

app = Flask(__name__)

@app.post("/dingtalk/robot")
def dingtalk_robot():
    ok = bot.verify_signature(
        request.headers.get("timestamp"),
        request.headers.get("sign"),
    )
    if not ok:
        return jsonify({"error": "invalid dingtalk signature"}), 401

    return jsonify(bot.build_reply_payload("收到", request.get_json()))
```

## 同步回复

`build_reply_payload` 用来构造可直接返回给钉钉 HTTP 回调的消息体。

普通文本：

```python
return jsonify(bot.build_reply_payload("你好，我收到了", request.get_json()))
```

Markdown：

```python
return jsonify(bot.build_reply_payload({
    "msgtype": "markdown",
    "title": "处理结果",
    "content": "## 处理结果\n\n- 状态：成功\n- 耗时：120ms",
}, request.get_json()))
```

如果是群聊回调，并且消息体里有 `senderStaffId`，回复会默认在第一行真实 @ 提问人。

## 自动发送单聊或群聊

`send_message` 适合在收到钉钉回调后使用。它会根据回调消息体里的 `conversationType` 自动判断发送方式：

- `conversationType == "1"`：通过 OpenAPI 发送单聊消息。
- `conversationType == "2"`：通过回调里的 `sessionWebhook` 发送群聊消息。

```python
bot.send_message(body, "这条消息会自动发到当前单聊或群聊")
```

群聊发送时默认会 @ 本次提问人。如果不想 @：

```python
bot.send_message(body, "这条群消息不艾特任何人", {"atSender": False})
```

发送 Markdown：

```python
bot.send_message(body, {
    "msgtype": "markdown",
    "title": "日报",
    "content": "## 今日日报\n\n- 完成机器人回复\n- 支持单聊和群聊",
})
```

## 主动发送单聊

```python
bot.send_private_message("用户 staffId", "你好")
```

多个用户：

```python
bot.send_private_message(["staffId1", "staffId2"], "批量单聊消息")
```

Markdown 单聊：

```python
bot.send_private_message("用户 staffId", {
    "msgtype": "markdown",
    "title": "通知",
    "content": "## 通知\n\n请查看最新处理结果。",
})
```

## 通过 sessionWebhook 发群聊

`sessionWebhook` 来自钉钉 HTTP 回调消息体，适合在用户 @ 机器人之后回复当前群聊。

```python
bot.send_group_message(body["sessionWebhook"], "群聊回复")
```

指定 @ 用户：

```python
bot.send_group_message(body["sessionWebhook"], "请关注这条消息", {
    "atUserIds": ["staffId1"],
})
```

Markdown 群聊：

```python
bot.send_group_message(body["sessionWebhook"], {
    "msgtype": "markdown",
    "title": "群聊通知",
    "content": "## 群聊通知\n\n- 已处理完成",
})
```

## 固定群机器人 Webhook 推送

如果你在群机器人设置里复制到了固定 Webhook：

```text
https://oapi.dingtalk.com/robot/send?access_token=xxx
```

可以使用 `send_webhook_message` 主动推送消息，适合监控告警、定时日报、定时巡检等场景，不需要等用户先私聊或 @ 机器人。

普通文本：

```python
bot.send_webhook_message(
    os.environ["DINGTALK_WEBHOOK_URL"],
    "监控告警：订单 API 响应超时",
)
```

Markdown：

```python
bot.send_webhook_message(os.environ["DINGTALK_WEBHOOK_URL"], {
    "msgtype": "markdown",
    "title": "监控告警",
    "content": "## 监控告警\n\n- 服务：订单 API\n- 状态：响应超时\n- 请及时处理",
})
```

@ 指定用户：

```python
bot.send_webhook_message(
    os.environ["DINGTALK_WEBHOOK_URL"],
    "请关注这条告警",
    {"atUserIds": ["staffId1"]},
)
```

@ 所有人：

```python
bot.send_webhook_message(
    os.environ["DINGTALK_WEBHOOK_URL"],
    "重要告警，请所有人关注",
    {"isAtAll": True},
)
```

如果开启了关键词安全设置，消息内容必须包含对应关键词，否则会被钉钉拒绝。

## 自定义机器人加签

如果群机器人安全设置勾选了「加签」，调用 webhook 时必须在 URL 上拼 `timestamp` 和 `sign`，本包已内置处理。

构造时传入 `webhook_secret`，后续所有 `send_webhook_message` 调用都会自动加签：

```python
bot = create_dingtalk_bot(webhook_secret=os.environ["DINGTALK_ROBOT_SECRET"])
bot.send_webhook_message(os.environ["DINGTALK_WEBHOOK_URL"], "已加签")
```

也可以在单次调用时通过 `secret` 参数覆盖（优先级高于构造时的 `webhook_secret`）：

```python
bot.send_webhook_message(webhook_url, "临时加签", secret="SECxxxxxx")
```

如果只想拿到签好的 URL 自己用，可以调用静态方法：

```python
from dingtalk_chat_bot import DingTalkBot

signed = DingTalkBot.sign_webhook_url(webhook_url, "SECxxxxxx")
```

`webhook_secret` / `secret` 留空时不加签，方便同一份代码同时兼容启用 / 未启用加签的机器人。

> `send_group_message` 走的是钉钉 HTTP 回调里的 `sessionWebhook`（临时地址，自带 token），不需要也不应该加签。

## 示例消息与文本路由 (`dingtalk_chat_bot.examples`)

开箱即用的菜单 / Markdown / 时间回复 / 默认路由，免去自己再写一份样板代码。

```python
from dingtalk_chat_bot.examples import (
    build_menu,            # → markdown 菜单
    build_markdown_demo,   # → markdown 示例
    build_text_demo,       # → 文本示例
    build_time_reply,      # → 当前服务器时间
    handle_user_message,   # → 默认文本路由
)

bot.send_webhook_message(WEBHOOK_URL, build_menu())
```

`handle_user_message(text)` 把用户文本映射到合适的示例回复，可直接接到 HTTP 回调里：

| 用户发送                          | 回复内容                |
| --------------------------------- | ----------------------- |
| `菜单` / `help` / `/help` / `帮助` | 菜单 markdown           |
| `文本` / `text`                    | 文本示例                |
| `markdown` / `md` / `示例markdown` | Markdown 示例           |
| `时间` / `time`                    | 当前服务器时间          |
| 空字符串                            | 提示重新发送            |
| 其他                                | 回声 + 菜单引导         |

接入示例：

```python
from dingtalk_chat_bot.examples import handle_user_message

@app.post("/dingtalk/robot")
def robot():
    body = request.get_json(silent=True) or {}
    text = (body.get("text") or {}).get("content", "").strip()
    return jsonify(bot.build_reply_payload(handle_user_message(text), body))
```

## 示例服务

仓库里的 `server.py` 是一个 Flask 示例，不会进入发布包。你可以本地运行它测试 HTTP 回调：

```bash
pip install "zy-dingtalk-chat-bot[server]"
cp .env.example .env
python server.py
```

默认接口：

- `POST /dingtalk/robot`：同步回复示例。
- `POST /dingtalk/robot-async`：先同步确认，再异步发送消息示例。
- `GET /health`：健康检查。

## 发布说明

`pyproject.toml` 的 `[tool.setuptools]` 字段只显式包含 `dingtalk_chat_bot` 包目录。因此 `.env`、`server.py`、`requirements-dev.txt`、`.venv` 等都不会进入发布包。
