飞书应用机器人@影刀技术问答助理的python代码

# -*- coding: utf-8 -*-
"""
在指定消息下创建话题，并 @技术问答助理
================================================================
场景：运营助理已发送卡片消息（拿到 message_id），
     在该消息下创建话题，话题里第一条消息带上 @技术问答助理 + 用户原文 + 图片，
     让技术问答助理通过事件订阅收到并开始处理。

依赖：pip install requests

参考自 cyber-monkey 项目：
  - handlers.py:318  _build_forward_post()   ← 拼装 post 内容（@+文+图）
  - messages.py:342  upload_image()           ← 上传图片拿 image_key
  - messages.py:79   reply API + reply_in_thread  ← 创建话题
"""

import requests
import json
import os

from typing import List, Optional
# try:
#     from xbot.app.logging import trace as print
# except:
#     from xbot import print


# ==================== 飞书 API 封装 ====================

def _get_tenant_token(app_id: str, app_secret: str) -> str:
    """获取飞书租户凭证（tenant_access_token）"""
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = {"app_id": app_id, "app_secret": app_secret}
    resp = requests.post(url, json=payload, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise Exception(f"获取 Token 失败: {data.get('msg')}")
    return data["tenant_access_token"]


def _upload_image(token: str, image_path: str) -> str:
    """
    上传图片到飞书，拿到 image_key。
    对应 cyber-monkey/messages.py:342 的 upload_image()
    """
    url = "https://open.feishu.cn/open-apis/im/v1/images"
    headers = {"Authorization": f"Bearer {token}"}
    # image_type=message 表示这张图是用来发消息的
    data = {"image_type": "message"}
    with open(image_path, "rb") as f:
        files = {"image": (os.path.basename(image_path), f, "image/png")}
        resp = requests.post(url, headers=headers, data=data, files=files, timeout=30)
    resp.raise_for_status()
    res = resp.json()
    if res.get("code") != 0:
        raise Exception(f"上传图片失败: {res.get('msg')}")
    return res["data"]["image_key"]


def _reply_in_thread(token: str, message_id: str, post_content: dict) -> str:
    """
    在指定消息下创建话题并回复。
    对应飞书 API: POST /im/v1/messages/{message_id}/reply  (reply_in_thread=true)

    参数：
        token       : tenant_access_token
        message_id  : 要在其下创建话题的那条消息的 ID（通常是卡片消息的 message_id）
        post_content: post 富文本内容

    返回：
        新消息的 message_id
    """
    url = f"https://open.feishu.cn/open-apis/im/v1/messages/{message_id}/reply"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }
    payload = {
        "content": json.dumps(post_content, ensure_ascii=False),
        "msg_type": "post",
        "reply_in_thread": True,   # ← 关键：创建话题
    }
    resp = requests.post(url, headers=headers, json=payload, timeout=10)
    resp.raise_for_status()
    res = resp.json()
    if res.get("code") != 0:
        raise Exception(f"创建话题失败: {res.get('msg')}")
    return res["data"]["message_id"]


# ==================== 核心逻辑 ====================

def _build_post_content(
    tech_bot_open_id: str,
    text_content: str,
    image_keys: List[str],
    title: str = "",
) -> dict:
    """
    构造 post（富文本）消息体。
    对应 cyber-monkey/handlers.py:318 的 _build_forward_post()

    结构：
      [[@bot], [正文], [图片1], [图片2], ...]
    每个 block 是一个 list，飞书要求 content 是 [[block1], [block2], ...] 嵌套结构。
    """
    content = []

    # Block 1：@技术问答助理（最前面，触发 mentions 路由）
    content.append([{"tag": "at", "user_id": tech_bot_open_id}])

    # Block 2：用户原文
    if text_content:
        content.append([{"tag": "text", "text": text_content}])

    # Block 3+：图片（每张图一个独立 block）
    for img_key in image_keys:
        content.append([{"tag": "img", "image_key": img_key}])

    return {"zh_cn": {"title": title, "content": content}}


def send_to_tech_assistant(app_id, app_secret, message_id, tech_bot_open_id, text_content, image_paths):
    """
    title: 在消息下创建话题并@技术问答助理
    description: 在指定的消息 %message_id% 下创建话题，话题第一条消息带上 @技术问答助理 + 用户原文 %text_content% + 图片，让技术问答助理通过事件订阅收到并开始处理。
    inputs:
        - app_id (str): 运营助理应用的 App ID，eg: "cli_a5db1166713a900c"
        - app_secret (str): 运营助理应用的 App Secret，eg: "Imdfwys4jb2VanubbZpoqd1elIhnItKy"
        - message_id (str): 要在其下创建话题的消息 ID（通常是卡片消息发送后返回的 message_id），eg: "om_x100b6706f8c124a4b0350002e54450b"
        - tech_bot_open_id (str): 技术问答助理机器人的 open_id（ou_ 开头），eg: "ou_180d04de9af8f925124332453f441473"
        - text_content (str): 用户问题原文，eg: "修改应用信息处的标签怎么设置，没有可选项"
        - image_paths (list): 本地图片路径列表，eg: ["D:\\Desktop\\1.png"]
    outputs:
        - message_id (str): 创建话题后返回的消息 ID，eg: "om_bc564911586866593607429846077ffa"
    """
    # 1. 获取 Token
    token = _get_tenant_token(app_id, app_secret)

    # 2. 上传所有图片，拿到 image_key 列表
    image_keys: List[str] = []
    if image_paths:
        for path in image_paths:
            if not os.path.exists(path):
                raise FileNotFoundError(f"图片不存在: {path}")
            image_key = _upload_image(token, path)
            image_keys.append(image_key)
            print(f"图片已上传: {path} -> {image_key}")

    # 3. 构造 post 富文本内容（@bot + 原文 + 图片）
    post_content = _build_post_content(
        tech_bot_open_id=tech_bot_open_id,
        text_content=text_content,
        image_keys=image_keys,
    )

    # 4. 在 message_id 对应的消息下创建话题 + @技术问答助理
    new_message_id = _reply_in_thread(
        token=token,
        message_id=message_id,
        post_content=post_content,
    )
    print(f"话题已创建并 @技术问答助理，消息 message_id={new_message_id}")
    return new_message_id


# ==================== 影刀调用入口 ====================

# 使用此指令前，请确保安装必要的 Python 库，例如使用以下命令安装：
# pip install requests

# === 填写你的实际参数 ===
APP_ID = "XX"
APP_SECRET = "XXX"

# 卡片消息发送后返回的 message_id（在这条消息下创建话题）
MESSAGE_ID = "XXX"

# 技术问答助理的 open_id（@的对象）
TECH_BOT_OPEN_ID = "XXX"

# 用户原始问题
USER_TEXT = "修改应用信息处的标签怎么设置，没有可选项"

# 用户原始图片（本地路径，没有就传 None）
USER_IMAGES = [
    r"D:\Desktop\1.png",
    # r"C:\path\to\screenshot2.png",   # 多张图就继续加
]

# === 执行 ===
message_id = send_to_tech_assistant(
    app_id=APP_ID,
    app_secret=APP_SECRET,
    message_id=MESSAGE_ID,
    tech_bot_open_id=TECH_BOT_OPEN_ID,
    text_content=USER_TEXT,
    image_paths=USER_IMAGES,
)

# 影刀组件输出
output = message_id