一、前言（业务痛点）

很多中小企业、跨境电商团队，大量业务数据（广告数据、库存数据、订单数据）都维护在飞书在线表格中。

传统方式存在巨大问题：

  • 人工导出 Excel 再导入 MySQL，重复劳动、效率极低

  • 数据更新不及时，报表、BI看板数据滞后

  • 没有日志、没有数据统计，不知道更新了多少条、是否成功

  • 报错无告警，出错没人知道

本文基于 Python + 飞书开放API + MySQL + 定时任务 + 飞书机器人推送，搭建一套企业级、可直接上线的飞书表格自动同步MySQL工程。

适配：Windows服务、bat一键执行、手动立即执行、定时每日同步

二、整体架构设计

2.1 工程能力

  • ✅ 读取飞书多维表格/在线表格数据

  • ✅ 自动清洗空值、空格、非法字符、日期兼容

  • ✅ 同步前清空原表，统计清空条数、新增条数

  • ✅ 全量覆盖、保证数据一致性

  • ✅ 支持 --run_once 立即执行 / 定时任务自动执行

  • ✅ 完整日志记录 + bat后台运行

  • ✅ 成功/异常 飞书机器人自动推送通知

  • ✅ 统一.env配置管理，代码零硬编码

2.2 目录结构（企业极简结构）

整套工程非常干净，无冗余文件，适合生产部署

ad_sync
 ├── main.py          # 程序入口、定时任务、参数控制
 ├── sync.py          # 核心同步业务逻辑
 ├── db.py            # 数据库统一封装
 ├── feishu.py        # 飞书读取+推送封装
 ├── config.py        # 任务配置、定时配置
 └── .env             # 数据库、飞书密钥

三、核心环境配置

3.1 依赖包

pip install pymysql python-dotenv requests apscheduler pandas

3.2 根目录 .env 配置

所有敏感信息统一管理，代码完全分离，安全、易维护。

# 数据库配置
COMPANY_DB_HOST=xxxx
COMPANY_DB_PORT=xxxx
COMPANY_DB_USER=xxxx
COMPANY_DB_PASSWORD=xxxx

# 飞书配置
FEISHU_APP_ID=xxxx
FEISHU_APP_SECRET=xxxx
FEISHU_BOT_WEBHOOK=xxxx

四、核心代码实现（可直接复用）

只贴核心通用代码，可直接用于自己的项目，脱敏无风险

4.1 数据库通用封装 db.py

统一连接、统一关闭、支持统计原表条数 + 清空表 + 批量插入

import os
import pymysql
from dotenv import load_dotenv

load_dotenv()

def get_db_conn():
    conn = pymysql.connect(
        host=os.getenv("COMPANY_DB_HOST"),
        port=int(os.getenv("COMPANY_DB_PORT")),
        user=os.getenv("COMPANY_DB_USER"),
        password=os.getenv("COMPANY_DB_PASSWORD"),
        charset="utf8mb4"
    )
    return conn

# 获取原表条数
def get_table_count(db_name, table_name):
    conn = get_db_conn()
    cur = conn.cursor()
    cur.execute(f"SELECT COUNT(*) FROM {db_name}.{table_name}")
    cnt = cur.fetchone()[0]
    cur.close()
    conn.close()
    return cnt

# 清空表
def truncate_table(db_name, table_name):
    conn = get_db_conn()
    cur = conn.cursor()
    cur.execute(f"TRUNCATE TABLE {db_name}.{table_name}")
    conn.commit()
    cur.close()
    conn.close()

# 批量插入数据
def batch_insert(db_name, table_name, cols, data_list):
    if not data_list:
        return 0
    conn = get_db_conn()
    cur = conn.cursor()
    col_str = ",".join(cols)
    val_str = ",".join(["%s"] * len(cols))
    sql = f"INSERT INTO {db_name}.{table_name} ({col_str}) VALUES ({val_str})"
    cur.executemany(sql, data_list)
    conn.commit()
    total = cur.rowcount
    cur.close()
    conn.close()
    return total

4.2 飞书工具类 feishu.py（读取表格+推送消息）

import os
import requests
from dotenv import load_dotenv

load_dotenv()
APP_ID = os.getenv("FEISHU_APP_ID")
APP_SECRET = os.getenv("FEISHU_APP_SECRET")
BOT_WEBHOOK = os.getenv("FEISHU_BOT_WEBHOOK")

# 获取飞书token
def get_feishu_token():
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    res = requests.post(url, json={"app_id":APP_ID,"app_secret":APP_SECRET})
    return res.json().get("tenant_access_token","")

# 读取飞书sheet数据
def read_feishu_sheet(spreadsheet_token, sheet_id):
    token = get_feishu_token()
    headers = {"Authorization":f"Bearer {token}"}
    url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{spreadsheet_token}/values/{sheet_id}"
    res = requests.get(url, headers=headers)
    return res.json()

# 飞书机器人推送
def feishu_send_msg(content):
    requests.post(BOT_WEBHOOK, json={
        "msg_type":"text",
        "content":{"text":content}
    })

4.3 核心同步逻辑 sync.py（通用模板）

