pip install lark_oapi loguru

import osapp_id = os.getenv("FEISHU_APP_ID")  # 从环境变量读取app_secret = os.getenv("FEISHU_APP_SECRET")

def create_table(table_name):    # 1. 创建客户端    client = lark.Client.builder() \            .app_id(app_id) \            .app_secret(app_secret) \            .build()    # 2. 构造请求    request = CreateAppRequest.builder() \            .request_body(ReqApp.builder().name(table_name).build()) \            .build()    # 3. 发送请求    response = client.bitable.v1.app.create(request)    return response.data.app.url  # 返回表格URL

def delete_default_fields():    default_fields = ["文本", "单选", "日期"]  # 系统默认字段    for field in get_fields():        if field['name'] in default_fields:            delete_field(field['id'])  # 逐个删除

// feishu_config.json{  "table_name": "运营数据表",  "url": "https://example.com",  "fields": ["日期", "UV", "PV"]}

def add_record(data):    record = AppTableRecord.builder() \            .fields({                "标题": data['title'],                "阅读量": data['views']            }).build()    response = client.bitable.v1.app_table_record.create(record)    print("写入成功！" if response.success() else "写入失败")

def batch_add_records(data_list):    # 分批次写入（每次100条）    for i in range(0, len(data_list), 100):        batch = data_list[i:i+100]        request = BatchCreateAppTableRecordRequest.builder() \                .records(batch).build()        client.bitable.v1.app_table_record.batch_create(request)

def clean_empty_records():    # 获取所有空记录ID    empty_ids = [r['id'] for r in get_records() if not r['fields']]    # 批量删除    for id in empty_ids:        client.bitable.v1.app_table_record.delete(id)

from tenacity import retry, stop_after_attempt@retry(stop=stop_after_attempt(3))  # 失败自动重试3次def safe_write(data):    write_one_record(data)

if __name__ == '__main__':    # 1. 初始化表格    init_table("小红书数据",               fields=["笔记ID", "点赞数", "评论数"])    # 2. 模拟采集数据    notes = [        {"笔记ID": "123", "点赞数": 1000},        {"笔记ID": "456", "点赞数": 2000}    ]    # 3. 批量写入    batch_write_records(notes)    # 4. 结果检查    print(f"成功写入{len(notes)}条数据！")

