飞书多维表格自动备份与清空

读取 A 表的全部记录 → 逐批写入 B 表 → 清空 A 表

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 飞书多维表格 通用备份清空工具 功能：将源表全部记录复制到目标表，然后清空源表 只需配置下方的 APP 凭证和表格 ID 即可运行 """ import time import requests # ============================================================ # 飞书配置（请改为你自己的信息） # ============================================================ FEISHU_APP_ID = "cli_xxxxxxxxxxxxx" # 替换为你的 App ID FEISHU_APP_SECRET = "your_app_secret" # 替换为你的 App Secret APP_TOKEN = "YOUR_APP_TOKEN" # 多维表格的 app_token SOURCE_TABLE_ID = "tblxxxxxxxxxxxxx" # 源表（将被复制并清空） TARGET_TABLE_ID = "tblyyyyyyyyyyyyy" # 目标表（备份目的地） # ============================================================ # 飞书 API 封装 # ============================================================ class FeishuBitable: def __init__(self, app_id, app_secret): self.app_id = app_id self.app_secret = app_secret self._token = None self._token_expire = 0 def _get_token(self): if self._token and time.time() < self._token_expire: return self._token url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" resp = requests.post(url, json={"app_id": self.app_id, "app_secret": self.app_secret}, timeout=15) if resp.status_code != 200: raise Exception(f"获取 token 失败，HTTP {resp.status_code}") result = resp.json() if result.get("code") != 0: raise Exception(f"获取 token 失败: {result}") self._token = result["tenant_access_token"] self._token_expire = time.time() + 7000 print("已获取 tenant_access_token") return self._token def get_all_records(self, table_id): """获取表中所有记录，返回原始 fields 字典列表（可直接用于创建）""" token = self._get_token() records = [] page_token = None while True: url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{table_id}/records/search" headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} body = {"page_size": 500} if page_token: body["page_token"] = page_token resp = requests.post(url, headers=headers, json=body, timeout=30) result = resp.json() if result.get("code") != 0: raise Exception(f"查询记录失败: {result}") data = result.get("data", {}) for item in data.get("items", []): records.append(item.get("fields", {})) if not data.get("has_more"): break page_token = data.get("page_token") print(f"从源表读取到 {len(records)} 条记录") return records def batch_create(self, table_id, records): """批量写入记录，records 为 fields 字典列表""" if not records: return token = self._get_token() url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{table_id}/records/batch_create" headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} # 一次最多创建 500 条，这里按 500 分批 for i in range(0, len(records), 500): batch = records[i:i+500] body = {"records": [{"fields": rec} for rec in batch]} resp = requests.post(url, headers=headers, json=body, timeout=60) result = resp.json() if result.get("code") != 0: raise Exception(f"批量写入失败: {result}") print(f" 已写入 {min(i+500, len(records))}/{len(records)} 条") print(f"成功备份全部 {len(records)} 条记录到目标表") def delete_all_records(self, table_id): """删除表中所有记录""" token = self._get_token() # 第一步：获取所有记录 ID record_ids = [] page_token = None while True: url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{table_id}/records/search" headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} body = {"page_size": 500} if page_token: body["page_token"] = page_token resp = requests.post(url, headers=headers, json=body, timeout=30) result = resp.json() if result.get("code") != 0: raise Exception(f"查询记录ID失败: {result}") data = result.get("data", {}) for item in data.get("items", []): record_ids.append(item["record_id"]) if not data.get("has_more"): break page_token = data.get("page_token") if not record_ids: print("源表已是空表，无需删除") return print(f"📋 共 {len(record_ids)} 条待删除") # 第二步：调用 batch_delete 删除 for i in range(0, len(record_ids), 500): batch = record_ids[i:i+500] del_url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{table_id}/records/batch_delete" headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} payload = {"records": batch} resp = requests.post(del_url, headers=headers, json=payload, timeout=30) result = resp.json() if result.get("code") != 0: raise Exception(f"删除失败: {result}") print(f" 已删除 {min(i+500, len(record_ids))}/{len(record_ids)} 条") print("源表已清空") # ============================================================ # 主流程：备份 + 清空 # ============================================================ def backup_and_clear(): print("=" * 50) print("开始执行：源表 → 目标表 备份 & 源表清空") feishu = FeishuBitable(FEISHU_APP_ID, FEISHU_APP_SECRET) # 1. 读取源表所有记录 records = feishu.get_all_records(SOURCE_TABLE_ID) if not records: print("源表无数据，跳过备份步骤") else: # 2. 写入目标表 feishu.batch_create(TARGET_TABLE_ID, records) # 3. 清空源表 feishu.delete_all_records(SOURCE_TABLE_ID) print("=" * 50) print("任务完成！") if __name__ == "__main__": backup_and_clear()