核心能力：读取 → 清洗 → 统计清空行数 → 覆盖写入 → 推送结果

from db import get_table_count, truncate_table, batch_insert
from feishu import read_feishu_sheet, feishu_send_msg

def clean_none(val):
    if val is None or str(val).strip()=="" or str(val).lower()=="nan":
        return None
    return val

def run_sync(task):
    """
    task结构: token/sheet_id/db/table/columns
    """
    try:
        # 1.读取飞书数据
        res = read_feishu_sheet(task["token"], task["sheet_id"])
        data = res.get("data",{}).get("valueRange",{}).get("values",[])
        if len(data) <= 1:
            feishu_send_msg(f"【同步提醒】{task['name']} 无有效数据")
            return

        rows = data[1:]
        clean_data = []
        for row in rows:
            new_row = [clean_none(item) for item in row]
            clean_data.append(new_row)

        # 2.统计并清空
        old_cnt = get_table_count(task["db"], task["table"])
        truncate_table(task["db"], task["table"])

        # 3.插入新数据
        new_cnt = batch_insert(task["db"], task["table"], task["columns"], clean_data)

        # 4.推送成功消息
        msg = f"""【{task['name']} 同步成功】
原表清空数据：{old_cnt} 条
本次新增入库：{new_cnt} 条"""
        feishu_send_msg(msg)
        print(msg)

    except Exception as e:
        err_msg = f"【{task['name']} 同步失败】\n错误信息：{str(e)}"
        feishu_send_msg(err_msg)
        print(err_msg)

4.4 入口 main.py（定时 + 手动执行双模式）

支持：python main.py 定时跑｜python main.py --run_once 立即跑

import argparse
from apscheduler.schedulers.background import BackgroundScheduler
from sync import run_sync
from config import TASK_LIST, CRON_HOUR, CRON_MINUTE

def run_all():
    for task in TASK_LIST:
        run_sync(task)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--run_once", action="store_true", help="立即执行一次")
    parser.add_argument("--task", type=str, default="", help="指定单个任务")
    args = parser.parse_args()

    if args.run_once:
        if args.task:
            t = [x for x in TASK_LIST if x["name"]==args.task][0]
            run_sync(t)
        else:
            run_all()
    else:
        scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
        scheduler.add_job(run_all, "cron", hour=CRON_HOUR, minute=CRON_MINUTE)
        print(f"定时任务启动成功：每日 {CRON_HOUR}:{CRON_MINUTE} 执行")
        scheduler.start()
        while True:
            pass

五、部署运行方式（生产可用）

5.1 手动立即执行

python main.py --run_once

5.2 后台定时自动执行

python main.py

5.3 BAT 一键启动 + 日志记录

@echo off
set PYTHONUTF8=1
chcp 65001 >nul
cd /d D:\app\python_project_tmp

"D:\app\python3.10\python.exe" "ad_sync\main.py" --run_once >> "bat\logs\sync_ad.log" 2>&1

六、项目亮点

网上很多教程只能读数据，该项目是企业级完整闭环：

  • 1、完全配置化，无硬编码，.env统一管理密钥

  • 2、自动清洗脏数据（空值、NAN、空格），解决真实业务脏数据问题

  • 3、每次同步统计清空条数、新增条数，数据可追溯

  • 4、成功/失败全部飞书告警，运维零感知

  • 5、支持定时 + 手动双模式，适配生产运维

  • 6、工程化分层：解耦、可扩展、可叠加多表格任务

  • 7、日志完整、可排查问题、可长期稳定运行

    七、常见踩坑总结（干货）

  • 坑1：飞书读取数据出现 NAN/空字符串 → 统一清洗函数过滤空值、无效字符

  • 坑2：中文乱码 → 强制 PYTHONUTF8=1 + chcp65001

  • 坑3：重复数据堆积 → 每次同步前TRUNCATE全量覆盖，保证数据纯净

  • 坑4：脚本运行无日志、报错看不见 → bat重定向日志文件

  • 坑5：密钥泄露 → 全部放入.env，不上传代码仓库

    八、适用场景

  • 跨境电商广告、库存、订单飞书表格自动入MySQL

  • 企业业务表格自动化同步数仓ODS层

  • FineBI报表数据源自动更新，告别手动导表

  • 日常办公数据自动化、报表自动化、告警自动化

    九、总结

    这套方案是真正可以直接上线企业使用的飞书表格同步方案，不是玩具demo。

    具备：工程化结构、配置化管理、数据清洗、数据统计、异常告警、定时调度、日志记录，完全满足中小企业数据自动化、BI报表自动化需求。

  • 拓展：支持定制多表格同步、增量同步、数据校验、FineBI看板对接、企业自动化报表平台搭建

# python # 自动化 # mysql # etl
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

老旧厂区改造怎么配机器人：从管线更新到智能运维的分步方案

avatar DAMO开发者矩阵
cover

每日资讯快报（2026-09-14）:AI治理, 企业Agent, NVIDIA OSMO, 具身智能

avatar DAMO开发者矩阵
cover

巡检数据怎么变成决策：从告警推送到预测性维护的闭环

avatar DAMO开发者矩阵