# mysql_csv_exporter

**Repository Path**: hephec321/mysql_csv_exporter

## Basic Information

- **Project Name**: mysql_csv_exporter
- **Description**: mysql数据导出为csv文件
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-12-17
- **Last Updated**: 2025-12-17

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# MySQL CSV导出工具

一个基于Python的MySQL数据导出CSV文件工具，支持定时任务调度、手动触发和灵活的配置选项。

## 功能特点

- 支持自定义SQL查询语句导出数据
- 可配置CSV文件格式（分隔符、编码等）
- 支持定时任务调度（基于APScheduler）
- 支持一次性任务和周期性任务
- 支持手动触发导出
- 完整的日志记录
- 简单易用的YAML配置文件

## 项目结构

```
mysql-csv-exporter/
├── config.yaml          # 数据库配置
├── exporter.py          # 核心导出逻辑
├── scheduler.py         # 定时任务（APScheduler）
├── tests/               # 单元测试
│   ├── __init__.py
│   ├── test_exporter.py
│   └── test_scheduler.py
├── requirements.txt     # 依赖包列表
└── README.md           # 项目说明文档
```

## 安装依赖

```bash
pip install -r requirements.txt
```

## 配置说明

### 数据库配置

```yaml
database:
  host: localhost
  port: 3306
  username: root
  password: password
  database: test_db
  charset: utf8mb4
```

### 导出配置

```yaml
export:
  # 导出目录
  output_dir: ./exports
  # CSV文件格式配置
  csv_format:
    # 是否包含表头
    include_header: true
    # 字段分隔符
    delimiter: ","
    # 引号字符
    quotechar: '"'
    # 行结束符
    lineterminator: "\n"
    # 编码格式
    encoding: "utf-8"
  # 导出任务配置
  tasks:
    - name: "用户数据导出"
      # SQL查询语句
      sql: "SELECT id, username, email, created_at FROM users WHERE created_at >= %s"
      # SQL参数（可选）
      params: ["2023-01-01"]
      # 输出文件名（支持时间格式化）
      filename: "users_{timestamp}.csv"
      # 是否启用此任务
      enabled: true
```

### 调度器配置

```yaml
scheduler:
  # 调度器类型: background, blocking
  type: background
  # 时区设置
  timezone: Asia/Shanghai
  # 任务配置
  jobs:
    - name: "每日用户数据导出"
      # 关联的导出任务名称
      task_name: "用户数据导出"
      # 调度类型: date, interval, cron
      trigger: cron
      # 调度表达式
      # cron表达式: 分 时 日 月 周
      cron: "0 2 * * *"  # 每天凌晨2点执行
      # 是否启用
      enabled: true
```

## 使用方法

### 1. 直接导出数据

```python
from exporter import MySQLExporter

# 使用上下文管理器确保连接正确关闭
with MySQLExporter() as exporter:
    # 导出单个任务
    output_file = exporter.export_to_csv("用户数据导出")
    print(f"导出完成: {output_file}")
    
    # 导出所有启用的任务
    output_files = exporter.export_all_enabled_tasks()
    print(f"所有导出完成: {output_files}")
```

### 2. 使用调度器

```python
from scheduler import TaskScheduler

# 使用上下文管理器确保调度器正确停止
with TaskScheduler() as scheduler:
    # 启动调度器
    scheduler.start()
    
    # 添加手动任务（立即执行）
    scheduler.add_manual_job("用户数据导出")
    
    # 添加手动任务（指定时间）
    scheduler.add_manual_job("用户数据导出", "2023-12-31 23:59:59")
    
    # 列出所有作业
    jobs = scheduler.list_jobs()
    print("当前作业列表:")
    for job in jobs:
        print(f"  {job['name']}: {job['next_run_time']}")
```

### 3. 命令行使用

```bash
# 直接导出数据
python exporter.py

# 启动调度器
python scheduler.py
```

## 调度器类型说明

### Cron表达式

支持标准的5位cron表达式：`分 时 日 月 周`

示例：
- `0 2 * * *` - 每天凌晨2点
- `0 */6 * * *` - 每6小时
- `0 0 * * 0` - 每周日午夜
- `0 0 1 * *` - 每月1日午夜

### 间隔调度

支持按时间间隔调度：
- `weeks`: 周
- `days`: 天
- `hours`: 小时
- `minutes`: 分钟
- `seconds`: 秒

### 一次性调度

支持指定具体时间执行一次：
- `run_date`: 执行时间，格式为 `YYYY-MM-DD HH:MM:SS`

## 运行测试

```bash
# 运行所有测试
pytest tests/

# 运行测试并生成覆盖率报告
pytest tests/ --cov=. --cov-report=html
```

## 日志配置

日志配置在 `config.yaml` 中：

```yaml
logging:
  # 日志级别: DEBUG, INFO, WARNING, ERROR, CRITICAL
  level: INFO
  # 日志文件路径
  file: ./logs/exporter.log
  # 日志格式
  format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
  # 日志文件最大大小(MB)
  max_size: 10
  # 保留的日志文件数量
  backup_count: 5
```

## 注意事项

1. 确保MySQL数据库连接参数正确
2. 确保导出目录有写入权限
3. 大数据量导出时注意内存使用情况
4. 定时任务依赖系统时间，确保系统时间准确
5. 生产环境建议使用 `background` 类型的调度器

## 常见问题

### 1. 连接数据库失败

检查数据库连接参数是否正确，网络是否通畅，数据库用户是否有足够权限。

### 2. 导出文件为空

检查SQL查询是否正确，是否有符合条件的数据。

### 3. 定时任务不执行

检查cron表达式是否正确，时区设置是否正确，调度器是否正常启动。

## 许可证

本项目采用MIT许可证，详见LICENSE文件。