Metadata-Version: 2.4
Name: zy-dingtalk-chat-bot
Version: 1.1.0
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。
- `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 上；如果开启了关键词安全设置，消息内容必须包含对应关键词。

## 示例服务

仓库里的 `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` 等都不会进入发布包。
