如何实现Excel数据更新时每日自动导入PostgreSQL数据库

方案覆盖「Excel更新实时触发同步」+「每日定时兜底同步」双逻辑，全程无需人工操作，不需要采购重型数据同步平台，用轻量脚本即可实现。

前置准备 #

• 部署Python3运行环境，安装依赖库：pip install pandas sqlalchemy openpyxl watchdog schedule psycopg2-binary
• 提前在PostgreSQL中创建目标表，字段顺序、数据类型和Excel列保持一致，给同步账号分配目标表的写入、清空权限
• 固定Excel文件的存放路径，不要随意移动位置，避免监控失效

核心同步代码 #

所有触发场景共用一套同步逻辑，避免重复维护：

import os
import time
import pandas as pd
from sqlalchemy import create_engine
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from datetime import datetime
import schedule
import threading

# -------------------------- 配置项 按实际情况修改 --------------------------
EXCEL_FILE_PATH = "/data/business_data.xlsx"  # Excel文件绝对路径
PG_CONN_STR = "postgresql://sync_user:your_password@127.0.0.1:5432/business_db"  # PG连接串
TARGET_TABLE = "biz_excel_data"  # PG目标表名
EXCEL_SHEET = 0  # 要同步的Sheet页，填名称或者序号都可以
DAILY_SYNC_TIME = "02:00"  # 每日定时同步时间
DEBOUNCE_INTERVAL = 3  # 文件监控防抖间隔，单位秒
# ---------------------------------------------------------------------------

def sync_excel_to_pg():
    """统一同步入口：读取Excel -> 数据清洗 -> 写入PostgreSQL"""
    try:
        # 读取Excel，自动跳过空行
        df = pd.read_excel(EXCEL_FILE_PATH, sheet_name=EXCEL_SHEET, engine="openpyxl")
        
        # 这里可以加自定义清洗逻辑，比如去重、空值填充、字段格式转换、加同步时间戳
        # df = df.dropna(subset=["订单编号"])
        # df["sync_time"] = datetime.now()

        engine = create_engine(PG_CONN_STR)
        with engine.begin() as conn:
            # 全量同步场景用下面这行清空旧数据，增量同步直接注释
            conn.execute(f"TRUNCATE TABLE {TARGET_TABLE};")
            # 分批写入，避免大文件占满内存
            df.to_sql(
                name=TARGET_TABLE,
                con=conn,
                if_exists="append",
                index=False,
                chunksize=1000
            )
        print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 同步完成，共写入{len(df)}条数据")
    except Exception as e:
        print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 同步异常：{str(e)}")

触发规则配置 #

1. Excel变更实时触发 #

用文件系统监听捕获Excel的保存修改事件，加防抖逻辑避免Excel保存临时文件导致重复触发：

class ExcelUpdateHandler(FileSystemEventHandler):
    def __init__(self):
        self.last_run_ts = 0

    def on_modified(self, event):
        if os.path.abspath(event.src_path) == os.path.abspath(EXCEL_FILE_PATH):
            now_ts = time.time()
            if now_ts - self.last_run_ts > DEBOUNCE_INTERVAL:
                self.last_run_ts = now_ts
                sync_excel_to_pg()

def run_file_monitor():
    event_handler = ExcelUpdateHandler()
    observer = Observer()
    monitor_dir = os.path.dirname(EXCEL_FILE_PATH) or "."
    observer.schedule(event_handler, path=monitor_dir, recursive=False)
    observer.start()
    print(f"文件监控已启动，正在监听：{EXCEL_FILE_PATH}")
    while True:
        time.sleep(1)

2. 每日定时兜底同步 #

配置固定时间全量同步，覆盖文件监控漏触发的异常场景（比如文件被整体替换、服务临时重启期间的变更）：

def run_scheduled_task():
    schedule.every().day.at(DAILY_SYNC_TIME).do(sync_excel_to_pg)
    print(f"定时任务已启动，每日{DAILY_SYNC_TIME}执行全量同步")
    while True:
        schedule.run_pending()
        time.sleep(30)

服务启动入口 #

用守护线程同时跑两个触发逻辑，避免互相阻塞：

if __name__ == "__main__":
    # 服务启动先跑一次全量同步
    sync_excel_to_pg()
    # 启动两个守护线程
    threading.Thread(target=run_file_monitor, daemon=True).start()
    threading.Thread(target=run_scheduled_task, daemon=True).start()
    # 保持主线程常驻
    while True:
        time.sleep(60)

常驻运行配置 #

脚本写完不要直接开个终端挂着，要配置成系统服务，保证开机自启、异常自动重启：

• Windows环境：写个bat启动脚本，加入系统「任务计划程序」，设置触发条件为「计算机启动时」，勾选「不管用户是否登录都要运行」
• Linux环境：配置systemd服务，示例配置如下，配置完执行systemctl enable --now excel-sync.service即可开机自启

[Unit]
Description=Excel to PG Auto Sync Service
After=network.target postgresql.service

[Service]
User=deploy
WorkingDirectory=/opt/excel_sync/
ExecStart=/usr/bin/python3 /opt/excel_sync/sync_main.py
Restart=always
RestartSec=5
StandardOutput=append:/var/log/excel_sync.log
StandardError=append:/var/log/excel_sync_error.log

[Install]
WantedBy=multi-user.target

注意事项：

• 不要给同步用的Excel设置打开密码，不要留合并单元格、跨表公式，否则读取会报错
• 增量同步场景去掉TRUNCATE逻辑，给表加业务唯一键，写入时做冲突更新避免重复数据
• 单表超过10万行的大文件，把chunksize调到500-1000，降低内存占用
• 日志默认输出到配置的系统日志路径，有问题直接查日志定位即可

内容的提问来源于stack exchange，提问作者George Zambrano