能否在exchangelib中实现推送通知？轮询触发函数是否可行？

当然可以！exchangelib完全支持Exchange Web Services (EWS) 提供的推送通知机制——不用再靠反复轮询收件箱这种低效方式触发自定义函数，主动推送的模式会让你的流程更高效、实时性更强。

exchangelib支持的三种订阅模式 #

exchangelib封装了EWS的三类事件订阅方案，其中推送订阅就是你要找的主动通知模式：

• 推送订阅：当目标文件夹（比如收件箱）发生指定事件（新邮件、邮件修改等）时，Exchange服务器会主动把通知推送到你指定的HTTP端点
• 拉取订阅：由你的代码主动向Exchange服务器查询是否有新事件，适合无法暴露公网端点的场景
• 流式订阅：保持一个长连接，服务器实时把事件推送给你，延迟极低，无需公网端点

实现推送订阅的完整示例 #

要实现推送订阅，你需要两个核心部分：一个能被Exchange服务器访问的HTTP端点（用来接收通知），以及用exchangelib创建订阅的代码。

第一步：搭建接收通知的HTTP服务 #

这里用Flask做个简单示例，你也可以用FastAPI、Django等其他框架：

from flask import Flask, request
app = Flask(__name__)

@app.route('/ews-notify', methods=['POST'])
def handle_ews_notification():
    # 解析Exchange发送的通知内容
    notification_content = request.data.decode('utf-8')
    # 触发你的自定义业务逻辑，比如处理新邮件
    process_new_mail_event(notification_content)
    # 必须返回200状态码，否则Exchange会重复推送
    return '', 200

def process_new_mail_event(notification):
    # 这里写你的自定义逻辑：比如解析通知、获取邮件详情、触发告警等
    print(f"收到新邮件通知:\n{notification}")

if __name__ == '__main__':
    # 注意：Exchange服务器需要能访问这个地址，本地开发可使用内网穿透工具（如ngrok）
    app.run(host='0.0.0.0', port=8000)

第二步：用exchangelib创建推送订阅 #

from exchangelib import Account, Credentials, PushSubscription

# 初始化Exchange账户
creds = Credentials(username='your_work_email@company.com', password='your_password')
account = Account(
    primary_smtp_address='your_work_email@company.com',
    credentials=creds,
    autodiscover=True,
    access_type='delegate'
)

# 选择要监控的文件夹（这里是收件箱）
target_folder = account.inbox

# 创建推送订阅
push_sub = PushSubscription.subscribe(
    account=account,
    folders=[target_folder],
    callback_url='http://your-public-ip:8000/ews-notify',  # 替换成你的服务公网地址
    event_types=['NewMail', 'Modified']  # 按需选择要监控的事件类型
)

print(f"推送订阅已创建，ID: {push_sub.id}")

# 后续如果需要取消订阅，执行：
# push_sub.delete()

关键注意事项 #

• 端点可访问性：Exchange服务器必须能访问你的回调URL，本地开发时可以用内网穿透工具（比如ngrok）把本地服务暴露到公网
• 订阅有效期：EWS推送订阅默认有有效期（一般30分钟到1小时），你需要定期调用push_sub.renew()来续订，避免订阅失效
• 事件类型：除了NewMail，还可以选择Created（新建项）、Modified（修改项）、Deleted（删除项）等，根据业务需求调整
• 错误处理：如果你的端点返回非200状态码，Exchange会重试推送，所以要保证服务稳定

替代方案：流式订阅 #

如果搭建公网端点有困难，流式订阅是个很好的替代——它不需要外部可访问的URL，只需要保持长连接就能实时接收事件：

from exchangelib import StreamingSubscription

stream_sub = StreamingSubscription.subscribe(
    account=account,
    folders=[target_folder],
    event_types=['NewMail']
)

# 启动监听，保持进程运行
for event in stream_sub.stream():
    print(f"收到实时事件: {event}")
    process_new_mail_event(event)

这种方式适合不需要对外暴露服务的场景，缺点是需要保持代码进程持续运行。

内容的提问来源于stack exchange，提问作者onuryartasi