如何编写自动化导入导出脚本

wen 实用脚本 2026-07-30 2

如何编写导入导出脚本（实战指南）

目录导读

自动化脚本的核心价值
前期准备：需求分析与环境搭建
脚本编写五步法
常见场景与代码示例
错误处理与性能优化
QA问答：解决实际难题

自动化脚本的核心价值

在企业数据运维中，手动导入导出数据如同“人肉搬运工”——耗时、易错、不可追溯，编写自动化脚本能将重复性工作转化为一键执行,尤其适用于以下场景：

如何编写自动化导入导出脚本

数据库迁移（MySQL → PostgreSQL）
报表数据定时生成（Excel → 数据库）
日志文件解析与清洗（CSV → JSON）

关键原则：脚本必须具备可复用性、健壮性、日志记录功能。

前期准备：需求分析与环境搭建

步骤拆解：

明确数据源与目标：源文件格式（CSV/JSON/Excel）、目标数据库类型（MySQL/SQLite/API）。
选择工具：Python（首选，库丰富）、Bash（适合简单文件操作）、PowerShell（Windows环境）。

安装依赖：

pip install pandas sqlalchemy openpyxl pymysql

环境变量管理：使用.env文件存储数据库密码,避免硬编码。

from dotenv import load_dotenv
load_dotenv()
db_pass = os.getenv("DB_PASSWORD")

脚本编写五步法

① 定义函数模块

将导入、导出、清洗分别封装为独立函数,便于单元测试。

② 数据读取与校验

import pandas as pd
def read_csv(file_path):
    try:
        df = pd.read_csv(file_path, encoding='utf-8-sig')
        print(f"成功读取{len(df)}行数据")
        # 基础校验：空值、类型
        assert df['id'].notna().all(), "ID列存在空值"
        return df
    except Exception as e:
        logging.error(f"读取失败: {e}")
        return None

③ 数据清洗与转换

常见操作：日期格式化、去重、列名映射。

df['date'] = pd.to_datetime(df['date'], errors='coerce')
df = df.drop_duplicates(subset=['order_id'])

④ 批量写入目标

使用to_sql实现数据库写入，注意if_exists参数。

from sqlalchemy import create_engine
engine = create_engine(f'mysql+pymysql://{user}:{pass}@{host}/{db}')
df.to_sql('orders', con=engine, if_exists='append', index=False)

⑤ 日志与通知

记录每次操作的开始、结束、影响行数,并支持发送邮件告警。

常见场景与代码示例

场景A：CSV导出为JSON（每行一个对象）

import json
def csv_to_json(csv_file, json_file):
    df = pd.read_csv(csv_file)
    with open(json_file, 'w', encoding='utf-8') as f:
        for record in df.to_dict(orient='records'):
            f.write(json.dumps(record, ensure_ascii=False) + '\n')

场景B：数据库表导出为Excel多工作表

with pd.ExcelWriter('report.xlsx', engine='openpyxl') as writer:
    df1.to_excel(writer, sheet_name='本月订单')
    df2.to_excel(writer, sheet_name='退款记录')

错误处理与性能优化

错误处理三要素

重试机制：网络中断时等待3秒重试3次。
数据快照：写入前备份原表,以防失败回滚。
断点续传：记录成功写入的行号,异常后从该行继续。

性能提升技巧

分块读取大文件：pd.read_csv(file, chunksize=5000)
使用批量插入（executemany）
禁用索引：to_sql(..., index=False)

QA问答：解决实际难题

Q1：编写脚本时，如何避免SQL注入？
A：永远不要拼接SQL字符串，使用参数化查询或ORM（如SQLAlchemy）。

engine.execute(text("INSERT INTO users (name) VALUES (:name)"), {"name": user_input})

Q2：脚本运行耗时太长，如何监控进度？
A：加入tqdm进度条库，配合chunksize分片处理。

from tqdm import tqdm
for chunk in tqdm(pd.read_csv(file, chunksize=1000)):
    chunk.to_sql(...)

Q3：如何处理不同编码的CSV文件？
A：使用chardet库自动检测编码，避免乱码。

import chardet
with open(file, 'rb') as f:
    encoding = chardet.detect(f.read())['encoding']
df = pd.read_csv(file, encoding=encoding)

Q4：脚本部署到生产环境后，如何管理依赖？
A：使用pip freeze > requirements.txt锁定版本,并通过Docker容器化运行确保环境一致性。

编写自动化导入导出脚本的核心是：模块化设计、强健的错误处理、清晰的日志追踪，从简单的CSV搬运到复杂的异构数据库同步，掌握上述方法后，你便能将90%的数据搬运工工作交给代码，而专注于更有价值的业务分析，先测试小样，再批量操作,永远保留原始备份。