import jsonimport osimport sysimport tracebackfrom pathlib import Pathfrom loguru import logger as logtry:    import lark_oapi as lark    from lark_oapi.api.bitable.v1 import *except ImportError:    log.info("错误: 未安装lark_oapi库，请使用 'pip install lark_oapi' 安装")    sys.exit(1)# 飞书应用凭证app_id = ""app_secret = ""# 配置文件路径CONFIG_FILE = "feishu_config.json"# 启用详细日志DEBUG = Truedef debug_log(message):    """输出日志信息"""    if DEBUG:        log.info(f"[DEBUG] {message}")def load_config():    """    加载配置文件    返回:    dict: 配置信息，如果配置文件不存在则返回None    """    debug_log(f"尝试加载配置文件: {CONFIG_FILE}")    if os.path.exists(CONFIG_FILE):        try:            with open(CONFIG_FILE, 'r', encoding='utf-8') as f:                config = json.load(f)                log.info(f"已加载配置文件，使用已有表格: {config['table_name']}")                debug_log(f"配置内容: {json.dumps(config, ensure_ascii=False)}")                return config        except json.JSONDecodeError as e:            log.info(f"配置文件格式错误: {e}")            debug_log(f"JSON解析错误: {str(e)}")            return None        except Exception as e:            log.info(f"读取配置文件失败: {e}")            debug_log(f"异常详情: {traceback.format_exc()}")            return None    else:        debug_log(f"配置文件不存在: {CONFIG_FILE}")    return Nonedef save_config(table_name, app_token, url, default_table_id, fields=None):    """    保存配置到文件    参数:    table_name (str): 表格名称    app_token (str): 应用token    url (str): 表格URL    default_table_id (str): 默认表格ID    fields (list): 已创建的字段列表    """    config = {        "table_name": table_name,        "app_token": app_token,        "url": url,        "default_table_id": default_table_id,        "fields": fields or []    }    debug_log(f"准备保存配置: {json.dumps(config, ensure_ascii=False)}")    try:        with open(CONFIG_FILE, 'w', encoding='utf-8') as f:            json.dump(config, f, ensure_ascii=False, indent=4)        log.info(f"配置已保存到 {CONFIG_FILE}")        return True    except Exception as e:        log.info(f"保存配置文件失败: {e}")        debug_log(f"保存配置异常: {traceback.format_exc()}")        return Falsedef create_table(table_name):    """    创建飞书多维表格    参数:    table_name (str): 表格名称    返回:    tuple: (app_token, url, default_table_id) 或 None(如果创建失败)    """    debug_log(f"开始创建表格: {table_name}")    try:        # 创建client        debug_log("初始化飞书客户端")        client = lark.Client.builder() \            .app_id(app_id) \            .app_secret(app_secret) \            .log_level(lark.LogLevel.DEBUG) \            .build()        # 构造请求对象        debug_log("构造创建表格请求")        request: CreateAppRequest = CreateAppRequest.builder() \            .request_body(ReqApp.builder()                .name(table_name)                .build()) \            .build()        # 发起请求        debug_log("发送创建表格请求")        response: CreateAppResponse = client.bitable.v1.app.create(request)        # 处理业务结果        debug_log("处理响应结果")        data = lark.JSON.marshal(response.data, indent=4)        debug_log(f"原始响应数据: {data}")        data = json.loads(data)        app_token = data['app']['app_token']        url = data['app']['url']        default_table_id = data['app']['default_table_id']        log.info(f"成功创建表格: {table_name}")        log.info(f"访问URL: {url}")        return app_token, url, default_table_id    except Exception as e:        log.info(f"创建表格失败: {e}")        debug_log(f"创建表格异常: {traceback.format_exc()}")        return Nonedef create_head(app_token, table_id, head_info):    """    创建表头字段    参数:    app_token (str): 应用token    table_id (str): 表格ID    head_info (str): 表头名称    返回:    bool: 是否创建成功    """    try:        # 创建client        client = lark.Client.builder() \            .app_id(app_id) \            .app_secret(app_secret) \            .log_level(lark.LogLevel.DEBUG) \            .build()        # 构造请求对象        request: CreateAppTableFieldRequest = CreateAppTableFieldRequest.builder() \            .app_token(app_token) \            .table_id(table_id) \            .request_body(AppTableField.builder()                        .field_name(head_info)                        .type(1)                        .build()) \            .build()        # 发起请求        response: CreateAppTableFieldResponse = client.bitable.v1.app_table_field.create(request)        log.info(f"成功创建表头: {head_info}")        return True    except Exception as e:        log.info(f"创建表头失败: {e}")        debug_log(f"创建表头异常: {traceback.format_exc()}")        return Falsedef get_table_fields_with_details(app_token, table_id):    """    获取表格已有字段的详细信息    参数:    app_token (str): 应用token    table_id (str): 表格ID    返回:    list: 字段详细信息列表    """    try:        # 创建client        client = lark.Client.builder() \            .app_id(app_id) \            .app_secret(app_secret) \            .log_level(lark.LogLevel.DEBUG) \            .build()        # 构造请求对象        request: ListAppTableFieldRequest = ListAppTableFieldRequest.builder() \            .app_token(app_token) \            .table_id(table_id) \            .build()        # 发起请求        response: ListAppTableFieldResponse = client.bitable.v1.app_table_field.list(request)        # 处理业务结果        data = lark.JSON.marshal(response.data, indent=4)        debug_log(f"获取字段详情响应: {data}")        data = json.loads(data)        fields = []        for item in data.get('items', []):            fields.append({                'field_id': item.get('field_id'),                'field_name': item.get('field_name'),                'is_primary': item.get('is_primary', False),                'type': item.get('type'),                'ui_type': item.get('ui_type')            })        log.info(f"获取到已有字段: {len(fields)}个")        return fields    except Exception as e:        log.info(f"获取表格字段失败: {e}")        debug_log(f"获取表格字段异常: {traceback.format_exc()}")        return []def get_table_fields(app_token, table_id):    """    获取表格已有字段名称列表    参数:    app_token (str): 应用token    table_id (str): 表格ID    返回:    list: 字段名称列表    """    fields_with_details = get_table_fields_with_details(app_token, table_id)    return [field['field_name'] for field in fields_with_details]def delete_field(app_token, table_id, field_id):    """    删除表格字段    参数:    app_token (str): 应用token    table_id (str): 表格ID    field_id (str): 字段ID    返回:    bool: 是否删除成功    """    try:        # 创建client        client = lark.Client.builder() \            .app_id(app_id) \            .app_secret(app_secret) \            .log_level(lark.LogLevel.DEBUG) \            .build()        # 构造请求对象        request: DeleteAppTableFieldRequest = DeleteAppTableFieldRequest.builder() \            .app_token(app_token) \            .table_id(table_id) \            .field_id(field_id) \            .build()        # 发起请求        response: DeleteAppTableFieldResponse = client.bitable.v1.app_table_field.delete(request)        if response.success():            log.info(f"成功删除字段ID: {field_id}")            return True        else:            log.info(f"删除字段失败: {response.msg}")            debug_log(f"删除字段失败响应: {response.code}, {response.msg}")            return False    except Exception as e:        log.info(f"删除字段异常: {e}")        debug_log(f"删除字段异常: {traceback.format_exc()}")        return Falsedef delete_default_fields(app_token, table_id):    """    删除默认生成的前四个字段（文本、单选、日期、附件）    参数:    app_token (str): 应用token    table_id (str): 表格ID    返回:    bool: 是否全部删除成功    """    log.info("开始删除默认字段...")    # 获取所有字段详情    fields = get_table_fields_with_details(app_token, table_id)    # 默认字段名称    default_field_names = ["文本", "单选", "日期", "附件"]    # 找出需要删除的字段ID    fields_to_delete = []    for field in fields:        if field['field_name'] in default_field_names:            # 检查是否是主键字段            if field.get('is_primary', False):                log.info(f"警告: 字段 '{field['field_name']}' 是主键字段，无法删除")                continue            fields_to_delete.append(field)    if not fields_to_delete:        log.info("没有找到需要删除的默认字段")        return True    # 删除字段    success = True    for field in fields_to_delete:        log.info(f"尝试删除字段: {field['field_name']} (ID: {field['field_id']})")        if not delete_field(app_token, table_id, field['field_id']):            success = False    return successdef init_table(table_name, required_fields, delete_defaults=True):    """    初始化表格：创建表格、添加字段、删除默认字段    参数:    table_name (str): 表格名称    required_fields (list): 需要创建的字段列表    delete_defaults (bool): 是否删除默认字段    返回:    tuple: (成功标志, 表格URL)    """    try:        # 尝试加载配置        config = load_config()        if config:            # 使用已有表格            app_token = config["app_token"]            url = config["url"]            default_table_id = config["default_table_id"]            existing_fields = config.get("fields", [])            # 获取最新的字段列表            current_fields = get_table_fields(app_token, default_table_id)            if current_fields:                existing_fields = current_fields            # 检查并添加缺失的字段            for field in required_fields:                if field not in existing_fields:                    if create_head(app_token, default_table_id, field):                        existing_fields.append(field)            # 更新配置文件中的字段列表            save_config(table_name, app_token, url, default_table_id, existing_fields)        else:            # 创建新表格            result = create_table(table_name)            if not result:                log.info("创建表格失败")                return False, None            app_token, url, default_table_id = result            existing_fields = []            # 创建所有需要的字段            for field in required_fields:                if create_head(app_token, default_table_id, field):                    existing_fields.append(field)            # 删除默认字段            if delete_defaults:                delete_default_fields(app_token, default_table_id)            # 保存配置            save_config(table_name, app_token, url, default_table_id, existing_fields)        log.info(f"表格设置完成，可通过以下URL访问: {url}")        return True, url    except Exception as e:        log.info(f"初始化表格失败: {e}")        debug_log(f"初始化表格异常: {traceback.format_exc()}")        return False, Nonedef main(table_name,required_fields):    """主函数，处理命令行参数并执行表格初始化"""    import argparse    parser = argparse.ArgumentParser(description='飞书多维表格管理工具')    parser.add_argument('--table-name', type=str, help='表格名称')    parser.add_argument('--fields', type=str, help='字段列表，用逗号分隔')    parser.add_argument('--keep-defaults', action='store_true', help='保留默认字段')    args = parser.parse_args()    # 如果没有提供命令行参数，使用默认值    # table_name = args.table_name or "小红书关键词每日采集入库"    # if args.fields:    #     required_fields = [field.strip() for field in args.fields.split(',')]    # else:    #     required_fields = [ "note_id", "user_id", "nick_name", "display_title",    #                       "shared_count", "liked_count", "comment_count", "collected_count",    #                       "note_link", "content"]    delete_defaults = not args.keep_defaults    log.info(f"开始初始化表格: {table_name}")    log.info(f"字段列表: {', '.join(required_fields)}")    log.info(f"是否删除默认字段: {delete_defaults}")    success, url = init_table(table_name, required_fields, delete_defaults)    if success:        log.info("表格初始化成功!")        if url:            log.info(f"表格URL: {url}")    else:        log.info("表格初始化失败!")        sys.exit(1)def init_create_table(table_name,required_fields,clear_mode='all'):    """初始化表格并清理空记录    参数:    table_name (str): 表格名称    required_fields (list): 需要创建的字段列表    clear_mode (str): 清理模式，'all'表示清理所有记录，'none'表示只清理空记录    返回:    无返回值，但会打印清理结果日志    """    try:        # 调用主函数初始化表格        main(table_name,required_fields)    except KeyboardInterrupt:        # 处理用户中断        log.info("\n程序被用户中断")        sys.exit(0)    except Exception as e:        # 处理其他异常        log.info(f"程序执行出错: {e}")        debug_log(f"异常详情: {traceback.format_exc()}")    else:        # 清理空记录        deleted, total = clear_init_null_record(clear_mode=clear_mode)        log.info(f"清理完成: 共 {total} 条记录，删除了 {deleted} 条记录")#写入单条数据def write_one_record(data_dict):    """    将单条记录写入飞书多维表格    参数:    data_dict (dict): 字段名和值的字典，例如 {"note_id": "123", "user_id": "456"}    返回:    bool: 是否写入成功    """    debug_log(f"准备写入数据: {json.dumps(data_dict, ensure_ascii=False)}")    try:        # 加载配置        config = load_config()        if not config:            log.info("错误: 未找到配置文件，请先初始化表格")            return False        app_token = config["app_token"]        table_id = config["default_table_id"]        # 创建client        client = lark.Client.builder() \            .app_id(app_id) \            .app_secret(app_secret) \            .log_level(lark.LogLevel.DEBUG) \            .build()        # 构造请求对象        request: CreateAppTableRecordRequest = CreateAppTableRecordRequest.builder() \            .app_token(app_token) \            .table_id(table_id) \            .request_body(AppTableRecord.builder()                        .fields(data_dict)                        .build()) \            .build()        # 发起请求        response: CreateAppTableRecordResponse = client.bitable.v1.app_table_record.create(request)        # 处理响应        if response.success():            data = lark.JSON.marshal(response.data, indent=4)            debug_log(f"写入数据成功，响应: {data}")            log.info("数据写入成功")            return True        else:            log.info(f"写入数据失败: {response.msg}")            debug_log(f"写入数据失败: 代码={response.code}, 消息={response.msg}")            return False    except Exception as e:        log.info(f"写入数据异常: {e}")        debug_log(f"写入数据异常: {traceback.format_exc()}")        return False# 写入多条数据def batch_write_records(data_list):    """    批量写入多条记录到飞书多维表格    参数:    data_list (list): 字典列表，每个字典包含一条记录的字段名和值                     例如 [{"note_id": "123"}, {"note_id": "456"}]    返回:    bool: 是否全部写入成功    """    debug_log(f"准备批量写入 {len(data_list)} 条数据")    try:        # 加载配置        config = load_config()        if not config:            log.info("错误: 未找到配置文件，请先初始化表格")            return False        app_token = config["app_token"]        table_id = config["default_table_id"]        # 创建client        client = lark.Client.builder() \            .app_id(app_id) \            .app_secret(app_secret) \            .log_level(lark.LogLevel.DEBUG) \            .build()        # 准备记录列表        records = []        for data_dict in data_list:            records.append(AppTableRecord.builder().fields(data_dict).build())        # 构造请求对象        request: BatchCreateAppTableRecordRequest = BatchCreateAppTableRecordRequest.builder() \            .app_token(app_token) \            .table_id(table_id) \            .request_body(BatchCreateAppTableRecordRequestBody.builder()                        .records(records)                        .build()) \            .build()        # 发起请求        response: BatchCreateAppTableRecordResponse = client.bitable.v1.app_table_record.batch_create(request)        # 处理响应        if response.success():            data = lark.JSON.marshal(response.data, indent=4)            debug_log(f"批量写入数据成功，响应: {data}")            log.info(f"成功批量写入 {len(data_list)} 条数据")            return True        else:            log.info(f"批量写入数据失败: {response.msg}")            debug_log(f"批量写入数据失败: 代码={response.code}, 消息={response.msg}")            return False    except Exception as e:        log.info(f"批量写入数据异常: {e}")        debug_log(f"批量写入数据异常: {traceback.format_exc()}")        return Falsedef clear_init_null_record(clear_mode='all'):    """    查询并删除表格中的空记录    返回:    tuple: (成功删除的记录数, 总记录数)    """    try:        # 加载配置        config = load_config()        if not config:            log.info("错误: 未找到配置文件，请先初始化表格")            return 0, 0        app_token = config["app_token"]        table_id = config["default_table_id"]        # 创建client        client = lark.Client.builder() \            .app_id(app_id) \            .app_secret(app_secret) \            .log_level(lark.LogLevel.DEBUG) \            .build()        # 查询记录        log.info("开始查询表格记录...")        # 查询记录，一次最多查询100条        search_request: SearchAppTableRecordRequest = SearchAppTableRecordRequest.builder() \            .app_token(app_token) \            .table_id(table_id) \            .page_size(100) \            .request_body(SearchAppTableRecordRequestBody.builder().build()) \            .build()        search_response: SearchAppTableRecordResponse = client.bitable.v1.app_table_record.search(search_request)        if not search_response.success():            log.info(f"查询记录失败: {search_response.msg}")            debug_log(f"查询记录失败: 代码={search_response.code}, 消息={search_response.msg}")            return 0, 0        # 解析响应数据        data = lark.JSON.marshal(search_response.data, indent=4)        debug_log(f"查询记录响应: {data}")        data = json.loads(data)        records = data.get('items', [])        total_records = data.get('total', 0)        log.info(f"查询到 {total_records} 条记录")        # 找出空记录        null_records = []        for record in records:            if not record.get('fields') or len(record.get('fields')) == 0:                null_records.append(record.get('record_id'))        log.info(f"发现 {len(null_records)} 条空记录")        clear_records=[]        if clear_mode == 'all':            log.info('清空模式为全部清空')            clear_records=records        elif clear_mode == 'none':            log.info('清空模式为清空空数据')            clear_records=null_records        # 删除空记录        deleted_count = 0        for record_id in clear_records:            log.info(f"正在删除记录 ID: {record_id}")            delete_request: DeleteAppTableRecordRequest = DeleteAppTableRecordRequest.builder() \                .app_token(app_token) \                .table_id(table_id) \                .record_id(record_id) \                .build()            delete_response: DeleteAppTableRecordResponse = client.bitable.v1.app_table_record.delete(delete_request)            if delete_response.success():                deleted_count += 1                log.info(f"成功删除记录 ID: {record_id}")            else:                log.info(f"删除记录 {record_id} 失败: {delete_response.msg}")                debug_log(f"删除记录失败: 代码={delete_response.code}, 消息={delete_response.msg}")        log.info(f"共删除 {deleted_count} 条空记录")        return deleted_count, total_records    except Exception as e:        log.info(f"清理空记录异常: {e}")        debug_log(f"清理空记录异常: {traceback.format_exc()}")        return 0, 0# if __name__ == '__main__':#     table_name = "小红书关键词每日采集入库"#     required_fields = ["note_id", "user_id", "nick_name", "display_title",#                       "shared_count", "liked_count", "comment_count", "collected_count",#                       "note_link", "content"]    # 初始化表格    # init_create_table(table_name, required_fields)    # 清理空记录    # deleted, total = clear_init_null_record()    # print(f"清理完成: 共 {total} 条记录，删除了 {deleted} 条空记录")    # 写入单条记录示例    # sample_data = {    #     "note_id": "123456789",    #     "user_id": "user123",    #     "nick_name": "测试用户",    #     "display_title": "测试标题",    #     "shared_count": "10",    #     "liked_count": "20",    #     "comment_count": "5",    #     "collected_count": "3",    #     "note_link": "https://example.com/note/123456789",    #     "content": "这是一条测试内容"    # }    # write_one_record(sample_data)    # 批量写入记录示例    # sample_batch_data = [    #     {    #         "note_id": "123456789",    #         "user_id": "user123",    #         "nick_name": "测试用户1"    #     },    #     {    #         "note_id": "987654321",    #         "user_id": "user456",    #         "nick_name": "测试用户2"    #     }    # ]    # batch_write_records(sample_batch_data)