如何用Python脚本检测Facebook Messenger新消息并存储内容

准备工作 #

• 安装依赖包：

  pip install fbchat python-dotenv
  

• 新建.env文件存储账号信息（避免硬编码泄露隐私）：

  FB_EMAIL=你的Facebook登录邮箱
  FB_PASSWORD=你的Facebook登录密码
  

核心代码实现 #

from fbchat import Client
from fbchat.models import Message
from dotenv import load_dotenv
import os
import time
import datetime

# 加载环境变量
load_dotenv()

# 自定义客户端类，继承fbchat的Client以处理消息回调
class MessengerClient(Client):
    def __init__(self, email, password, **kwargs):
        super().__init__(email, password, **kwargs)
        self.last_message_id = 0  # 记录最后处理过的消息ID，避免重复处理
        self.contact_messages = {}  # 按联系人ID存储未合并的消息：{联系人ID: [消息1, 消息2...]}
        self.last_merge_time = {}  # 记录每个联系人上次合并消息的时间

    def onMessage(self, mid, author_id, message_object, thread_id, thread_type, **kwargs):
        # 跳过自己发送的消息
        if author_id == self.uid:
            return
        
        # 只处理未处理过的新消息
        if int(mid) > self.last_message_id:
            self.last_message_id = int(mid)
            current_time = datetime.datetime.now()

            # 把消息添加到对应联系人的列表中
            if author_id not in self.contact_messages:
                self.contact_messages[author_id] = []
                self.last_merge_time[author_id] = current_time
            self.contact_messages[author_id].append(message_object.text)

            # 自定义合并条件：满足任一条件就合并
            # 条件1：同一联系人消息数≥3条；条件2：距离上次合并超过10秒
            message_count = len(self.contact_messages[author_id])
            time_diff = (current_time - self.last_merge_time[author_id]).total_seconds()
            if message_count >= 3 or time_diff > 10:
                # 合并消息为一段文本
                merged_text = "\n".join(self.contact_messages[author_id])
                # 这里可以把merged_text赋值给你需要的变量，用于后续操作
                # 示例：processed_msg = merged_text
                print(f"[联系人{author_id}]合并后的消息：\n{merged_text}\n")
                
                # 重置该联系人的消息列表和合并时间
                self.contact_messages[author_id] = []
                self.last_merge_time[author_id] = current_time

if __name__ == "__main__":
    try:
        # 初始化客户端，登录时若需验证码会自动提示输入
        client = MessengerClient(os.getenv("FB_EMAIL"), os.getenv("FB_PASSWORD"))
        print("登录成功！开始监听新消息...")
        
        # 持续监听消息，每5秒检查一次（间隔别太短，避免触发Facebook反制）
        while True:
            client.listen()
            time.sleep(5)
    except Exception as e:
        print(f"运行出错：{e}")
    finally:
        client.logout()

关键功能说明 #

• 新消息检测：通过onMessage回调监听消息，用last_message_id过滤已处理消息，避免重复操作。
• 同一联系人消息合并：用字典按联系人ID分组存储消息，你可以自由调整合并条件（比如消息数量、时间间隔），合并后的文本直接赋值给变量即可用于后续操作。
• 新手友好：代码附带详细注释，用环境变量管理账号，轮询逻辑简单易懂，无需复杂的WebSocket配置。

注意事项 #

• Facebook有反自动化机制，轮询间隔建议设置为5-10秒，避免账号被临时限制。
• 优先用测试账号调试，不要用常用主账号。
• 若登录时提示验证码，按终端提示输入即可。
• 若fbchat出现登录失败问题，尝试升级到最新版本：pip install --upgrade fbchat

内容的提问来源于stack exchange，提问作者Mr. Stone