OpenClaw龙虾在飞书中可以发附件吗？

可以，强调让龙虾走飞书附件通道，默认会用message工具发附件，但是有一定概率失败。

用api直接调用比较稳。

#!/usr/bin/env python3
"""
飞书文件/附件发送器 - 直接调用飞书 API
流程: 获取 token → 上传文件 → 发送附件消息

凭证从 OpenClaw 配置文件读取，不硬编码。
支持所有文件类型，自动识别 mime。
"""

import sys
import json
import os
import mimetypes
import requests

# 配置文件路径
CONFIG_PATH = os.path.expanduser("~/.openclaw/openclaw.json")
BASE_URL = "https://open.feishu.cn/open-apis"

def load_feishu_config(account_id: str = None) -> dict:
    """
    从 OpenClaw 配置文件读取飞书凭证。

    优先级（由高到低）：
    1. 环境变量 OPENCLAW_FEISHU_ACCOUNT_ID 指定账号名
    2. 参数 account_id（调用方传入）
    3. channels.feishu.accounts.<account_id>（需 account_id 不为空）
    4. channels.feishu（顶层默认账号配置）
    """
    if not os.path.exists(CONFIG_PATH):
        raise FileNotFoundError(f"配置文件不存在: {CONFIG_PATH}")

    with open(CONFIG_PATH) as f:
        config = json.load(f)

    feishu = config.get("channels", {}).get("feishu", {})
    accounts = feishu.get("accounts", {})

    effective_account = os.environ.get("OPENCLAW_FEISHU_ACCOUNT_ID", account_id)

    if effective_account and effective_account in accounts:
        acc = accounts[effective_account]
        if acc.get("appId") and acc.get("appSecret"):
            return {"app_id": acc["appId"], "app_secret": acc["appSecret"]}

    # 兜底：尝试顶层配置（默认账号）
    if feishu.get("appId") and feishu.get("appSecret"):
        return {"app_id": feishu["appId"], "app_secret": feishu["appSecret"]}

    raise ValueError(
        f"未在 ~/.openclaw/openclaw.json 中找到飞书凭证。"
        f"请确保存在 channels.feishu.accounts.<账号名> 或 channels.feishu 顶层配置，"
        f"并通过环境变量 OPENCLAW_FEISHU_ACCOUNT_ID 指定账号名。"
    )

def get_tenant_access_token(app_id: str, app_secret: str) -> str:
    """获取 tenant access token"""
    url = f"{BASE_URL}/auth/v3/tenant_access_token/internal"
    resp = requests.post(url, json={"app_id": app_id, "app_secret": app_secret}, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"获取 token 失败: {data}")
    return data["tenant_access_token"]

def detect_file_type(file_path: str) -> str:
    """根据文件扩展名识别飞书 file_type"""
    ext = os.path.splitext(file_path)[1].lower()
    type_map = {
        ".pdf": "pdf",
        ".doc": "doc",
        ".docx": "doc",
        ".xls": "xls",
        ".xlsx": "xls",
        ".ppt": "ppt",
        ".pptx": "ppt",
        ".mp4": "mp4",
        ".opus": "opus",
        ".ogg": "opus",
        ".ogg": "opus",
    }
    return type_map.get(ext, "stream")

def get_mime_type(file_path: str) -> str:
    """获取文件的 MIME 类型"""
    mime, _ = mimetypes.guess_type(file_path)
    if mime:
        return mime
    ext = os.path.splitext(file_path)[1].lower()
    if ext == ".ogg":
        return "audio/ogg"
    if ext == ".opus":
        return "audio/opus"
    if ext == ".md":
        return "text/markdown"
    if ext == ".py":
        return "text/x-python"
    if ext == ".json":
        return "application/json"
    if ext == ".txt":
        return "text/plain"
    return "application/octet-stream"

def upload_file(token: str, file_path: str) -> str:
    """
    上传文件到飞书，返回 file_key
    file_type: opus=语音, mp4=视频, pdf/xls/doc/ppt=Office, stream=其他
    """
    url = f"{BASE_URL}/im/v1/files"
    headers = {"Authorization": f"Bearer {token}"}

    file_name = os.path.basename(file_path)
    file_type = detect_file_type(file_path)
    mime_type = get_mime_type(file_path)

    with open(file_path, "rb") as f:
        files = {
            "file": (file_name, f, mime_type),
            "file_name": (None, file_name),
            "file_type": (None, file_type),
        }
        resp = requests.post(url, headers=headers, files=files, timeout=60)

    resp.raise_for_status()
    result = resp.json()
    if result.get("code") != 0:
        raise RuntimeError(f"上传文件失败: {result}")
    return result["data"]["file_key"]

def send_file_message(token: str, receive_id: str, file_key: str, file_name: str):
    """发送文件附件"""
    url = f"{BASE_URL}/im/v1/messages"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    content = json.dumps({
        "file_key": file_key,
        "file_name": file_name
    })

    params = {"receive_id_type": "open_id"}
    payload = {
        "receive_id": receive_id,
        "msg_type": "file",
        "content": content
    }

    resp = requests.post(url, headers=headers, params=params, json=payload, timeout=15)
    resp.raise_for_status()
    result = resp.json()
    if result.get("code") != 0:
        raise RuntimeError(f"发送消息失败: {result}")
    return result["data"]["message_id"]

def main():
    if len(sys.argv) < 3:
        print("用法:")
        print("  python3 feishu_file_api.py <文件路径> <用户open_id> [账号名]")
        print("")
        print("示例:")
        print("  python3 feishu_file_api.py /tmp/report.pdf ou_xxxxx")
        print("  FEISHU_USER_ID=ou_xxxxx python3 feishu_file_api.py /tmp/report.pdf ou_xxxxx todo")
        print("")
        print("凭证从 ~/.openclaw/openclaw.json 自动读取")
        print("可通过环境变量 OPENCLAW_FEISHU_ACCOUNT_ID 指定飞书账号")
        sys.exit(1)

    file_path = sys.argv[1]
    user_id = sys.argv[2]
    account_id = sys.argv[3] if len(sys.argv) > 3 else None

    if not os.path.exists(file_path):
        print(f"错误: 文件不存在 {file_path}")
        sys.exit(1)

    # 从配置文件读取凭证
    creds = load_feishu_config(account_id)
    print(f"=== 飞书文件发送 ===")
    print(f"账号: {creds['app_id']}")
    print(f"文件: {file_path}")
    print(f"用户: {user_id}")

    # 获取 token
    token = get_tenant_access_token(creds["app_id"], creds["app_secret"])
    print("\n[1/3] ✓ token 获取成功")

    # 上传文件
    file_key = upload_file(token, file_path)
    print(f"[2/3] ✓ 文件上传成功 (key: {file_key[:20]}...)")

    # 发送消息
    file_name = os.path.basename(file_path)
    print(f"[3/3] 发送附件...")
    msg_id = send_file_message(token, user_id, file_key, file_name)
    print(f"  ✓ 发送成功! message_id: {msg_id}")

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"\n✗ 错误: {e}")
        sys.exit(1)