You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
108 lines
3.5 KiB
108 lines
3.5 KiB
#!/usr/bin/env python3
|
|
"""
|
|
飞书图片下载脚本 - 从飞书群消息中下载图片资源
|
|
|
|
问题：飞书群消息中的图片无法通过 lark_api 工具直接下载（二进制写入限制），
|
|
需使用 requests 直接调用飞书 Open API。
|
|
|
|
使用方式：
|
|
python feishu_download.py --msg-id <msg_id> --file-key <file_key> --output image.jpg
|
|
python feishu_download.py --config ~/.mini-bridge/config.yaml --msg-id <msg_id> --file-key <file_key>
|
|
|
|
配置文件格式（~/.mini-bridge/config.yaml）：
|
|
feishu:
|
|
app_id: cli_xxxxxxxx
|
|
app_secret: xxxxxxxxxxxxxxxx
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
FEISHU_TOKEN_URL = (
|
|
"https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
|
|
)
|
|
FEISHU_RESOURCE_URL = (
|
|
"https://open.feishu.cn/open-apis/im/v1/messages/{msg_id}/resources/{file_key}"
|
|
)
|
|
|
|
|
|
def load_config(config_path: str) -> dict:
|
|
"""从 YAML 配置文件读取飞书凭证"""
|
|
import yaml
|
|
|
|
with open(config_path) as f:
|
|
config = yaml.safe_load(f)
|
|
feishu = config.get("feishu", config)
|
|
return {
|
|
"app_id": feishu["app_id"],
|
|
"app_secret": feishu["app_secret"],
|
|
}
|
|
|
|
|
|
def get_token(app_id: str, app_secret: str) -> str:
|
|
"""获取 tenant_access_token"""
|
|
resp = requests.post(
|
|
FEISHU_TOKEN_URL,
|
|
json={"app_id": app_id, "app_secret": app_secret},
|
|
timeout=10,
|
|
)
|
|
data = resp.json()
|
|
if data.get("code") != 0:
|
|
raise RuntimeError(f"获取 token 失败：{data}")
|
|
return data["tenant_access_token"]
|
|
|
|
|
|
def download_image(msg_id: str, file_key: str, output_path: Path, token: str) -> None:
|
|
"""下载飞书消息中的图片资源"""
|
|
url = FEISHU_RESOURCE_URL.format(msg_id=msg_id, file_key=file_key)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
params = {"type": "image"}
|
|
|
|
resp = requests.get(url, headers=headers, params=params, timeout=30)
|
|
if resp.status_code != 200:
|
|
raise RuntimeError(f"下载失败 HTTP {resp.status_code}: {resp.text[:200]}")
|
|
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(output_path, "wb") as f:
|
|
f.write(resp.content)
|
|
|
|
print(f"✅ 下载完成：{output_path} ({len(resp.content) / 1024:.1f} KB)")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="飞书图片下载工具")
|
|
parser.add_argument("--msg-id", required=True, help="飞书消息 ID")
|
|
parser.add_argument("--file-key", required=True, help="图片 file_key")
|
|
parser.add_argument("--output", "-o", required=True, help="输出文件路径")
|
|
parser.add_argument(
|
|
"--config", default="~/.mini-bridge/config.yaml", help="飞书凭证配置文件路径"
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# 加载配置
|
|
config_path = Path(args.config).expanduser()
|
|
if not config_path.exists():
|
|
print(f"❌ 配置文件不存在：{config_path}")
|
|
print(" 请先配置 ~/.mini-bridge/config.yaml，格式：")
|
|
print(" feishu:")
|
|
print(" app_id: cli_xxxxxxxx")
|
|
print(" app_secret: xxxxxxxxxxxxxxxx")
|
|
sys.exit(1)
|
|
|
|
credentials = load_config(str(config_path))
|
|
|
|
# 获取 token
|
|
print("🔑 获取飞书 access_token...")
|
|
token = get_token(credentials["app_id"], credentials["app_secret"])
|
|
|
|
# 下载图片
|
|
print(f"📥 下载图片（msg_id={args.msg_id}, file_key={args.file_key}）...")
|
|
download_image(args.msg_id, args.file_key, Path(args.output), token)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|