在Python中使用Webhook发送数据

在 Python 中使用 Webhook 发送数据，本质上是发送一个 HTTP POST 请求。最常用的库是 requests，因为它简洁且功能强大。

以下是针对不同场景的实现方式：

1. 使用 requests 发送 JSON 数据 (最常用)

这是最标准的 Webhook 发送方式，适用于对接 Slack、钉钉、企业微信或自定义 API。

import requests
import json

# Webhook 地址
webhook_url = "https://your-webhook-endpoint.com/api"

# 要发送的数据
data = {
    "event": "new_user",
    "user_id": 12345,
    "message": "Hello, this is a webhook notification!"
}

try:
    # 发送 POST 请求
    # json=data 会自动设置 Content-Type 为 application/json
    response = requests.post(webhook_url, json=data, timeout=10)

    # 检查状态码
    if response.status_code == 200:
        print("Webhook 发送成功!")
    else:
        print(f"发送失败，状态码: {response.status_code}, 响应: {response.text}")

except requests.exceptions.RequestException as e:
    print(f"请求发生异常: {e}")

2. 发送带验证信息的 Webhook

如果你的 Webhook 需要 API Key 或 Token 验证：

headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}

response = requests.post(webhook_url, data=json.dumps(data), headers=headers)

3. 常见办公软件 Webhook 格式示例

不同的平台对数据结构有特定要求。

钉钉 (DingTalk)

data = {
    "msgtype": "text",
    "text": {"content": "监控报警：服务器 CPU 占用过高"}
}

企业微信 (WeChat Work)

data = {
    "msgtype": "markdown",
    "markdown": {"content": "## 运行报告\n > 状态：<font color=\"info\">正常</font>"}
}

4. 进阶：异步发送 Webhook (httpx)

如果你的爬虫或程序需要高并发发送 Webhook，而不希望阻塞主线程，可以使用 httpx：

import asyncio
import httpx

async def send_webhook():
    async with httpx.AsyncClient() as client:
        data = {"status": "finished"}
        response = await client.post("https://your-webhook.com", json=data)
        print(response.status_code)

asyncio.run(send_webhook())

5. 调试工具推荐

在发送 Webhook 时，如果一直报错，建议使用以下工具查看你发出的数据包到底长什么样：

• Webhook.site: 它可以生成一个临时的 URL，你往这个地址发请求，它会实时显示收到的所有 Header 和 Body，非常适合排查格式问题。
• Postman: 用于模拟发送请求。

总结：Webhook 与 Selenium 结合

在你的爬虫脚本中，可以将 Webhook 放在 try...finally 块中，用于实时通知爬虫状态：

try:
    # ... Selenium 爬取逻辑 ...
    msg = "爬虫任务运行成功"
except Exception as e:
    msg = f"爬虫任务失败，错误原因: {str(e)}"
finally:
    requests.post(webhook_url, json={"summary": msg})
    driver.quit()