基于WhatsApp Business Cloud API每日消息及媒体采集方案咨询

你的需求完全可以通过主动调用WhatsApp Business Cloud API的消息列表接口实现，无需长期运行Webhook服务，只需每日定时执行一次脚本即可完成当日消息和媒体的采集。

核心思路 #

WhatsApp Business Cloud API提供了GET /{phone-number-id}/messages接口，支持按时间范围过滤消息，你可以：

• 每日固定时间运行脚本，构造当日的时间范围参数
• 调用接口拉取该时间段内的所有消息
• 遍历消息，对包含媒体附件的消息，调用媒体下载接口获取并保存文件
• 用系统定时任务（如Linux Cron、Windows任务计划）自动执行脚本

完整实现代码 #

下面是基于Python的实现示例，包含消息拉取、媒体下载、分页处理：

import os
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv

# 加载环境变量（建议把敏感信息存在.env文件，不要硬编码）
load_dotenv()
ACCESS_TOKEN = os.getenv("WHATSAPP_ACCESS_TOKEN")
PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID")
MEDIA_SAVE_PATH = "./whatsapp_media"  # 媒体文件保存路径

# 创建媒体保存目录
os.makedirs(MEDIA_SAVE_PATH, exist_ok=True)

def get_today_time_range():
    # 获取当日UTC时间范围（WhatsApp API使用UTC时间）
    today_start = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
    today_end = datetime.utcnow()
    # 格式化为API要求的ISO 8601格式
    return today_start.isoformat() + "Z", today_end.isoformat() + "Z"

def fetch_messages(time_start, time_end):
    messages = []
    url = f"https://graph.facebook.com/v18.0/{PHONE_NUMBER_ID}/messages"
    params = {
        "access_token": ACCESS_TOKEN,
        "updated_time[gte]": time_start,
        "updated_time[lte]": time_end,
        "limit": 100  # 每次拉取的最大数量，可调整
    }

    while True:
        response = requests.get(url, params=params)
        response.raise_for_status()  # 抛出请求错误
        data = response.json()
        messages.extend(data.get("data", []))
        
        # 处理分页
        paging = data.get("paging")
        if not paging or "next" not in paging:
            break
        url = paging["next"]
    
    return messages

def download_media(media_id):
    # 获取媒体文件的下载链接
    media_url = f"https://graph.facebook.com/v18.0/{media_id}"
    params = {"access_token": ACCESS_TOKEN}
    response = requests.get(media_url, params=params)
    response.raise_for_status()
    media_data = response.json()
    
    # 下载媒体文件
    download_link = media_data["url"]
    headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
    media_response = requests.get(download_link, headers=headers)
    media_response.raise_for_status()
    
    # 保存文件，用media_id作为文件名，保留原扩展名
    file_ext = media_data["mime_type"].split("/")[-1]
    file_path = os.path.join(MEDIA_SAVE_PATH, f"{media_id}.{file_ext}")
    with open(file_path, "wb") as f:
        f.write(media_response.content)
    print(f"媒体文件已保存：{file_path}")

if __name__ == "__main__":
    # 获取当日时间范围
    time_start, time_end = get_today_time_range()
    print(f"开始拉取 {time_start} 到 {time_end} 的消息")
    
    # 拉取消息
    messages = fetch_messages(time_start, time_end)
    print(f"共拉取到 {len(messages)} 条消息")
    
    # 处理每条消息
    for msg in messages:
        print(f"消息ID: {msg['id']}, 发送方: {msg['from']}, 内容类型: {msg['type']}")
        
        # 如果是媒体类型消息，下载媒体
        if msg["type"] in ["image", "video", "audio", "document", "sticker"]:
            media_id = msg[msg["type"]]["id"]
            download_media(media_id)

关键说明 #

1. 环境变量配置：创建.env文件存储敏感信息，避免硬编码：

   WHATSAPP_ACCESS_TOKEN=你的API令牌
   WHATSAPP_PHONE_NUMBER_ID=你的电话号码ID
   

2. 时间范围处理：WhatsApp API使用UTC时间，脚本中用UTC时间构造当日范围，避免时区偏差
3. 分页处理：消息数量较多时接口会分页返回，需循环拉取所有页面的消息
4. 媒体下载：媒体文件需先通过媒体ID获取下载链接，再携带Authorization头下载
5. 定时执行：
   • Linux系统：用Cron设置每日运行，例如每天凌晨1点执行：

     0 1 * * * /usr/bin/python3 /path/to/your/script.py >> /path/to/logfile.log 2>&1
     

   • Windows系统：使用「任务计划程序」创建每日任务执行Python脚本

权限要求 #

确保你的WhatsApp Business应用已获得whatsapp_business_messaging权限，且API令牌具备read_messages和read_media权限

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