如何写用户数据导出脚本

wen 实用脚本 2026-06-29 190

本文目录导读：

如何写用户数据导出脚本

Python + PostgreSQL 导出到CSV
Python + MySQL 导出到Excel
带筛选条件和分批导出的高级版本
命令行工具版本
导出配置示例
使用建议

我来帮你编写一个用户数据导出脚本，根据不同的场景,我提供几种常见的实现方案：

Python + PostgreSQL 导出到CSV

import csv
import psycopg2
from datetime import datetime
import os
def export_users_to_csv(db_config, output_file=None):
    """
    从数据库导出用户数据到CSV文件
    Args:
        db_config (dict): 数据库配置信息
        output_file (str): 输出文件路径
    """
    # 如果未指定输出文件，使用时间戳命名
    if not output_file:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = f"users_export_{timestamp}.csv"
    try:
        # 连接数据库
        conn = psycopg2.connect(**db_config)
        cursor = conn.cursor()
        # 查询用户数据
        query = """
        SELECT 
            id,
            username,
            email,
            created_at,
            is_active
        FROM users
        WHERE deleted_at IS NULL
        ORDER BY id
        """
        cursor.execute(query)
        rows = cursor.fetchall()
        # 获取列名
        column_names = [desc[0] for desc in cursor.description]
        # 写入CSV文件
        with open(output_file, 'w', newline='', encoding='utf-8') as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(column_names)  # 写入表头
            writer.writerows(rows)  # 写入数据
        print(f"成功导出 {len(rows)} 条用户数据到 {output_file}")
        return output_file
    except Exception as e:
        print(f"导出失败: {str(e)}")
        raise
    finally:
        if conn:
            cursor.close()
            conn.close()
# 使用示例
if __name__ == "__main__":
    db_config = {
        'host': 'localhost',
        'database': 'your_database',
        'user': 'your_user',
        'password': 'your_password',
        'port': 5432
    }
    export_users_to_csv(db_config)

Python + MySQL 导出到Excel

import pandas as pd
import pymysql
from datetime import datetime
def export_users_to_excel(db_config, output_file=None):
    """
    从MySQL数据库导出用户数据到Excel
    Args:
        db_config (dict): 数据库配置
        output_file (str): 输出文件路径
    """
    # 设置默认文件名
    if not output_file:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = f"users_export_{timestamp}.xlsx"
    try:
        # 连接数据库
        conn = pymysql.connect(**db_config)
        # SQL查询
        query = """
        SELECT 
            u.id,
            u.username,
            u.email,
            u.phone,
            u.created_at,
            u.last_login,
            u.status
        FROM users u
        WHERE u.deleted_at IS NULL
        ORDER BY u.created_at DESC
        LIMIT 10000
        """
        # 使用pandas读取数据
        df = pd.read_sql(query, conn)
        # 数据处理
        if not df.empty:
            # 格式化日期列
            df['created_at'] = pd.to_datetime(df['created_at'])
            df['created_at'] = df['created_at'].dt.strftime('%Y-%m-%d %H:%M:%S')
            if 'last_login' in df.columns:
                df['last_login'] = pd.to_datetime(df['last_login'])
                df['last_login'] = df['last_login'].dt.strftime('%Y-%m-%d %H:%M:%S')
        # 导出到Excel
        with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
            df.to_excel(writer, sheet_name='用户数据', index=False)
            # 调整列宽
            worksheet = writer.sheets['用户数据']
            for column in df:
                column_width = max(df[column].astype(str).str.len().max(), len(column) + 2)
                col_idx = df.columns.get_loc(column) + 1
                worksheet.column_dimensions[chr(64 + col_idx)].width = min(column_width, 30)
        print(f"成功导出 {len(df)} 条用户数据到 {output_file}")
        return output_file
    except Exception as e:
        print(f"导出失败: {str(e)}")
        raise
    finally:
        if conn:
            conn.close()
# 使用示例
if __name__ == "__main__":
    db_config = {
        'host': 'localhost',
        'user': 'your_user',
        'password': 'your_password',
        'database': 'your_database',
        'charset': 'utf8mb4'
    }
    export_users_to_excel(db_config)

带筛选条件和分批导出的高级版本

