影刀RPA与飞书多维表格：自动化数据录入与更新

作者：林焱

飞书多维表格（Bitable）是很多团队的协作核心——项目进度、客户管理、数据统计都在里面。但手动维护多维表格数据效率极低，特别是需要从其他系统同步数据的时候。影刀RPA可以通过飞书开放API实现多维表格的自动化读写，这篇文章从API对接到实战场景，把飞书多维表格的自动化操作讲透。

---

一、飞书多维表格API基础

1.1 准备工作

前置条件：
1. 创建飞书应用 → https://open.feishu.cn/app
2. 获取 App ID 和 App Secret
3. 添加权限：bitable:app、bitable:app:readonly
4. 获取多维表格的 app_token 和 table_id
   - app_token：从URL中获取，如 https://xxx.feishu.cn/base/XXXXXX 中的 XXXXXX
   - table_id：通过API获取或从URL中获取

1.2 认证与Token管理

import requests
import json
import time

class FeishuBitable:
    """飞书多维表格操作类"""
    
    def __init__(self, app_id, app_secret):
        self.app_id = app_id
        self.app_secret = app_secret
        self.tenant_token = None
        self.token_expire = 0
        self.base_url = "https://open.feishu.cn/open-apis"
    
    def _ensure_token(self):
        """确保token有效"""
        if self.tenant_token and time.time() < self.token_expire:
            return
        
        url = f"{self.base_url}/auth/v3/tenant_access_token/internal"
        response = requests.post(url, json={
            "app_id": self.app_id,
            "app_secret": self.app_secret
        })
        
        result = response.json()
        self.tenant_token = result.get("tenant_access_token")
        self.token_expire = time.time() + result.get("expire", 7200) - 300
    
    def _request(self, method, path, **kwargs):
        """发送API请求"""
        self._ensure_token()
        
        url = f"{self.base_url}{path}"
        headers = {
            "Authorization": f"Bearer {self.tenant_token}",
            "Content-Type": "application/json"
        }
        
        response = requests.request(
            method, url, headers=headers, **kwargs
        )
        
        result = response.json()
        
        if result.get("code") != 0:
            print(f"API错误：{result.get('msg')}")
        
        return result
    
    # ===== 表格信息 =====
    
    def list_tables(self, app_token):
        """获取多维表格的所有数据表"""
        return self._request("GET", f"/bitable/v1/apps/{app_token}/tables")
    
    def get_table_fields(self, app_token, table_id):
        """获取表的字段信息"""
        return self._request("GET", f"/bitable/v1/apps/{app_token}/tables/{table_id}/fields")
    
    # ===== 记录操作 =====
    
    def list_records(self, app_token, table_id, page_size=100, filter_expr=None, sort=None):
        """获取记录列表"""
        params = {"page_size": page_size}
        
        if filter_expr:
            params["filter"] = filter_expr
        if sort:
            params["sort"] = sort
        
        all_records = []
        page_token = None
        
        while True:
            if page_token:
                params["page_token"] = page_token
            
            result = self._request(
                "GET",
                f"/bitable/v1/apps/{app_token}/tables/{table_id}/records",
                params=params
            )
            
            if result.get("code") != 0:
                break
            
            records = result.get("data", {}).get("items", [])
            all_records.extend(records)
            
            page_token = result.get("data", {}).get("page_token")
            if not page_token or not result.get("data", {}).get("has_more"):
                break
        
        return all_records
    
    def create_record(self, app_token, table_id, fields):
        """创建单条记录"""
        return self._request(
            "POST",
            f"/bitable/v1/apps/{app_token}/tables/{table_id}/records",
            json={"fields": fields}
        )
    
    def batch_create_records(self, app_token, table_id, records):
        """批量创建记录"""
        return self._request(
            "POST",
            f"/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_create",
            json={"records": [{"fields": r} for r in records]}
        )
    
    def update_record(self, app_token, table_id, record_id, fields):
        """更新单条记录"""
        return self._request(
            "PUT",
            f"/bitable/v1/apps/{app_token}/tables/{table_id}/records/{record_id}",
            json={"fields": fields}
        )
    
    def batch_update_records(self, app_token, table_id, updates):
        """批量更新记录"""
        return self._request(
            "POST",
            f"/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_update",
            json={"records": updates}
        )
    
    def delete_record(self, app_token, table_id, record_id):
        """删除单条记录"""
        return self._request(
            "DELETE",
            f"/bitable/v1/apps/{app_token}/tables/{table_id}/records/{record_id}"
        )

---

二、数据录入实战

2.1 从Excel批量导入

import openpyxl

