如何用Python配置推送通知监听器接收Google Drive文件变更

作为编程新手，别慌，咱们一步步来把这个推送通知的事儿捋清楚——你已经搞定了Sheets和Drive API的配置，这就成功了一半！下面是具体的实现步骤，都是实操性的内容：

一、先搞懂Drive推送通知的核心逻辑 #

Drive的变更推送是Webhook机制：你告诉Google一个可公开访问的URL，当指定的表格文件发生变更时，Google会主动给这个URL发送请求；你在这个URL对应的服务里接收通知，再去拉取表格的最新数据做处理。

⚠️ 注意：如果是本地开发，你需要用工具把本地端口暴露到公网（比如ngrok），不然Google没法访问你的本地服务。

二、步骤1：搭建接收推送的Web服务（用Flask快速实现） #

咱们用Flask写一个简单的Web服务，用来接收Google的推送通知，同时完成订阅验证：

首先安装Flask：

pip install flask

然后写基础服务代码：

from flask import Flask, request, jsonify

app = Flask(__name__)

# 这个路由用来处理Google的订阅验证和变更通知
@app.route('/drive-webhook', methods=['GET', 'POST'])
def drive_webhook():
    # 处理Google的订阅验证请求（GET方法）
    if request.method == 'GET':
        # Google会发一个带hub.challenge参数的GET请求，你需要原样返回这个参数
        challenge = request.args.get('hub.challenge')
        return challenge, 200
    
    # 处理实际的变更通知（POST方法）
    elif request.method == 'POST':
        # 先打印收到的通知内容，方便调试
        notification = request.json
        print("收到表格变更通知：", notification)
        
        # 这里可以调用Sheets API拉取最新数据，执行你的业务操作
        # 比如后面要写的sheets_get_data()函数
        
        return jsonify({"status": "success"}), 200

if __name__ == '__main__':
    # 启动服务，端口选个自己喜欢的，比如5000
    app.run(port=5000, debug=True)

三、步骤2：用ngrok把本地服务暴露到公网 #

启动上面的Flask服务后，打开新终端运行ngrok（如果没装，先去官网下载安装）：

ngrok http 5000

运行后会得到一个类似https://xxxx-xx-xx-xx-xx.ngrok.io的公网URL，把这个地址记下来，后面订阅推送要用。

四、步骤3：向Drive API订阅指定表格的变更 #

现在需要调用Drive API的files/watch接口，订阅你要监控的表格。确保你已经有授权凭证（比如service account的密钥文件），然后写订阅代码：

from googleapiclient.discovery import build
from google.oauth2.service_account import Credentials

# 加载你的凭证文件（比如service-account-key.json）
creds = Credentials.from_service_account_file('service-account-key.json')
drive_service = build('drive', 'v3', credentials=creds)

# 要监控的表格文件ID（从表格URL里提取：https://docs.google.com/spreadsheets/d/[FILE_ID]/edit）
TARGET_SHEET_ID = "你的表格ID"
# 刚才ngrok的公网URL加上路由
WEBHOOK_URL = "https://xxxx-xx-xx-xx-xx.ngrok.io/drive-webhook"

# 构造订阅请求体
subscription = {
    "id": "my-sheet-subscription-001",  # 随便写一个唯一ID，用来标识这个订阅
    "type": "web_hook",
    "address": WEBHOOK_URL,
    "payload": True  # 设为True会推送详细变更信息，False只通知有变更发生
}

# 发送订阅请求
response = drive_service.files().watch(fileId=TARGET_SHEET_ID, body=subscription).execute()
print("订阅成功！返回信息：", response)

⚠️ 注意：订阅有有效期（默认约1周），需要定期调用files/watch刷新订阅，避免失效。

五、步骤4：处理通知，拉取最新表格数据 #

当收到推送通知后，调用Sheets API拉取最新数据。写一个简单的拉取函数：

def sheets_get_data(spreadsheet_id):
    sheets_service = build('sheets', 'v4', credentials=creds)
    # 拉取指定范围的数据，比如Sheet1的A1到Z100
    result = sheets_service.spreadsheets().values().get(
        spreadsheetId=spreadsheet_id, range="Sheet1!A1:Z100"
    ).execute()
    values = result.get('values', [])
    
    if not values:
        print("表格里没有数据")
    else:
        print("最新表格数据：")
        for row in values:
            print(row)
    return values

然后把这个函数加到Webhook的POST处理逻辑里：

elif request.method == 'POST':
    notification = request.json
    print("收到表格变更通知：", notification)
    # 拉取并处理最新数据
    sheets_get_data(TARGET_SHEET_ID)
    return jsonify({"status": "success"}), 200

六、关键注意事项 #

• 权限验证：确保你的凭证账号有目标表格的只读权限（你已经配置过，但可以再去Google Drive里确认下权限设置）。
• 重复通知：Google可能会重复发送通知，所以你的处理逻辑最好是幂等的（重复执行也不会出问题）。
• ngrok限制：免费版ngrok每次重启URL会变，测试完成后可以考虑部署到云服务器，用固定域名接收通知。
• 订阅刷新：记得写定时任务定期刷新订阅，避免通知中断。

这样一步步走下来，你就能实现自动监控表格变更并处理数据啦！如果遇到具体报错（比如权限不足、Webhook收不到通知），再针对性排查即可。

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