Python实现将MySQL中所有表的数据都导出为CSV文件并压缩

import os
import csv
import zipfile
import logging
import mysql.connector
from datetime import datetime
import time
import concurrent.futures
import glob

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('data_export.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

def export_table_to_csv(table_name, csv_path, db_config, batch_size=1000):
    """导出单个表的数据到CSV文件，分批处理"""
    conn = None
    cursor = None
    total_rows = 0
    try:
        conn = mysql.connector.connect(**db_config)
        cursor = conn.cursor()

        # 获取数据并写入CSV
        with open(csv_path, 'w', newline='', encoding='utf-8') as csvfile:
            writer = csv.writer(csvfile)
            
            # 执行查询并获取列名
            cursor.execute(f"SELECT * FROM `{table_name}`")
            columns = [col[0] for col in cursor.description]
            writer.writerow(columns)
            
            # 分批获取数据
            while True:
                rows = cursor.fetchmany(batch_size)
                if not rows:
                    break
                writer.writerows(rows)
                total_rows += len(rows)
                logger.debug(f"{table_name} 已导出 {total_rows} 行")

        logger.info(f"{table_name} CSV导出完成，总行数：{total_rows}")
        return total_rows

    except Exception as e:
        logger.error(f"导出表 {table_name} 失败: {str(e)}", exc_info=True)
        raise
    finally:
        if cursor:
            cursor.close()
        if conn and conn.is_connected():
            conn.close()

def compress_to_zip(source_path, zip_path):
    """压缩文件为ZIP格式"""
    try:
        with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
            zipf.write(source_path, arcname=os.path.basename(source_path))
        logger.info(f"成功压缩 {source_path} 到 {zip_path}")
    except Exception as e:
        logger.error(f"压缩 {source_path} 失败: {str(e)}", exc_info=True)
        raise

def process_table(table_name, db_config, csv_dir, zip_dir):
    """处理单个表的导出和压缩"""
    start_time = time.time()
    logger.info(f"开始处理表: {table_name}")
    status = "成功"
    rows_exported = 0

    try:
        # 定义文件路径
        csv_filename = f"{table_name}.csv"
        zip_filename = f"{table_name}.zip"
        csv_path = os.path.join(csv_dir, csv_filename)
        zip_path = os.path.join(zip_dir, zip_filename)

        # 导出CSV
        rows_exported = export_table_to_csv(table_name, csv_path, db_config)
        
        # 压缩文件
        compress_to_zip(csv_path, zip_path)

    except Exception as e:
        status = f"失败: {str(e)}"
        # 清理可能存在的中间文件
        for path in [csv_path, zip_path]:
            if path and os.path.exists(path):
                try:
                    os.remove(path)
                    logger.warning(f"已清理文件: {path}")
                except Exception as clean_error:
                    logger.error(f"清理文件失败: {clean_error}")

    finally:
        duration = time.time() - start_time
        log_message = (
            f"表处理完成 - 表名: {table_name}, "
            f"状态: {status}, "
            f"导出行数: {rows_exported}, "
            f"耗时: {duration:.2f}秒"
        )
        logger.info(log_message)

def unzip_files(zip_dir, unzip_dir):
    """解压指定目录中的所有ZIP文件"""
    zip_files = glob.glob(os.path.join(zip_dir, '*.zip'))
    if not zip_files:
        logger.warning("未找到ZIP文件，跳过解压")
        return

    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = []
        for zip_path in zip_files:
            futures.append(executor.submit(
                lambda: extract_zip(zip_path, unzip_dir)
            ))
        for future in concurrent.futures.as_completed(futures):
            try:
                future.result()
            except Exception as e:
                logger.error(f"解压过程中发生错误: {str(e)}")

def extract_zip(zip_path, unzip_dir):
    """解压单个ZIP文件"""
    try:
        start_time = time.time()
        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
            zip_ref.extractall(unzip_dir)
        duration = time.time() - start_time
        logger.info(f"解压完成: {zip_path} => {unzip_dir} (耗时: {duration:.2f}秒)")
    except Exception as e:
        logger.error(f"解压 {zip_path} 失败: {str(e)}", exc_info=True)
        raise

def main():
    # 配置参数
    db_config = {
        'host': 'localhost',
        'user': 'your_username',
        'password': 'your_password',
        'database': 'your_database'
    }
    
    # 目录配置
    base_dir = os.path.dirname(os.path.abspath(__file__))
    csv_dir = os.path.join(base_dir, 'csv_exports')
    zip_dir = os.path.join(base_dir, 'zip_archives')
    unzip_dir = os.path.join(base_dir, 'unzipped_files')

    # 创建目录
    for dir_path in [csv_dir, zip_dir, unzip_dir]:
        os.makedirs(dir_path, exist_ok=True)
        logger.info(f"目录已准备: {dir_path}")

    # 获取所有表名
    try:
        conn = mysql.connector.connect(**db_config)
        cursor = conn.cursor()
        cursor.execute("SHOW TABLES")
        tables = [table[0] for table in cursor.fetchall()]
        logger.info(f"发现 {len(tables)} 个需要处理的表")
    except Exception as e:
        logger.error(f"获取数据库表失败: {str(e)}", exc_info=True)
        return
    finally:
        if 'cursor' in locals():
            cursor.close()
        if 'conn' in locals() and conn.is_connected():
            conn.close()

    # 处理所有表（多线程导出和压缩）
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        futures = []
        for table in tables:
            futures.append(executor.submit(
                process_table,
                table,
                db_config,
                csv_dir,
                zip_dir
            ))

        # 处理任务结果
        for future in concurrent.futures.as_completed(futures):
            try:
                future.result()
            except Exception as e:
                logger.error(f"表处理异常: {str(e)}")

    # 解压所有ZIP文件（多线程解压）
    logger.info("开始解压所有ZIP文件")
    unzip_files(zip_dir, unzip_dir)
    logger.info("全部处理流程完成")

if __name__ == "__main__":
    main()

pip install mysql-connector-python

python script.py