def Excel导入多维表格(bitable, app_token, table_id, excel_path, field_mapping):
    """将Excel数据导入飞书多维表格"""
    
    # 1. 读取Excel
    wb = openpyxl.load_workbook(excel_path)
    ws = wb.active
    
    headers = [cell.value for cell in ws[1]]
    rows = []
    for row in ws.iter_rows(min_row=2, values_only=True):
        row_dict = dict(zip(headers, row))
        rows.append(row_dict)
    
    # 2. 字段映射与格式转换
    records = []
    for row in rows:
        fields = {}
        for excel_col, bitable_field in field_mapping.items():
            value = row.get(excel_col)
            
            if value is None:
                continue
            
            # 根据多维表格字段类型转换
            field_type = bitable_field.get("type", "text")
            
            if field_type == "text":
                fields[bitable_field["name"]] = str(value)
            elif field_type == "number":
                fields[bitable_field["name"]] = float(value) if value else 0
            elif field_type == "date":
                fields[bitable_field["name"]] = str(value)
            elif field_type == "select":
                fields[bitable_field["name"]] = str(value)
            else:
                fields[bitable_field["name"]] = str(value)
        
        records.append(fields)
    
    # 3. 批量写入（每批最多500条）
    batch_size = 500
    total = len(records)
    success = 0
    
    for i in range(0, total, batch_size):
        batch = records[i:i + batch_size]
        result = bitable.batch_create_records(app_token, table_id, batch)
        
        if result.get("code") == 0:
            success += len(batch)
            print(f"已导入{success}/{total}条")
        else:
            print(f"第{i//batch_size + 1}批导入失败：{result.get('msg')}")
    
    print(f"导入完成：成功{success}条，共{total}条")
    return success

2.2 采集数据实时写入

def 采集并写入飞书(keyword, bitable, app_token, table_id):
    """采集数据后实时写入飞书多维表格"""
    
    # 采集
    items = 执行采集(keyword)
    
    # 转换为飞书字段格式
    for item in items:
        fields = {
            "商品名称": item.get("name", ""),
            "价格": item.get("price", 0),
            "链接": {"link": item.get("url", ""), "text": "查看商品"},
            ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/0b2ae664480b4c9995d62f0b2955503d.png#pic_center)

            "分类": item.get("category", "未分类"),
            "采集时间": datetime.now().strftime("%Y-%m-%d %H:%M"),
        }
        
        result = bitable.create_record(app_token, table_id, fields)
        
        if result.get("code") == 0:
            print(f"✅ 写入成功：{item.get('name')}")
        else:
            print(f"❌ 写入失败：{result.get('msg')}")

---

三、数据更新实战

3.1 增量更新（基于唯一标识）

def 增量更新(bitable, app_token, table_id, new_data, unique_field="商品ID"):
    """增量更新：新数据与飞书数据对比，新增/更新/跳过"""
    
    # 1. 获取飞书现有数据
    existing = bitable.list_records(app_token, table_id)
    
    # 构建索引：unique_field → record_id
    existing_map = {}
    for record in existing:
        fields = record.get("fields", {})
        key = fields.get(unique_field, "")
        existing_map[key] = record.get("record_id")
    
    # 2. 对比并更新
    stats = {"created": 0, "updated": 0, "skipped": 0}
    
    for item in new_data:
        key = item.get(unique_field, "")
        
        if key in existing_map:
            # 已存在，检查是否需要更新
            record_id = existing_map[key]
            
            # 构建更新字段
            update_fields = {}
            for field_name, value in item.items():
                if field_name != unique_field:
                    update_fields[field_name] = value
            
            result = bitable.update_record(app_token, table_id, record_id, update_fields)
            
            if result.get("code") == 0:
                stats["updated"] += 1
            else:
                print(f"更新失败：{key} - {result.get('msg')}")
        else:
            # 不存在，新增
            result = bitable.create_record(app_token, table_id, item)
            
            if result.get("code") == 0:
                stats["created"] += 1
            else:
                print(f"新增失败：{key} - {result.get('msg')}")
    
    print(f"增量更新完成：新增{stats['created']}条，更新{stats['updated']}条，跳过{stats['skipped']}条")
    return stats

3.2 价格监控更新

