knowledge/feishu_client.py

# feishu_client.py
import requests
import json
import logging
import time

class FeishuClient:
    """
    封装飞书多维表格的API操作。
    负责获取Access Token，以及对多维表格的读写操作。
    """
    def __init__(self, app_id: str, app_secret: str):
        self.app_id = app_id
        self.app_secret = app_secret
        self._access_token = None
        self._token_expires_at = 0 # Unix timestamp, 用于缓存Token过期时间

    def _get_access_token(self) -> str:
        """
        获取飞书应用的 Access Token。
        如果当前token未过期，则直接返回；否则，重新请求获取。
        Token有效期通常为2小时，这里会提前120秒刷新。
        """
        # 检查缓存的token是否仍然有效
        if self._access_token and self._token_expires_at > time.time():
            return self._access_token

        url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
        headers = {"Content-Type": "application/json; charset=utf-8"}
        payload = {
            "app_id": self.app_id,
            "app_secret": self.app_secret
        }
        try:
            logging.info("正在请求飞书 Access Token...")
            response = requests.post(url, headers=headers, json=payload, timeout=10)
            response.raise_for_status() # 检查HTTP响应状态码，如果不是2xx则抛出HTTPError
            data = response.json()
            if data.get("code") == 0:
                self._access_token = data.get("app_access_token")
                # 记录过期时间，提前2分钟刷新
                self._token_expires_at = time.time() + data.get("expire", 7200) - 120
                logging.info("飞书 Access Token 获取成功。")
                return self._access_token
            else:
                logging.error(f"获取飞书 Access Token 失败: {data.get('msg')} (Code: {data.get('code')})")
                raise Exception(f"Feishu Token Error: {data.get('msg')}")
        except requests.exceptions.RequestException as e:
            logging.error(f"请求飞书 Access Token 发生网络错误: {e}")
            raise Exception(f"Feishu Token Request Failed: {e}")
        except Exception as e:
            logging.error(f"处理飞书 Access Token 响应失败: {e}")
            raise Exception(f"Feishu Token Response Handling Failed: {e}")

    def read_records(self, base_id: str, table_id: str, field_names: list, page_size: int = 100) -> list:
        """
        从飞书多维表格读取记录。
        Args:
            base_id (str): 多维表格的 Base ID (应用 Token)。
            table_id (str): 表的 Table ID。
            field_names (list): 要读取的字段名称列表。
            page_size (int): 每次请求的记录数 (最大500)。
        Returns:
            list: 包含记录字典的列表。
        Raises:
            Exception: 如果读取失败。
        """
        access_token = self._get_access_token()
        url = (
            f"https://open.feishu.cn/open-apis/bitable/v1/apps/{base_id}/tables/{table_id}/records"
        )
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }
        params = {
            "page_size": min(page_size, 500), # 飞书API page_size 最大500
            "field_names": json.dumps(field_names)
        }
        records = []
        try:
            logging.info(f"正在从飞书多维表格读取数据 (表: {table_id}, 请求字段: {field_names}, 每页: {params['page_size']})...")
            response = requests.get(url, headers=headers, params=params, timeout=30)
            response.raise_for_status()
            data = response.json()
            if data.get("code") == 0:
                records.extend(data.get("data", {}).get("items", []))
                # 注意：这里只处理了第一页数据。如果需要读取所有数据，您需要实现分页循环逻辑
                if data.get("data", {}).get("has_more"):
                    logging.warning(f"飞书多维表格 (表: {table_id}) 存在更多数据未读取。当前实现仅获取了第一页。")
                logging.info(f"成功读取 {len(records)} 条记录。")
                return records
            else:
                logging.error(f"从飞书多维表格读取数据失败: {data.get('msg')} (Code: {data.get('code')})")
                raise Exception(f"Feishu Read Error: {data.get('msg')}")
        except requests.exceptions.RequestException as e:
            logging.error(f"请求飞书多维表格读取数据发生网络错误: {e}")
            raise Exception(f"Feishu Read Request Failed: {e}")
        except Exception as e:
            logging.error(f"处理飞书多维表格读取响应失败: {e}")
            raise Exception(f"Feishu Read Response Handling Failed: {e}")

    def update_records(self, base_id: str, table_id: str, records_data: list) -> bool:
        """
        更新飞书多维表格中的记录。
        Args:
            base_id (str): 多维表格的 Base ID。
            table_id (str): 表的 Table ID。
            records_data (list): 包含要更新的记录字典的列表，每个字典需包含 record_id 和 fields。
                                 例如: [{"record_id": "recxxxxxxxx", "fields": {"字段名": "新内容"}}]
                                 每次批量更新最多支持500条记录。
        Returns:
            bool: True 如果更新成功，False 如果失败。
        Raises:
            Exception: 如果更新失败。
        """
        if not records_data:
            logging.info("没有要更新的记录，跳过写入飞书操作。")
            return True

        access_token = self._get_access_token()
        url = (
            f"https://open.feishu.cn/open-apis/bitable/v1/apps/{base_id}/tables/{table_id}/records"
        )
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }
        
        # 飞书批量更新API限制每批最多500条
        chunk_size = 500
        all_success = True

        for i in range(0, len(records_data), chunk_size):
            batch_records = records_data[i:i + chunk_size]
            payload = {
                "records": batch_records
            }
            try:
                logging.info(f"正在将 {len(batch_records)} 条记录分批写回飞书多维表格 (表: {table_id}, 批次: {i//chunk_size + 1})...")
                response = requests.post(url, headers=headers, json=payload, timeout=30)
                response.raise_for_status()
                data = response.json()
                if data.get("code") == 0:
                    logging.info(f"批次 {i//chunk_size + 1} 成功写入/更新 {len(batch_records)} 条记录。")
                else:
                    logging.error(f"将数据写入飞书多维表格失败 (批次 {i//chunk_size + 1}): {data.get('msg')} (Code: {data.get('code')})")
                    all_success = False
                    # 可以在这里选择抛出异常中断，或者继续处理下一批
                    # raise Exception(f"Feishu Write Error in batch {i//chunk_size + 1}: {data.get('msg')}")
            except requests.exceptions.RequestException as e:
                logging.error(f"请求飞书多维表格写入数据发生网络错误 (批次 {i//chunk_size + 1}): {e}")
                all_success = False
                # raise Exception(f"Feishu Write Request Failed in batch {i//chunk_size + 1}: {e}")
            except Exception as e:
                logging.error(f"处理飞书多维表格写入响应失败 (批次 {i//chunk_size + 1}): {e}")
                all_success = False
                # raise Exception(f"Feishu Write Response Handling Failed in batch {i//chunk_size + 1}: {e}")
        
        return all_success

