这个需求可以拆成三步：先安全地拿到待处理的邮件列表，再调用 GenMail 的摘要生成能力，最后把摘要 POST 到群聊 Webhook。由于不同邮件服务和群聊机器人的接口存在差异，下面给出的脚本骨架使用占位端点和通用 REST 风格，你需要根据实际接口替换域名、路径、请求头和参数。
本方案适用于具备 REST API 的邮件服务和支持自定义机器人的群聊工具。操作上先拉取未读邮件列表，再调用摘要接口生成短文本，最后通过 Webhook 推送到群聊。验证方式是通过脚本日志查看每步请求的 URL、状态码和响应结构。风险在于接口鉴权、限流和消息格式因服务而异，需要先结合环境确认。
搭建读取邮件列表的请求骨架
获取邮件列表是整条链路的第一步。邮件服务一般提供 REST API，通常需要 OAuth2 或 Bearer Token 鉴权。为了不把凭据写死在代码里，建议通过环境变量读取。下面是用 requests 库实现的分页拉取骨架：
import os
import requests
BASE_URL = os.getenv("MAIL_API_BASE_URL", "https://mail.example.com/api/v1")
TOKEN = os.getenv("MAIL_API_TOKEN", "replace-with-real-token")
headers = {
"Authorization": f"Bearer {TOKEN}",
"Accept": "application/json"
}
def fetch_unread_emails(page_size=20):
params = {
"query": "is:unread",
"limit": page_size,
"page_token": ""
}
emails = []
while True:
resp = requests.get(
f"{BASE_URL}/emails",
headers=headers,
params=params,
timeout=10
)
resp.raise_for_status()
data = resp.json()
emails.extend(data.get("items", []))
next_token = data.get("next_page_token")
if not next_token:
break
params["page_token"] = next_token
return emails这段代码先按未读条件请求第一页，再根据返回的 next_page_token 继续翻页，直到拉完所有未读邮件。这里的 base_url、分页参数名和未读过滤字段都需要替换成实际邮件 API 的类型。如果邮件服务只支持日期范围或文件夹 ID，就把 params 里的 query 改为对应字段。运行前先单独测试 fetch_unread_emails，确认返回的 items 里包含邮件 ID 和正文或正文链接。
调用摘要生成接口并处理响应
拿到邮件列表后，把邮件 ID 或正文文本传给摘要接口。GenMail 这类服务的接口通常接收一个文本列表，返回每封邮件的 summary。发送时建议加超时和重试，避免网络抖动导致任务中断。下面是一个 POST 请求示例：
def generate_summary(email_ids, email_contents):
summarize_url = os.getenv("SUMMARY_API_URL", "https://genmail.example.com/v1/summarize")
payload = {
"email_ids": email_ids,
"contents": email_contents,
"max_length": 120
}
for attempt in range(3):
try:
resp = requests.post(
summarize_url,
json=payload,
headers=headers,
timeout=30
)
resp.raise_for_status()
data = resp.json()
return data.get("summaries", [])
except (requests.Timeout, requests.RequestException) as e:
if attempt == 2:
raise e
time.sleep(2 * (attempt + 1))注意几个细节：responses.json() 里若是包含 summaries 列表，就按 email_id 和 summary 字段组装；如果摘要接口直接返回 Markdown 或纯文本，需要先解析再拿内容。超时时间要根据邮件正文大小调整，正文很大时 30 秒可能不够。重试逻辑里也建议记录 attempt 次数，方便日志审计。
构造群聊 Webhook 发送程序
群聊推送目前最常见的三种方式：Slack Incoming Webhook、钉钉自定义机器人、企业微信机器人。它们的请求体结构不同，但核心都是 POST JSON。下面封装了一个通用函数，用 webhook_type 区分格式：
def send_webhook(webhook_url, webhook_type, title, text):
if webhook_type == "slack":
payload = {"text": f"*{title}*\n{text}"}
elif webhook_type == "dingtalk":
payload = {"msgtype": "text", "text": {"content": f"{title}\n{text}"}}
elif webhook_type == "wecom":
payload = {"msgtype": "text", "text": {"content": f"{title}\n{text}"}}
else:
raise ValueError(f"Unsupported webhook_type: {webhook_type}")
resp = requests.post(webhook_url, json=payload, timeout=10)
resp.raise_for_status()
return resp.json()消息模板建议包含邮件发件人、主题、摘要和原始链接。钉钉和企业微信机器人通常需要加关键字或签名，如果推送不成功，先检查安全设置是否匹配。Slack 可以直接显示 Markdown，但钉钉和企业微信只支持部分格式，所以模板里统一用普通文本加换行。
运行端到端测试与日志审计
把以上函数串成完整流程后，需要确保每一步都有日志。下面是一个测试主函数，打印每个阶段的关键信息：
import logging
logging.basicConfig(level=logging.INFO)
def main():
emails = fetch_unread_emails()
logging.info("Fetched %d emails", len(emails))
email_ids = [e["id"] for e in emails]
contents = [e.get("snippet") or e.get("body") for e in emails]
summaries = generate_summary(email_ids, contents)
logging.info("Summaries received: %s", len(summaries))
for item in summaries:
email_id = item["email_id"]
summary = item["summary"]
title = f"邮件摘要 {email_id}"
send_webhook(os.getenv("GROUP_WEBHOOK_URL"), "dingtalk", title, summary)
logging.info("Sent summary for %s to webhook", email_id)
if __name__ == "__main__":
main()日志里要记录请求的 URL、状态码和响应摘要。可以在每个请求后加一条类似 logging.info("POST %s -> %s", url, resp.status_code) 的输出，但上面为了简洁只写了阶段日志。建议在开发时先用一个测试群聊或临时 webhook 跑一次，确认消息能送达。常见异常包括：401 鉴权失败、404 路径错误、429 限流、以及 JSON 解析错误。碰到 429 时退避重试，碰到 401 时检查 token 是否过期。最后，整个脚本可以挂到 cron 或 CI 定时任务中，但要注意邮件服务只读取不标记已读，否则会重复收到同一批邮件。