109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
#!/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()
|