# 可以在这里添加一些简单的测试代码，但通常在main.py中进行集成测试
if __name__ == '__main__':
    # 仅作示例，请勿在生产环境直接硬编码敏感信息
    # 请替换为您的实际飞书应用信息
    # 配置日志，用于独立测试
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

    print("--- 飞书客户端独立测试 (请确保替换YOUR_...) ---")
    TEST_APP_ID = "YOUR_FEISHU_APP_ID"
    TEST_APP_SECRET = "YOUR_FEISHU_APP_SECRET"
    TEST_BASE_ID = "YOUR_FEISHU_BASE_ID"
    TEST_TABLE_ID = "YOUR_FEISHU_TABLE_ID"
    TEST_INPUT_FIELD = "测试文本" # 确保您的表中有此字段
    TEST_OUTPUT_FIELD = "测试输出" # 确保您的表中有此字段

    if "YOUR_" in TEST_APP_ID:
        logging.warning("请替换 feishu_client.py 中的 YOUR_ 占位符为您的实际飞书应用信息以运行测试。")
    else:
        try:
            feishu_client = FeishuClient(TEST_APP_ID, TEST_APP_SECRET)
            
            # 测试读取
            print("\n--- 测试读取记录 ---")
            read_records_data = feishu_client.read_records(TEST_BASE_ID, TEST_TABLE_ID, [TEST_INPUT_FIELD, TEST_OUTPUT_FIELD], page_size=2)
            for record in read_records_data:
                print(f"Record ID: {record.get('record_id')}, Fields: {record.get('fields')}")

            # 测试写入/更新 (仅更新第一条记录的某个字段)
            if read_records_data:
                print("\n--- 测试更新记录 ---")
                first_record_id = read_records_data[0]['record_id']
                update_payload = [{
                    "record_id": first_record_id,
                    "fields": {TEST_OUTPUT_FIELD: f"Updated by test at {time.strftime('%Y-%m-%d %H:%M:%S')}"}
                }]
                update_success = feishu_client.update_records(TEST_BASE_ID, TEST_TABLE_ID, update_payload)
                print(f"更新操作是否成功: {update_success}")
            else:
                print("\n没有可读取的记录，跳过更新测试。")

        except Exception as e:
            logging.error(f"飞书客户端独立测试失败: {e}")
    print("--- 飞书客户端独立测试结束 ---")