import csv
import json
import logging
from datetime import datetime, timedelta
class UserExporter:
    """用户数据导出器"""
    def __init__(self, db_connection):
        self.db = db_connection
        self.logger = logging.getLogger(__name__)
    def export_with_filters(self, filters=None, batch_size=1000, output_file=None):
        """
        带筛选条件的批量导出
        Args:
            filters (dict): 筛选条件
                - date_from: 开始日期
                - date_to: 结束日期
                - status: 用户状态
                - user_type: 用户类型
            batch_size (int): 每批处理数量
            output_file (str): 输出文件
        """
        if not output_file:
            output_file = f"users_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        # 构建查询条件
        conditions = ["u.deleted_at IS NULL"]
        params = []
        if filters:
            if filters.get('date_from'):
                conditions.append("u.created_at >= %s")
                params.append(filters['date_from'])
            if filters.get('date_to'):
                conditions.append("u.created_at <= %s")
                params.append(filters['date_to'])
            if filters.get('status'):
                conditions.append("u.status = %s")
                params.append(filters['status'])
            if filters.get('user_type'):
                conditions.append("u.user_type = %s")
                params.append(filters['user_type'])
        where_clause = " AND ".join(conditions)
        # 批量处理
        offset = 0
        total_exported = 0
        with open(output_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['ID', '用户名', '邮箱', '手机号', '创建时间', '状态'])
            while True:
                # 分页查询
                query = f"""
                SELECT 
                    u.id, u.username, u.email, u.phone,
                    u.created_at, u.status
                FROM users u
                WHERE {where_clause}
                ORDER BY u.id
                LIMIT {batch_size} OFFSET {offset}
                """
                cursor = self.db.cursor()
                cursor.execute(query, params)
                rows = cursor.fetchall()
                if not rows:
                    break
                # 写入CSV
                for row in rows:
                    writer.writerow(row)
                total_exported += len(rows)
                offset += batch_size
                self.logger.info(f"已导出 {total_exported} 条记录")
                cursor.close()
        self.logger.info(f"导出完成，共 {total_exported} 条记录")
        return output_file
    def export_with_progress(self, query, output_file):
        """
        带进度显示的导出
        """
        import sys
        cursor = self.db.cursor()
        cursor.execute(query)
        total_rows = cursor.rowcount
        processed = 0
        with open(output_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            # 写入列名
            writer.writerow([desc[0] for desc in cursor.description])
            while True:
                rows = cursor.fetchmany(100)
                if not rows:
                    break
                writer.writerows(rows)
                processed += len(rows)
                # 显示进度
                if total_rows > 0:
                    progress = (processed / total_rows) * 100
                    sys.stdout.write(f"\r进度: {progress:.1f}% ({processed}/{total_rows})")
                    sys.stdout.flush()
        print(f"\n导出完成！文件位置: {output_file}")

命令行工具版本

#!/usr/bin/env python3
"""
用户数据导出命令行工具
"""
import argparse
import sys
from datetime import datetime
def main():
    parser = argparse.ArgumentParser(description='用户数据导出工具')
    parser.add_argument('--db-host', required=True, help='数据库主机')
    parser.add_argument('--db-user', required=True, help='数据库用户')
    parser.add_argument('--db-password', required=True, help='数据库密码')
    parser.add_argument('--db-name', required=True, help='数据库名')
    parser.add_argument('--output', '-o', help='输出文件路径')
    parser.add_argument('--format', choices=['csv', 'json', 'excel'], default='csv', help='导出格式')
    parser.add_argument('--date-from', help='开始日期 (YYYY-MM-DD)')
    parser.add_argument('--date-to', help='结束日期 (YYYY-MM-DD)')
    parser.add_argument('--status', help='用户状态 (active/inactive)')
    args = parser.parse_args()
    # 构建数据库配置
    db_config = {
        'host': args.db_host,
        'user': args.db_user,
        'password': args.db_password,
        'database': args.db_name
    }
    # 构建筛选条件
    filters = {}
    if args.date_from:
        filters['date_from'] = args.date_from
    if args.date_to:
        filters['date_to'] = args.date_to
    if args.status:
        filters['status'] = args.status
    # 执行导出
    try:
        # 这里调用上面的导出函数
        # export_users_to_csv(db_config, args.output)
        print(f"开始导出用户数据...")
        print(f"导出格式: {args.format}")
        print(f"筛选条件: {filters}")
        # 实际导出逻辑...
        print("导出成功完成！")
    except Exception as e:
        print(f"导出失败: {e}", file=sys.stderr)
        sys.exit(1)
if __name__ == "__main__":
    main()

导出配置示例

# export_config.py
EXPORT_CONFIG = {
    'csv': {
        'delimiter': ',',
        'encoding': 'utf-8-sig',  # 解决Excel中文乱码
        'fields': [
            'id', 'username', 'email', 'phone', 
            'created_at', 'status', 'user_type'
        ]
    },
    'json': {
        'indent': 2,
        'ensure_ascii': False,
        'fields': [
            'id', 'username', 'email', 'phone',
            'created_at', 'status', 'user_type'
        ]
    },
    'excel': {
        'sheet_name': '用户数据',
        'freeze_panes': 'A2',  # 冻结首行
        'auto_width': True
    }
}

使用建议

性能优化：
- 大数据量时使用分批导出
- 添加索引优化查询
- 考虑使用异步IO
数据安全：
- 敏感字段（密码、token等）不要导出
- 导出文件添加访问控制
- 定期清理临时文件
错误处理：
- 添加重试机制
- 记录错误日志
- 断点续传功能
监控告警：
- 导出量监控
- 执行时间监控
- 失败通知

需要根据你的具体需求（数据库类型、数据量、导出频率等）选择合适的方案。

本文地址： https://www.bmjup.com/post/11935.html