feat: image_daemon 改用飞书 WebSocket 事件订阅，实时接收图片消息

Dockerfile

@@ -6,7 +6,7 @@ RUN apt-get update && apt-get install -y \
     && rm -rf /var/lib/apt/lists/*
 
 # Python 依赖
-RUN pip3 install flask psutil requests --break-system-packages
+RUN pip3 install flask psutil requests lark-oapi --break-system-packages
 
 # 安装 OpenClaw
 RUN npm install -g openclaw@latest

image_daemon.py

@@ -1,115 +1,48 @@
 #!/usr/bin/env python3
 """
-飞书图片预处理守护进程 (image_daemon.py) v2
-后台运行，自动检测飞书聊天中的新图片消息，下载并上传到图床，
-然后以文本消息回复图片 URL，让 OpenClaw agent 直接看到可访问的 URL。
+飞书图片预处理守护进程 (image_daemon.py) v3 — WebSocket 事件驱动
+通过 lark-oapi SDK 的 WebSocket 长连接实时接收消息事件，
+检测图片消息后下载、上传到图床、回复 URL。
 """
-import os, sys, json, time, requests
+import os, sys, json, time, requests, threading
 
 FEISHU_BASE = "https://open.feishu.cn/open-apis"
-POLL_INTERVAL = int(os.environ.get("IMAGE_POLL_INTERVAL", "5"))
-processed_file = "/tmp/image_daemon_processed.json"
+APP_ID = os.environ.get("FEISHU_APP_ID", "")
+APP_SECRET = os.environ.get("FEISHU_APP_SECRET", "")
 
+# ---------- 日志 ----------
 def log(msg):
     ts = time.strftime("%H:%M:%S")
     print(f"[image_daemon {ts}] {msg}", flush=True)
 
-def load_processed():
-    try:
-        with open(processed_file) as f:
-            return set(json.load(f))
-    except:
-        return set()
-
-def save_processed(s):
-    with open(processed_file, "w") as f:
-        json.dump(list(s)[-500:], f)
-
-def get_tenant_token():
-    app_id = os.environ.get("FEISHU_APP_ID")
-    app_secret = os.environ.get("FEISHU_APP_SECRET")
-    if not app_id or not app_secret:
-        log(f"❌ 环境变量缺失: FEISHU_APP_ID={'set' if app_id else 'MISSING'}, FEISHU_APP_SECRET={'set' if app_secret else 'MISSING'}")
-        return None
-    resp = requests.post(f"{FEISHU_BASE}/auth/v3/tenant_access_token/internal",
-        json={"app_id": app_id, "app_secret": app_secret}, timeout=10)
-    data = resp.json()
-    if data.get("code") != 0:
-        log(f"❌ Token 获取失败: {data}")
-        return None
-    return data["tenant_access_token"]
-
-def get_bot_chats(token):
-    """获取 bot 参与的所有聊天"""
-    try:
-        headers = {"Authorization": f"Bearer {token}"}
-        resp = requests.get(f"{FEISHU_BASE}/im/v1/chats",
-            headers=headers, params={"page_size": 20}, timeout=10)
-        data = resp.json()
-        code = data.get("code", -1)
-        if code != 0:
-            msg = data.get("msg", "unknown")
-            log(f"❌ 获取聊天列表失败 (code={code}): {msg}")
-            log(f"   需要权限: im:chat / im:chat:readonly")
-            return []
-        items = data.get("data", {}).get("items", [])
-        return [c["chat_id"] for c in items]
-    except Exception as e:
-        log(f"❌ get_bot_chats 异常: {type(e).__name__}: {e}")
-        return []
-
-def get_recent_messages(token, chat_id, limit=50):
-    """获取聊天中最近的消息（不用 start_time，飞书 API 对新消息有索引延迟）"""
-    try:
-        headers = {"Authorization": f"Bearer {token}"}
-        resp = requests.get(f"{FEISHU_BASE}/im/v1/messages",
-            headers=headers,
-            params={
-                "container_id_type": "chat",
-                "container_id": chat_id,
-                "sort_type": "ByCreateTimeDesc",
-                "page_size": limit
-            }, timeout=10)
-        data = resp.json()
-        code = data.get("code", -1)
-        if code != 0:
-            msg = data.get("msg", "unknown")
-            log(f"❌ 获取消息失败 chat={chat_id} (code={code}): {msg}")
-            return []
-        return data.get("data", {}).get("items", [])
-    except Exception as e:
-        log(f"❌ get_recent_messages 异常: {type(e).__name__}: {e}")
-        return []
-
-def extract_image_keys(msg):
-    """从消息中提取所有 image_key"""
-    msg_type = msg.get("msg_type", "")
-    body_str = msg.get("body", {}).get("content", "{}")
-    try:
-        content = json.loads(body_str)
-    except:
-        return []
-
-    keys = []
-    if msg_type == "image":
-        k = content.get("image_key", "")
-        if k:
-            keys.append(k)
-    elif msg_type == "post":
-        for lang in ["zh_cn", "en_us"]:
-            lang_content = content.get(lang, {})
-            if isinstance(lang_content, dict):
-                for row in lang_content.get("content", []):
-                    if isinstance(row, list):
-                        for elem in row:
-                            if isinstance(elem, dict) and elem.get("tag") == "img":
-                                k = elem.get("image_key", "")
-                                if k:
-                                    keys.append(k)
-    return keys
+# ---------- Token 管理 ----------
+_token = None
+_token_time = 0
+_token_lock = threading.Lock()
+
+def get_token():
+    """获取 tenant_access_token，30分钟自动刷新"""
+    global _token, _token_time
+    with _token_lock:
+        if _token and time.time() - _token_time < 1800:
+            return _token
+        try:
+            resp = requests.post(f"{FEISHU_BASE}/auth/v3/tenant_access_token/internal",
+                json={"app_id": APP_ID, "app_secret": APP_SECRET}, timeout=10)
+            data = resp.json()
+            if data.get("code") == 0:
+                _token = data["tenant_access_token"]
+                _token_time = time.time()
+                log("🔑 Token 已刷新")
+                return _token
+            log(f"❌ Token 获取失败: {data}")
+        except Exception as e:
+            log(f"❌ Token 异常: {e}")
+    return _token  # 返回旧的，总比没有好
 
+# ---------- 图片下载 ----------
 def download_image(token, message_id, file_key):
-    """通过 获取消息中的资源文件 API 下载用户发送的图片"""
+    """通过消息资源 API 下载用户发送的图片"""
     headers = {"Authorization": f"Bearer {token}"}
     url = f"{FEISHU_BASE}/im/v1/messages/{message_id}/resources/{file_key}"
     log(f"📥 API: GET {url}?type=image")
@@ -118,12 +51,12 @@ def download_image(token, message_id, file_key):
         log(f"✅ 下载成功: {len(resp.content)} bytes")
         return resp.content
     log(f"❌ 下载图片失败 {file_key}: HTTP {resp.status_code}, {resp.text[:200]}")
-    log(f"   需要权限: im:resource")
     return None
 
+# ---------- 图片上传（多重 fallback） ----------
 def upload_image(data):
-    """上传图片到图床，按可用性排序，多重 fallback"""
-    # 1. tmpfiles.org — 当前最稳定
+    """上传图片到图床，多重 fallback"""
+    # 1. tmpfiles.org
     try:
         resp = requests.post("https://tmpfiles.org/api/v1/upload",
             files={"file": ("img.jpg", data, "image/jpeg")}, timeout=30)
@@ -137,9 +70,8 @@ def upload_image(data):
     except Exception as e:
         log(f"⚠️ tmpfiles 异常: {e}")
 
-    # 2. Telegraph (Telegra.ph) — 无需 API key
+    # 2. Telegraph
     try:
-        # Telegraph 的图片上传接口
         resp = requests.post("https://telegra.ph/upload",
             files={"file": ("img.jpg", data, "image/jpeg")}, timeout=30)
         if resp.status_code == 200:
@@ -154,20 +86,20 @@ def upload_image(data):
     except Exception as e:
         log(f"⚠️ telegraph 异常: {e}")
 
-    # 3. file.io — 一次性链接但可用性高
+    # 3. file.io
     try:
         resp = requests.post("https://file.io",
             files={"file": ("img.jpg", data, "image/jpeg")}, timeout=30)
         if resp.status_code == 200:
             url = resp.json().get("link", "")
             if url:
-                log(f"📤 file.io 成功 (注意: 链接仅可下载一次)")
+                log(f"📤 file.io 成功")
                 return url
         log(f"⚠️ file.io 失败: HTTP {resp.status_code}")
     except Exception as e:
         log(f"⚠️ file.io 异常: {e}")
 
-    # 4. catbox.moe — 近期返回 412，保留作为后备
+    # 4. catbox.moe
     try:
         resp = requests.post("https://catbox.moe/user/api.php",
             data={"reqtype": "fileupload"},
@@ -179,7 +111,7 @@ def upload_image(data):
     except Exception as e:
         log(f"⚠️ catbox 异常: {e}")
 
-    # 5. 0x0.st — 近期返回 403，保留作为最后手段
+    # 5. 0x0.st
     try:
         resp = requests.post("https://0x0.st",
             files={"file": ("img.jpg", data, "image/jpeg")}, timeout=30)
@@ -190,10 +122,10 @@ def upload_image(data):
     except Exception as e:
         log(f"⚠️ 0x0 异常: {e}")
 
-
     return None
 
-def reply_text(token, chat_id, msg_id, text):
+# ---------- 回复消息 ----------
+def reply_text(token, msg_id, text):
     headers = {
         "Authorization": f"Bearer {token}",
         "Content-Type": "application/json; charset=utf-8"
@@ -208,122 +140,125 @@ def reply_text(token, chat_id, msg_id, text):
     code = data.get("code", -1)
     if code != 0:
         log(f"❌ 回复消息失败 (code={code}): {data.get('msg', '')}")
-        log(f"   需要权限: im:message / im:message:create_as_bot")
     return data
 
-def process_message(token, chat_id, msg):
-    msg_id = msg.get("message_id", "")
-    sender_type = msg.get("sender", {}).get("sender_type", "")
-    if sender_type == "app":
+# ---------- 处理图片消息 ----------
+def handle_image_message(message_id, chat_id, image_key):
+    """下载 → 上传 → 回复"""
+    token = get_token()
+    if not token:
+        log("❌ 无法获取 token，跳过")
         return
 
-    image_keys = extract_image_keys(msg)
-    if not image_keys:
-        return
+    log(f"🖼️ 处理图片 image_key={image_key[:20]}... (msg={message_id[:16]}...)")
 
-    log(f"🖼️ 发现 {len(image_keys)} 张图片 (msg_id={msg_id[:16]}...)")
-    urls = []
-    for key in image_keys[:3]:
-        log(f"📥 下载图片 {key} (msg={msg_id[:16]}...)")
-        data = download_image(token, msg_id, key)
-        if data:
-            log(f"📥 {len(data)} bytes, 上传中...")
-            url = upload_image(data)
-            if url:
-                urls.append(url)
-                log(f"✅ {url}")
-            else:
-                path = f"/tmp/{key}.jpg"
-                with open(path, "wb") as f:
-                    f.write(data)
-                log(f"⚠️ 图��全部失败，本地保存: {path}")
-
-    if urls:
-        url_text = "\n".join(urls)
-        reply = f"[系统] 图片已自动处理，可通过以下链接查看：\n{url_text}"
-        result = reply_text(token, chat_id, msg_id, reply)
-        log(f"📤 已回复 (code={result.get('code', '?')})")
-
-def main():
-    log("🚀 启动中...")
-
-    # 环境诊断
-    app_id = os.environ.get("FEISHU_APP_ID", "")
-    log(f"📋 飞书 App ID: {app_id[:10]}..." if app_id else "❌ FEISHU_APP_ID 未设置!")
-
-    processed = load_processed()
-    token = None
-    token_time = 0
-    cycle = 0
-
-    while True:
-        try:
-            cycle += 1
-            # 每 30 分钟刷新一次 token
-            if not token or time.time() - token_time > 1800:
-                token = get_tenant_token()
-                if not token:
-                    log("⏳ Token 获取失败，60 秒后重试...")
-                    time.sleep(60)
-                    continue
-                token_time = time.time()
-                log("🔑 Token 已刷新")
-
-            chats = get_bot_chats(token)
-
-            # 每轮打印心跳（每 20 轮详细，其余简短）
-            if cycle == 1 or cycle % 20 == 0:
-                log(f"💓 轮询中... {len(chats)} 个聊天, 已处理 {len(processed)} 条 (第 {cycle} 轮)")
-            else:
-                log(f"💓 #{cycle} ok, {len(chats)} chats")
-
-            if not chats and cycle == 1:
-                log("⚠️ 未找到任何聊天！")
-
-            new_imgs = 0
-            for chat_id in chats:
-                messages = get_recent_messages(token, chat_id)
-
-                # 首轮打印消息类型统计
-                if cycle == 1:
-                    type_counts = {}
-                    for m in messages:
-                        t = m.get("msg_type", "unknown")
-                        type_counts[t] = type_counts.get(t, 0) + 1
-                    log(f"📊 聊天 {chat_id[:12]}... 共 {len(messages)} 条")
-                    log(f"   消息类型分布: {type_counts}")
-
-                new_in_chat = 0
-                for msg in messages:
-                    msg_id = msg.get("message_id", "")
-                    if msg_id in processed:
-                        continue
-                    processed.add(msg_id)
-                    new_in_chat += 1
+    # 下载
+    img_data = download_image(token, message_id, image_key)
+    if not img_data:
+        return
 
-                    msg_type = msg.get("msg_type", "")
-                    sender_type = msg.get("sender", {}).get("sender_type", "")
-                    body_preview = str(msg.get("body", {}).get("content", ""))[:100]
+    log(f"📥 {len(img_data)} bytes, 上传中...")
 
-                    log(f"📨 新消息: type={msg_type}, sender={sender_type}, preview={body_preview}")
+    # 上传
+    url = upload_image(img_data)
+    if url:
+        log(f"✅ {url}")
+        reply = f"[系统] 图片已自动处理，可通过以下链接查看：\n{url}"
+        result = reply_text(token, message_id, reply)
+        log(f"📤 已回复 (code={result.get('code', '?')})")
+    else:
+        path = f"/tmp/{image_key[:30]}.jpg"
+        with open(path, "wb") as f:
+            f.write(img_data)
+        log(f"⚠️ 图床全部失败，本地保存: {path}")
+
+# ---------- 事件处理 ----------
+def on_message_receive(data):
+    """im.message.receive_v1 事件回调"""
+    try:
+        event = data.event
+        message = event.message
+        sender = event.sender
+
+        msg_id = message.message_id
+        chat_id = message.chat_id
+        msg_type = message.message_type
+        sender_type = getattr(sender, 'sender_type', '') if sender else ''
+
+        # 只处理用户发的图片
+        if sender_type == "app":
+            return
+        if msg_type != "image":
+            return
+
+        content = json.loads(message.content)
+        image_key = content.get("image_key", "")
+        if not image_key:
+            log(f"⚠️ 图片消息但无 image_key: {message.content}")
+            return
+
+        log(f"📨 实时收到图片消息: msg_id={msg_id[:16]}..., image_key={image_key[:20]}...")
+
+        # 在新线程中处理，避免阻塞事件循环
+        t = threading.Thread(target=handle_image_message, args=(msg_id, chat_id, image_key))
+        t.daemon = True
+        t.start()
 
-                    # 处理非文本消息（图片、富文本等）
-                    if msg_type != "text" and sender_type != "app":
-                        process_message(token, chat_id, msg)
-                        new_imgs += 1
+    except Exception as e:
+        log(f"❌ on_message_receive 异常: {type(e).__name__}: {e}")
+        import traceback
+        traceback.print_exc()
 
-                if new_in_chat > 0 or cycle % 20 == 0:
-                    log(f"📊 chat={chat_id[:12]}... 拉取 {len(messages)} 条, 新 {new_in_chat} 条, 已知 {len(processed)} 条")
+# ---------- 主入口 ----------
+def main():
+    log("🚀 启动中... (WebSocket 事件驱动模式)")
+    log(f"📋 飞书 App ID: {APP_ID[:10]}..." if APP_ID else "❌ FEISHU_APP_ID 未设置!")
 
-            if new_imgs > 0:
-                save_processed(processed)
+    if not APP_ID or not APP_SECRET:
+        log("❌ FEISHU_APP_ID 或 FEISHU_APP_SECRET 未设置，退出")
+        sys.exit(1)
 
-        except Exception as e:
-            log(f"❌ 错误: {type(e).__name__}: {e}")
-            import traceback
-            traceback.print_exc()
+    # 预热 token
+    token = get_token()
+    if token:
+        log("✅ Token 获取成功")
+    else:
+        log("⚠️ Token 获取失败，稍后重试")
 
-        time.sleep(POLL_INTERVAL)
+    # 初始化 lark-oapi WebSocket 客户端
+    try:
+        import lark_oapi as lark
+        from lark_oapi.api.im.v1 import P2ImMessageReceiveV1
+        log("✅ lark-oapi SDK 已加载")
+    except ImportError:
+        log("❌ lark-oapi 未安装! 请执行: pip install lark-oapi")
+        sys.exit(1)
+
+    # 构建事件处理器
+    handler = lark.EventDispatcherHandler.builder("", "") \
+        .register_p2_im_message_receive_v1(on_message_receive) \
+        .build()
+
+    # 构建 WebSocket 客户端
+    from lark_oapi.ws import Client as WsClient
+    ws_client = WsClient(APP_ID, APP_SECRET, event_handler=handler, log_level=lark.LogLevel.INFO)
+
+    log("🔌 正在连接飞书 WebSocket...")
+    log("   订阅事件: im.message.receive_v1")
+    log("   等待图片消息...")
+
+    # 启动心跳线程
+    def heartbeat():
+        count = 0
+        while True:
+            time.sleep(60)
+            count += 1
+            log(f"� WebSocket 运行中 ({count} 分钟)")
+    hb = threading.Thread(target=heartbeat, daemon=True)
+    hb.start()
+
+    # 启动 WebSocket（阻塞）
+    ws_client.start()
 
 if __name__ == "__main__":
     main()