使用Python实现定时自动导入不同CSV文件的技术咨询

核心思路 #

将每个CSV文件与对应的执行时间建立映射关系，通过调度框架或系统级定时任务，在指定时间触发CSV导入逻辑。导入逻辑可根据需求扩展（如数据校验、存入数据库等）。

方案一：Python + Schedule 库（程序化调度） #

这是最灵活的程序化实现方式，适合需要在代码中统一管理所有定时任务的场景。

1. 安装依赖 #

pip install schedule pandas

2. 完整实现代码 #

import schedule
import time
import pandas as pd
import logging
from pathlib import Path

# 配置日志，替代print方便排查问题
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("csv_import.log"), logging.StreamHandler()]
)

# 定义CSV文件路径与执行时间的映射，根据实际需求补充完整50个条目
CSV_TASKS = [
    {"path": "/data/csv/file-1.csv", "run_time": "22:00"},
    {"path": "/data/csv/file-2.csv", "run_time": "22:30"},
    # ... 剩余48个文件的配置
]

def import_csv(file_path: str):
    """执行CSV文件导入逻辑，可按需扩展"""
    file = Path(file_path)
    if not file.exists():
        logging.error(f"文件不存在: {file_path}")
        return
    
    try:
        # 读取CSV文件
        df = pd.read_csv(file)
        # 这里添加自定义处理逻辑：如数据清洗、写入数据库等
        logging.info(f"成功导入文件 {file_path}，共 {len(df)} 条记录")
    except Exception as e:
        logging.error(f"导入文件 {file_path} 失败: {str(e)}")

# 批量注册定时任务
for task in CSV_TASKS:
    schedule.every().day.at(task["run_time"]).do(import_csv, task["path"])

if __name__ == "__main__":
    logging.info("CSV定时导入服务已启动，按Ctrl+C终止")
    while True:
        schedule.run_pending()
        time.sleep(60)  # 每分钟检查一次待执行任务

方案二：系统级定时任务（更稳定） #

如果不需要在代码中管理调度，可直接使用操作系统的定时任务工具，适合长期稳定运行的场景。

Linux/macOS（Crontab） #

1. 编写通用导入脚本 csv_importer.py：

import sys
import pandas as pd
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s")

def import_csv(file_path):
    try:
        df = pd.read_csv(file_path)
        logging.info(f"成功导入 {file_path}")
        # 添加自定义处理逻辑
    except Exception as e:
        logging.error(f"导入失败: {str(e)}")

if __name__ == "__main__":
    if len(sys.argv) != 2:
        logging.error("用法: python csv_importer.py <CSV文件路径>")
        sys.exit(1)
    import_csv(sys.argv[1])

2. 编辑Crontab任务：
   执行 crontab -e 添加以下条目（根据时间需求调整）：

# 每晚22:00导入file-1.csv
0 22 * * * /usr/bin/python3 /path/to/csv_importer.py /data/csv/file-1.csv >> /var/log/csv_import.log 2>&1
# 每晚22:30导入file-2.csv
30 22 * * * /usr/bin/python3 /path/to/csv_importer.py /data/csv/file-2.csv >> /var/log/csv_import.log 2>&1
# ... 剩余48个任务的Crontab配置

Windows（任务计划程序） #

1. 编写与上述相同的csv_importer.py脚本。
2. 打开「任务计划程序」，创建基本任务：
   • 设置触发时间（如每天22:00）
   • 操作选择「启动程序」，选择Python.exe路径，添加参数：/path/to/csv_importer.py /data/csv/file-1.csv
   • 重复此步骤为每个CSV文件创建对应任务

关键注意事项 #

• 文件权限：确保脚本运行用户有CSV文件的读取权限，以及后续操作（如写入数据库）的权限。
• 文件冲突处理：若CSV文件可能被实时更新，可添加文件锁或检查文件修改时间，避免导入不完整数据。
• 容错与重试：对于方案一，可结合tenacity库实现失败重试；方案二可通过Crontab的重试机制或任务计划程序的失败重试设置。
• 监控：定期检查日志文件，确保任务正常执行；可添加告警逻辑（如邮件通知）当任务失败时触发。

内容的提问来源于stack exchange，提问作者Mohammed Abdulhameed