本文目录导读：
我来帮你编写一个用户数据导出脚本，根据不同的场景,我提供几种常见的实现方案：
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等）不要导出
- 导出文件添加访问控制
- 定期清理临时文件
-
错误处理：
- 添加重试机制
- 记录错误日志
- 断点续传功能
-
监控告警：
- 导出量监控
- 执行时间监控
- 失败通知
需要根据你的具体需求（数据库类型、数据量、导出频率等）选择合适的方案。
