feat: implement feishu / lark media file handling utilities for file, audio and video processing (#4938)

* feat: implement media file handling utilities for audio and video processing

* feat: refactor file upload handling for audio and video in LarkMessageEvent

* feat: add cleanup for failed audio and video conversion outputs in media_utils

* feat: add utility methods for sending messages and uploading files in LarkMessageEvent
This commit is contained in:
Soulter
2026-02-07 12:40:05 +08:00
committed by GitHub
parent 6416707e35
commit 22a109c2ae
3 changed files with 632 additions and 62 deletions

View File

@@ -3,13 +3,10 @@ import base64
import json
import re
import time
import uuid
from typing import Any, cast
import lark_oapi as lark
from lark_oapi.api.im.v1 import (
CreateMessageRequest,
CreateMessageRequestBody,
GetMessageResourceRequest,
)
from lark_oapi.api.im.v1.processor import P2ImMessageReceiveV1Processor
@@ -125,44 +122,23 @@ class LarkPlatformAdapter(Platform):
session: MessageSesion,
message_chain: MessageChain,
):
if self.lark_api.im is None:
logger.error("[Lark] API Client im 模块未初始化,无法发送消息")
return
res = await LarkMessageEvent._convert_to_lark(message_chain, self.lark_api)
wrapped = {
"zh_cn": {
"title": "",
"content": res,
},
}
if session.message_type == MessageType.GROUP_MESSAGE:
id_type = "chat_id"
if "%" in session.session_id:
session.session_id = session.session_id.split("%")[1]
receive_id = session.session_id
if "%" in receive_id:
receive_id = receive_id.split("%")[1]
else:
id_type = "open_id"
receive_id = session.session_id
request = (
CreateMessageRequest.builder()
.receive_id_type(id_type)
.request_body(
CreateMessageRequestBody.builder()
.receive_id(session.session_id)
.content(json.dumps(wrapped))
.msg_type("post")
.uuid(str(uuid.uuid4()))
.build(),
)
.build()
# 复用 LarkMessageEvent 中的通用发送逻辑
await LarkMessageEvent.send_message_chain(
message_chain,
self.lark_api,
receive_id=receive_id,
receive_id_type=id_type,
)
response = await self.lark_api.im.v1.message.acreate(request)
if not response.success():
logger.error(f"发送飞书消息失败({response.code}): {response.msg}")
await super().send_by_session(session, message_chain)
def meta(self) -> PlatformMetadata:

View File

@@ -6,6 +6,8 @@ from io import BytesIO
import lark_oapi as lark
from lark_oapi.api.im.v1 import (
CreateFileRequest,
CreateFileRequestBody,
CreateImageRequest,
CreateImageRequestBody,
CreateMessageReactionRequest,
@@ -17,10 +19,15 @@ from lark_oapi.api.im.v1 import (
from astrbot import logger
from astrbot.api.event import AstrMessageEvent, MessageChain
from astrbot.api.message_components import At, Plain
from astrbot.api.message_components import At, File, Plain, Record, Video
from astrbot.api.message_components import Image as AstrBotImage
from astrbot.core.utils.astrbot_path import get_astrbot_data_path
from astrbot.core.utils.io import download_image_by_url
from astrbot.core.utils.media_utils import (
convert_audio_to_opus,
convert_video_format,
get_media_duration,
)
class LarkMessageEvent(AstrMessageEvent):
@@ -35,6 +42,144 @@ class LarkMessageEvent(AstrMessageEvent):
super().__init__(message_str, message_obj, platform_meta, session_id)
self.bot = bot
@staticmethod
async def _send_im_message(
lark_client: lark.Client,
*,
content: str,
msg_type: str,
reply_message_id: str | None = None,
receive_id: str | None = None,
receive_id_type: str | None = None,
) -> bool:
"""发送飞书 IM 消息的通用辅助函数
Args:
lark_client: 飞书客户端
content: 消息内容JSON字符串
msg_type: 消息类型post/file/audio/media等
reply_message_id: 回复的消息ID用于回复消息
receive_id: 接收者ID用于主动发送
receive_id_type: 接收者ID类型用于主动发送
Returns:
是否发送成功
"""
if lark_client.im is None:
logger.error("[Lark] API Client im 模块未初始化")
return False
if reply_message_id:
request = (
ReplyMessageRequest.builder()
.message_id(reply_message_id)
.request_body(
ReplyMessageRequestBody.builder()
.content(content)
.msg_type(msg_type)
.uuid(str(uuid.uuid4()))
.reply_in_thread(False)
.build()
)
.build()
)
response = await lark_client.im.v1.message.areply(request)
else:
from lark_oapi.api.im.v1 import (
CreateMessageRequest,
CreateMessageRequestBody,
)
if receive_id_type is None or receive_id is None:
logger.error(
"[Lark] 主动发送消息时receive_id 和 receive_id_type 不能为空",
)
return False
request = (
CreateMessageRequest.builder()
.receive_id_type(receive_id_type)
.request_body(
CreateMessageRequestBody.builder()
.receive_id(receive_id)
.content(content)
.msg_type(msg_type)
.uuid(str(uuid.uuid4()))
.build()
)
.build()
)
response = await lark_client.im.v1.message.acreate(request)
if not response.success():
logger.error(f"[Lark] 发送飞书消息失败({response.code}): {response.msg}")
return False
return True
@staticmethod
async def _upload_lark_file(
lark_client: lark.Client,
*,
path: str,
file_type: str,
duration: int | None = None,
) -> str | None:
"""上传文件到飞书的通用辅助函数
Args:
lark_client: 飞书客户端
path: 文件路径
file_type: 文件类型stream/opus/mp4等
duration: 媒体时长(毫秒),可选
Returns:
成功返回file_key失败返回None
"""
if not path or not os.path.exists(path):
logger.error(f"[Lark] 文件不存在: {path}")
return None
if lark_client.im is None:
logger.error("[Lark] API Client im 模块未初始化,无法上传文件")
return None
try:
with open(path, "rb") as file_obj:
body_builder = (
CreateFileRequestBody.builder()
.file_type(file_type)
.file_name(os.path.basename(path))
.file(file_obj)
)
if duration is not None:
body_builder.duration(duration)
request = (
CreateFileRequest.builder()
.request_body(body_builder.build())
.build()
)
response = await lark_client.im.v1.file.acreate(request)
if not response.success():
logger.error(
f"[Lark] 无法上传文件({response.code}): {response.msg}"
)
return None
if response.data is None:
logger.error("[Lark] 上传文件成功但未返回数据(data is None)")
return None
file_key = response.data.file_key
logger.debug(f"[Lark] 文件上传成功: {file_key}")
return file_key
except Exception as e:
logger.error(f"[Lark] 无法打开或上传文件: {e}")
return None
@staticmethod
async def _convert_to_lark(message: MessageChain, lark_client: lark.Client) -> list:
ret = []
@@ -103,6 +248,18 @@ class LarkMessageEvent(AstrMessageEvent):
ret.append(_stage)
ret.append([{"tag": "img", "image_key": image_key}])
_stage.clear()
elif isinstance(comp, File):
# 文件将通过 _send_file_message 方法单独发送,这里跳过
logger.debug("[Lark] 检测到文件组件,将单独发送")
continue
elif isinstance(comp, Record):
# 音频将通过 _send_audio_message 方法单独发送,这里跳过
logger.debug("[Lark] 检测到音频组件,将单独发送")
continue
elif isinstance(comp, Video):
# 视频将通过 _send_media_message 方法单独发送,这里跳过
logger.debug("[Lark] 检测到视频组件,将单独发送")
continue
else:
logger.warning(f"飞书 暂时不支持消息段: {comp.type}")
@@ -110,40 +267,270 @@ class LarkMessageEvent(AstrMessageEvent):
ret.append(_stage)
return ret
async def send(self, message: MessageChain):
res = await LarkMessageEvent._convert_to_lark(message, self.bot)
wrapped = {
"zh_cn": {
"title": "",
"content": res,
},
}
@staticmethod
async def send_message_chain(
message_chain: MessageChain,
lark_client: lark.Client,
reply_message_id: str | None = None,
receive_id: str | None = None,
receive_id_type: str | None = None,
):
"""通用的消息链发送方法
request = (
ReplyMessageRequest.builder()
.message_id(self.message_obj.message_id)
.request_body(
ReplyMessageRequestBody.builder()
.content(json.dumps(wrapped))
.msg_type("post")
.uuid(str(uuid.uuid4()))
.reply_in_thread(False)
.build(),
)
.build()
)
if self.bot.im is None:
logger.error("[Lark] API Client im 模块未初始化,无法回复消息")
Args:
message_chain: 要发送的消息链
lark_client: 飞书客户端
reply_message_id: 回复的消息ID用于回复消息
receive_id: 接收者ID用于主动发送
receive_id_type: 接收者ID类型'open_id', 'chat_id'(用于主动发送)
"""
if lark_client.im is None:
logger.error("[Lark] API Client im 模块未初始化")
return
response = await self.bot.im.v1.message.areply(request)
# 分离文件、音频、视频组件和其他组件
file_components: list[File] = []
audio_components: list[Record] = []
media_components: list[Video] = []
other_components = []
if not response.success():
logger.error(f"回复飞书消息失败({response.code}): {response.msg}")
for comp in message_chain.chain:
if isinstance(comp, File):
file_components.append(comp)
elif isinstance(comp, Record):
audio_components.append(comp)
elif isinstance(comp, Video):
media_components.append(comp)
else:
other_components.append(comp)
# 先发送非文件内容(如果有)
if other_components:
temp_chain = MessageChain()
temp_chain.chain = other_components
res = await LarkMessageEvent._convert_to_lark(temp_chain, lark_client)
if res: # 只在有内容时发送
wrapped = {
"zh_cn": {
"title": "",
"content": res,
},
}
await LarkMessageEvent._send_im_message(
lark_client,
content=json.dumps(wrapped),
msg_type="post",
reply_message_id=reply_message_id,
receive_id=receive_id,
receive_id_type=receive_id_type,
)
# 发送附件
for file_comp in file_components:
await LarkMessageEvent._send_file_message(
file_comp, lark_client, reply_message_id, receive_id, receive_id_type
)
for audio_comp in audio_components:
await LarkMessageEvent._send_audio_message(
audio_comp, lark_client, reply_message_id, receive_id, receive_id_type
)
for media_comp in media_components:
await LarkMessageEvent._send_media_message(
media_comp, lark_client, reply_message_id, receive_id, receive_id_type
)
async def send(self, message: MessageChain):
"""发送消息链到飞书,然后交给父类做框架级发送/记录"""
await LarkMessageEvent.send_message_chain(
message,
self.bot,
reply_message_id=self.message_obj.message_id,
)
await super().send(message)
@staticmethod
async def _send_file_message(
file_comp: File,
lark_client: lark.Client,
reply_message_id: str | None = None,
receive_id: str | None = None,
receive_id_type: str | None = None,
):
"""发送文件消息
Args:
file_comp: 文件组件
lark_client: 飞书客户端
reply_message_id: 回复的消息ID用于回复消息
receive_id: 接收者ID用于主动发送
receive_id_type: 接收者ID类型用于主动发送
"""
file_path = file_comp.file or ""
file_key = await LarkMessageEvent._upload_lark_file(
lark_client, path=file_path, file_type="stream"
)
if not file_key:
return
content = json.dumps({"file_key": file_key})
await LarkMessageEvent._send_im_message(
lark_client,
content=content,
msg_type="file",
reply_message_id=reply_message_id,
receive_id=receive_id,
receive_id_type=receive_id_type,
)
@staticmethod
async def _send_audio_message(
audio_comp: Record,
lark_client: lark.Client,
reply_message_id: str | None = None,
receive_id: str | None = None,
receive_id_type: str | None = None,
):
"""发送音频消息
Args:
audio_comp: 音频组件
lark_client: 飞书客户端
reply_message_id: 回复的消息ID用于回复消息
receive_id: 接收者ID用于主动发送
receive_id_type: 接收者ID类型用于主动发送
"""
# 获取音频文件路径
try:
original_audio_path = await audio_comp.convert_to_file_path()
except Exception as e:
logger.error(f"[Lark] 无法获取音频文件路径: {e}")
return
if not original_audio_path or not os.path.exists(original_audio_path):
logger.error(f"[Lark] 音频文件不存在: {original_audio_path}")
return
# 转换为opus格式
converted_audio_path = None
try:
audio_path = await convert_audio_to_opus(original_audio_path)
# 如果转换后路径与原路径不同,说明生成了新文件
if audio_path != original_audio_path:
converted_audio_path = audio_path
else:
audio_path = original_audio_path
except Exception as e:
logger.error(f"[Lark] 音频格式转换失败,将尝试直接上传: {e}")
# 如果转换失败,继续尝试直接上传原始文件
audio_path = original_audio_path
# 获取音频时长
duration = await get_media_duration(audio_path)
# 上传音频文件
file_key = await LarkMessageEvent._upload_lark_file(
lark_client,
path=audio_path,
file_type="opus",
duration=duration,
)
# 清理转换后的临时音频文件
if converted_audio_path and os.path.exists(converted_audio_path):
try:
os.remove(converted_audio_path)
logger.debug(f"[Lark] 已删除转换后的音频文件: {converted_audio_path}")
except Exception as e:
logger.warning(f"[Lark] 删除转换后的音频文件失败: {e}")
if not file_key:
return
await LarkMessageEvent._send_im_message(
lark_client,
content=json.dumps({"file_key": file_key}),
msg_type="audio",
reply_message_id=reply_message_id,
receive_id=receive_id,
receive_id_type=receive_id_type,
)
@staticmethod
async def _send_media_message(
media_comp: Video,
lark_client: lark.Client,
reply_message_id: str | None = None,
receive_id: str | None = None,
receive_id_type: str | None = None,
):
"""发送视频消息
Args:
media_comp: 视频组件
lark_client: 飞书客户端
reply_message_id: 回复的消息ID用于回复消息
receive_id: 接收者ID用于主动发送
receive_id_type: 接收者ID类型用于主动发送
"""
# 获取视频文件路径
try:
original_video_path = await media_comp.convert_to_file_path()
except Exception as e:
logger.error(f"[Lark] 无法获取视频文件路径: {e}")
return
if not original_video_path or not os.path.exists(original_video_path):
logger.error(f"[Lark] 视频文件不存在: {original_video_path}")
return
# 转换为mp4格式
converted_video_path = None
try:
video_path = await convert_video_format(original_video_path, "mp4")
# 如果转换后路径与原路径不同,说明生成了新文件
if video_path != original_video_path:
converted_video_path = video_path
else:
video_path = original_video_path
except Exception as e:
logger.error(f"[Lark] 视频格式转换失败,将尝试直接上传: {e}")
# 如果转换失败,继续尝试直接上传原始文件
video_path = original_video_path
# 获取视频时长
duration = await get_media_duration(video_path)
# 上传视频文件
file_key = await LarkMessageEvent._upload_lark_file(
lark_client,
path=video_path,
file_type="mp4",
duration=duration,
)
# 清理转换后的临时视频文件
if converted_video_path and os.path.exists(converted_video_path):
try:
os.remove(converted_video_path)
logger.debug(f"[Lark] 已删除转换后的视频文件: {converted_video_path}")
except Exception as e:
logger.warning(f"[Lark] 删除转换后的视频文件失败: {e}")
if not file_key:
return
await LarkMessageEvent._send_im_message(
lark_client,
content=json.dumps({"file_key": file_key}),
msg_type="media",
reply_message_id=reply_message_id,
receive_id=receive_id,
receive_id_type=receive_id_type,
)
async def react(self, emoji: str):
if self.bot.im is None:
logger.error("[Lark] API Client im 模块未初始化,无法发送表情")

View File

@@ -0,0 +1,207 @@
"""媒体文件处理工具
提供音视频格式转换、时长获取等功能。
"""
import asyncio
import os
import subprocess
import uuid
from astrbot import logger
from astrbot.core.utils.astrbot_path import get_astrbot_data_path
async def get_media_duration(file_path: str) -> int | None:
"""使用ffprobe获取媒体文件时长
Args:
file_path: 媒体文件路径
Returns:
时长毫秒如果获取失败返回None
"""
try:
# 使用ffprobe获取时长
process = await asyncio.create_subprocess_exec(
"ffprobe",
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
file_path,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
stdout, stderr = await process.communicate()
if process.returncode == 0 and stdout:
duration_seconds = float(stdout.decode().strip())
duration_ms = int(duration_seconds * 1000)
logger.debug(f"[Media Utils] 获取媒体时长: {duration_ms}ms")
return duration_ms
else:
logger.warning(f"[Media Utils] 无法获取媒体文件时长: {file_path}")
return None
except FileNotFoundError:
logger.warning(
"[Media Utils] ffprobe未安装或不在PATH中无法获取媒体时长。请安装ffmpeg: https://ffmpeg.org/"
)
return None
except Exception as e:
logger.warning(f"[Media Utils] 获取媒体时长时出错: {e}")
return None
async def convert_audio_to_opus(audio_path: str, output_path: str | None = None) -> str:
"""使用ffmpeg将音频转换为opus格式
Args:
audio_path: 原始音频文件路径
output_path: 输出文件路径如果为None则自动生成
Returns:
转换后的opus文件路径
Raises:
Exception: 转换失败时抛出异常
"""
# 如果已经是opus格式直接返回
if audio_path.lower().endswith(".opus"):
return audio_path
# 生成输出文件路径
if output_path is None:
temp_dir = os.path.join(get_astrbot_data_path(), "temp")
os.makedirs(temp_dir, exist_ok=True)
output_path = os.path.join(temp_dir, f"{uuid.uuid4()}.opus")
try:
# 使用ffmpeg转换为opus格式
# -y: 覆盖输出文件
# -i: 输入文件
# -acodec libopus: 使用opus编码器
# -ac 1: 单声道
# -ar 16000: 采样率16kHz
process = await asyncio.create_subprocess_exec(
"ffmpeg",
"-y",
"-i",
audio_path,
"-acodec",
"libopus",
"-ac",
"1",
"-ar",
"16000",
output_path,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
# 清理可能已生成但无效的临时文件
if output_path and os.path.exists(output_path):
try:
os.remove(output_path)
logger.debug(
f"[Media Utils] 已清理失败的opus输出文件: {output_path}"
)
except OSError as e:
logger.warning(f"[Media Utils] 清理失败的opus输出文件时出错: {e}")
error_msg = stderr.decode() if stderr else "未知错误"
logger.error(f"[Media Utils] ffmpeg转换音频失败: {error_msg}")
raise Exception(f"ffmpeg conversion failed: {error_msg}")
logger.debug(f"[Media Utils] 音频转换成功: {audio_path} -> {output_path}")
return output_path
except FileNotFoundError:
logger.error(
"[Media Utils] ffmpeg未安装或不在PATH中无法转换音频格式。请安装ffmpeg: https://ffmpeg.org/"
)
raise Exception("ffmpeg not found")
except Exception as e:
logger.error(f"[Media Utils] 转换音频格式时出错: {e}")
raise
async def convert_video_format(
video_path: str, output_format: str = "mp4", output_path: str | None = None
) -> str:
"""使用ffmpeg转换视频格式
Args:
video_path: 原始视频文件路径
output_format: 目标格式默认mp4
output_path: 输出文件路径如果为None则自动生成
Returns:
转换后的视频文件路径
Raises:
Exception: 转换失败时抛出异常
"""
# 如果已经是目标格式,直接返回
if video_path.lower().endswith(f".{output_format}"):
return video_path
# 生成输出文件路径
if output_path is None:
temp_dir = os.path.join(get_astrbot_data_path(), "temp")
os.makedirs(temp_dir, exist_ok=True)
output_path = os.path.join(temp_dir, f"{uuid.uuid4()}.{output_format}")
try:
# 使用ffmpeg转换视频格式
process = await asyncio.create_subprocess_exec(
"ffmpeg",
"-y",
"-i",
video_path,
"-c:v",
"libx264",
"-c:a",
"aac",
output_path,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
# 清理可能已生成但无效的临时文件
if output_path and os.path.exists(output_path):
try:
os.remove(output_path)
logger.debug(
f"[Media Utils] 已清理失败的{output_format}输出文件: {output_path}"
)
except OSError as e:
logger.warning(
f"[Media Utils] 清理失败的{output_format}输出文件时出错: {e}"
)
error_msg = stderr.decode() if stderr else "未知错误"
logger.error(f"[Media Utils] ffmpeg转换视频失败: {error_msg}")
raise Exception(f"ffmpeg conversion failed: {error_msg}")
logger.debug(f"[Media Utils] 视频转换成功: {video_path} -> {output_path}")
return output_path
except FileNotFoundError:
logger.error(
"[Media Utils] ffmpeg未安装或不在PATH中无法转换视频格式。请安装ffmpeg: https://ffmpeg.org/"
)
raise Exception("ffmpeg not found")
except Exception as e:
logger.error(f"[Media Utils] 转换视频格式时出错: {e}")
raise