def 价格监控更新(bitable, app_token, table_id, price_field="价格"):
    """定期更新价格数据，记录变动"""
    
    # 1. 获取飞书中的商品列表
    records = bitable.list_records(app_token, table_id)
    
    for record in records:
        fields = record.get("fields", {})
        record_id = record.get("record_id")
        
        product_name = fields.get("商品名称", "")
        old_price = fields.get(price_field, 0)
        product_url = fields.get("链接", {}).get("link", "")
        
        if not product_url:
            continue
        
        # 2. 采集最新价格
        try:
            new_price = 采集最新价格(product_url)
            
            if new_price and new_price != old_price:
                # 3. 更新价格和变动记录
                change = new_price - old_price
                change_pct = (change / old_price * 100) if old_price > 0 else 0
                
                update_fields = {
                    price_field: new_price,
                    "上次价格": old_price,
                    "价格变动": f"{'+' if change > 0 else ''}{change:.2f} ({change_pct:+.1f}%)",
                    "更新时间": datetime.now().strftime("%Y-%m-%d %H:%M"),
                }
                
                bitable.update_record(app_token, table_id, record_id, update_fields)
                
                # 大幅变动发通知
                if abs(change_pct) > 10:
                    发送企微通知(f"⚠️ {product_name} 价格变动{change_pct:+.1f}%，从¥{old_price}变为¥{new_price}")
                
        except Exception as e:
            print(f"采集{product_name}价格失败：{str(e)}")

---

四、数据查询与筛选

4.1 条件查询

def 查询低库存商品(bitable, app_token, table_id):
    """查询库存低于安全线的商品"""
    
    # 构建筛选条件
    filter_expr = json.dumps({
        "conjunction": "and",
        "conditions": [
            {
                "field_name": "库存",
                "operator": "isLess",
                "value": ["100"]  # 库存<100
            }
        ]
    })
    
    records = bitable.list_records(
        app_token, table_id,
        filter_expr=filter_expr,
        sort='{"field_name":"库存","desc":false}'
    )
    
    return records

4.2 数据统计

def 统计分类数据(bitable, app_token, table_id):
    """统计各分类的商品数量和平均价格"""
    from collections import defaultdict
    
    records = bitable.list_records(app_token, table_id)
    
    stats = defaultdict(lambda: {"count": 0, "total_price": 0})
    
    for record in records:
        fields = record.get("fields", {})
        category = fields.get("分类", "未分类")
        price = fields.get("价格", 0)
        
        stats[category]["count"] += 1
        stats[category]["total_price"] += price
    
    # 输出统计结果
    print("分类统计：")
    for category, data in stats.items():
        avg_price = data["total_price"] / data["count"] if data["count"] > 0 else 0
        print(f"  {category}：{data['count']}个商品，均价¥{avg_price:,.0f}")
    
    return dict(stats)

---

五、自动化流程集成

5.1 定时同步流水线

def 定时同步飞书数据():
    """每天定时从ERP同步数据到飞书多维表格"""
    
    # 初始化
    bitable = FeishuBitable(
        app_id=os.getenv("FEISHU_APP_ID"),
        app_secret=os.getenv("FEISHU_APP_SECRET")
    )
    
    app_token = "your_app_token"
    table_id = "your_table_id"
    
    # 1. 从ERP获取最新数据
    erp_data = 从ERP获取商品数据()
    
    # 2. 增量更新到飞书
    ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/b5695d4875c04f47925b26ffdbf9d65e.png#pic_center)

    stats = 增量更新(bitable, app_token, table_id, erp_data, unique_field="商品ID")
    
    # 3. 更新价格
    价格监控更新(bitable, app_token, table_id)
    
    # 4. 发送同步报告
    发送企微通知(f"""📊 飞书数据同步完成
新增：{stats['created']}条
更新：{stats['updated']}条
同步时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}""")

# 每天早上9点执行
import schedule
schedule.every().day.at("09:00").do(定时同步飞书数据)

---

六、常见问题

问题	原因	解决方案
权限不足	应用未添加bitable权限	在飞书开放平台添加权限
字段类型不匹配	写入数据类型与字段定义不一致	按字段类型转换数据
批量写入超时	一次写入数据过多	分批写入，每批500条
Token过期	tenant_access_token有效期2小时	自动刷新Token
限流	API调用频率限制	控制请求频率，添加重试

---

总结

飞书多维表格自动化的核心流程：

环节	操作	API
数据录入	从Excel/API批量导入	batch_create_records
数据更新	增量对比+更新	update_record / batch_update
数据查询	条件筛选+排序	list_records + filter
数据同步	定时从外部系统同步	全流程自动化
通知推送	关键变动实时通知	企微/飞书消息

飞书多维表格API的坑在于字段类型——不同类型（文本/数字/日期/链接/人员）的数据格式要求不同，写入前一定要先获取字段信息，按类型转换。

---

作者：林焱 | 觉得有用就收藏，后续分享更多影刀RPA实战技巧