#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 飞书 API 客户端模块 提供飞书图片、文件下载功能 """ import os import json import base64 import logging import requests from datetime import datetime # 添加项目根目录到 Python 路径 import sys BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, BASE_DIR) logger = logging.getLogger('feishu_api') class FeishuAPIClient: """飞书 API 客户端""" def __init__(self, app_id=None, app_secret=None): """ 初始化飞书 API 客户端 Args: app_id: 飞书应用 App ID app_secret: 飞书应用 App Secret """ # 加载配置 self.config = self._load_config() self.app_id = app_id or self.config.get('FEISHU_APP_ID', '') self.app_secret = app_secret or self.config.get('FEISHU_APP_SECRET', '') self.api_base = 'https://open.feishu.cn/open-apis' self.access_token = None self.token_expires_at = 0 logger.info(f"🔑 飞书 API 客户端初始化 - App ID: {self.app_id[:10]}...") def _load_config(self): """加载配置""" try: from feishu_config import FEISHU_APP_ID, FEISHU_APP_SECRET return {'FEISHU_APP_ID': FEISHU_APP_ID, 'FEISHU_APP_SECRET': FEISHU_APP_SECRET} except ImportError: return {} def get_access_token(self): """ 获取访问令牌 Returns: str: 访问令牌 """ # 检查令牌是否过期 if self.access_token and datetime.now().timestamp() < self.token_expires_at: return self.access_token try: url = f'{self.api_base}/auth/v3/tenant_access_token/internal' response = requests.post( url, json={ 'app_id': self.app_id, 'app_secret': self.app_secret }, timeout=30 ) if response.status_code == 200: result = response.json() if result.get('code') == 0: self.access_token = result.get('tenant_access_token', '') self.token_expires_at = datetime.now().timestamp() + result.get('expire', 7200) - 300 logger.info("✅ 获取飞书访问令牌成功") return self.access_token else: logger.error(f"获取飞书令牌失败：{result.get('msg', '未知错误')}") return None else: logger.error(f"获取飞书令牌失败 - HTTP {response.status_code}") return None except Exception as e: logger.error(f"获取飞书令牌异常：{str(e)}") return None def download_image(self, image_key, save_dir=None): """ 下载飞书图片 Args: image_key: 图片 key save_dir: 保存目录 Returns: str: 图片本地路径 """ if not image_key: logger.warning("图片 key 为空") return None # 获取访问令牌 token = self.get_access_token() if not token: logger.error("无法获取飞书访问令牌") return None # 构建下载 URL url = f'{self.api_base}/im/v1/images/{image_key}' try: headers = { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json' } response = requests.get(url, headers=headers, timeout=30) if response.status_code == 200: result = response.json() if result.get('code') == 0: image_data = result.get('image', '') image_type = result.get('image_type', 'jpg') # 生成文件名 timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f"feishu_image_{timestamp}.{image_type}" # 保存目录 if not save_dir: save_dir = os.path.join(BASE_DIR, 'temp') os.makedirs(save_dir, exist_ok=True) file_path = os.path.join(save_dir, filename) # 保存图片（base64 解码） try: image_bytes = base64.b64decode(image_data) with open(file_path, 'wb') as f: f.write(image_bytes) logger.info(f"✅ 图片解码成功：{len(image_bytes)} 字节") except Exception as e: logger.error(f"图片解码失败：{str(e)}") return None logger.info(f"✅ 图片下载成功：{file_path}") return file_path else: logger.error(f"下载图片失败：{result.get('msg', '未知错误')}") return None else: logger.error(f"下载图片失败 - HTTP {response.status_code}") return None except Exception as e: logger.error(f"下载图片异常：{str(e)}") return None def download_file(self, file_key, file_name, save_dir=None): """ 下载飞书文件 Args: file_key: 文件 key file_name: 文件名 save_dir: 保存目录 Returns: str: 文件本地路径 """ if not file_key: logger.warning("文件 key 为空") return None # 获取访问令牌 token = self.get_access_token() if not token: logger.error("无法获取飞书访问令牌") return None # 构建下载 URL url = f'{self.api_base}/im/v1/files/{file_key}' try: headers = { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json' } response = requests.get(url, headers=headers, timeout=60) if response.status_code == 200: result = response.json() if result.get('code') == 0: file_data = result.get('file', '') # 生成文件名 timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f"feishu_file_{timestamp}_{file_name}" # 保存目录 if not save_dir: save_dir = os.path.join(BASE_DIR, 'temp') os.makedirs(save_dir, exist_ok=True) file_path = os.path.join(save_dir, filename) # 保存文件（base64 解码） try: file_bytes = base64.b64decode(file_data) with open(file_path, 'wb') as f: f.write(file_bytes) logger.info(f"✅ 文件解码成功：{len(file_bytes)} 字节") except Exception as e: logger.error(f"文件解码失败：{str(e)}") return None logger.info(f"✅ 文件下载成功：{file_path}") return file_path else: logger.error(f"下载文件失败：{result.get('msg', '未知错误')}") return None else: logger.error(f"下载文件失败 - HTTP {response.status_code}") return None except Exception as e: logger.error(f"下载文件异常：{str(e)}") return None def get_message_images(self, message_id): """ 获取消息中的图片列表 Args: message_id: 消息 ID Returns: list: 图片列表 """ token = self.get_access_token() if not token: return [] url = f'{self.api_base}/im/v1/messages/{message_id}/resources' try: headers = { 'Authorization': f'Bearer {token}' } response = requests.get(url, headers=headers, timeout=30) if response.status_code == 200: result = response.json() if result.get('code') == 0: return result.get('items', []) else: logger.error(f"获取消息资源失败：{result.get('msg', '未知错误')}") return [] else: logger.error(f"获取消息资源失败 - HTTP {response.status_code}") return [] except Exception as e: logger.error(f"获取消息资源异常：{str(e)}") return [] def create_feishu_client(app_id=None, app_secret=None): """创建飞书 API 客户端实例""" return FeishuAPIClient(app_id, app_secret) if __name__ == '__main__': # 测试 client = create_feishu_client() token = client.get_access_token() if token: print(f"✅ 飞书 API 客户端初始化成功") print(f" 访问令牌：{token[:20]}...") else: print("❌ 飞书 API 客户端初始化失败")