RPA+数据库：批量处理百万条数据，比脚本更稳更快

┌─────────────────────────────────────────┐
│  调度层：定时触发 / API触发 / 手动触发    │
├─────────────────────────────────────────┤
│  处理层：分批读取 → 数据清洗 → 批量写入  │
├─────────────────────────────────────────┤
│  持久层：源数据库 → 临时表 → 目标库/文件  │
└─────────────────────────────────────────┘

import pymysql
import pandas as pd
from datetime import datetime

def migrate_orders():
    conn = pymysql.connect(host='localhost', user='root', 
                           password='xxx', database='shop', charset='utf8mb4')
    try:
        # 隐患1：一次性加载80万条，内存风险
        df = pd.read_sql("""
            SELECT * FROM orders 
            WHERE create_time BETWEEN '2026-04-01' AND '2026-06-30'
        """, conn)

        # 隐患2：无异常隔离，一条脏数据全崩
        df['phone'] = df['phone'].apply(clean_phone)

        # 隐患3：单条插入，性能极差
        for _, row in df.iterrows():
            cursor.execute("INSERT INTO orders_summary ...", row)

    except Exception as e:
        print(f"任务失败: {e}")  # 隐患4：失败即结束，无断点续跑
        conn.rollback()
    finally:
        conn.close()

# db_config.py
DB_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'rpa_user',
    'password': 'your_password',
    'database': 'shop',
    'charset': 'utf8mb4',
    'connect_timeout': 30,
    'read_timeout': 300,
    'write_timeout': 300
}

# 连接池配置，避免频繁创建连接
POOL_CONFIG = {
    'pool_name': 'rpa_pool',
    'pool_size': 5,
    'pool_reset_session': True
}

import mysql.connector
from mysql.connector import pooling

def create_connection_pool():
    """创建数据库连接池，复用连接降低开销"""
    return mysql.connector.pooling.MySQLConnectionPool(
        pool_name="batch_pool",
        pool_size=5,
        **DB_CONFIG
    )

def fetch_in_batches(pool, batch_size=5000):
    """
    使用服务器端游标分批读取，避免客户端内存爆炸
    每次只加载 batch_size 条记录到内存
    """
    conn = pool.get_connection()
    # SSCursor (Server Side Cursor) 关键：数据不一次性拉到客户端
    cursor = conn.cursor(dictionary=True, buffered=False)

    try:
        cursor.execute("""
            SELECT id, order_no, phone, province, amount, create_time 
            FROM orders 
            WHERE create_time BETWEEN '2026-04-01' AND '2026-06-30'
            ORDER BY id
        """)

        batch = []
        for row in cursor:
            batch.append(row)
            if len(batch) >= batch_size:
                yield batch
                batch = []

        if batch:  # 最后一批
            yield batch

    finally:
        cursor.close()
        conn.close()

import re
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('data_clean.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

def clean_phone(phone):
    """清洗手机号：去空格、统一格式"""
    if not phone:
        return None
    # 去除所有非数字字符
    digits = re.sub(r'\D', '', str(phone))
    # 统一添加+86前缀（如果没有）
    if len(digits) == 11 and digits.startswith('1'):
        return f'+86{digits}'
    elif len(digits) == 13 and digits.startswith('86'):
        return f'+{digits}'
    return digits  # 异常格式保留原样，后续人工复核

def process_batch(batch_records):
    """
    处理一批数据，单条异常不影响整批
    返回: (成功列表, 失败列表)
    """
    success = []
    failed = []

    for record in batch_records:
        try:
            cleaned = {
                'id': record['id'],
                'order_no': record['order_no'],
                'phone': clean_phone(record['phone']),
                'province': record['province'] or '未知',
                'amount': float(record['amount']) if record['amount'] else 0.0,
                'create_time': record['create_time']
            }
            success.append(cleaned)
        except Exception as e:
            failed.append({
                'record': record,
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            })
            logging.warning(f"记录清洗失败 id={record.get('id')}: {e}")

    return success, failed

def batch_insert(pool, table_name, records, columns):
    """
    使用executemany批量插入，配合事务保证原子性
    """
    if not records:
        return 0

    conn = pool.get_connection()
    cursor = conn.cursor()

    # 构造INSERT语句
    placeholders = ', '.join(['%s'] * len(columns))
    sql = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({placeholders})"

    try:
        # 提取值列表
        values = [[r[col] for col in columns] for r in records]

        cursor.executemany(sql, values)
        conn.commit()

        return cursor.rowcount

    except Exception as e:
        conn.rollback()
        logging.error(f"批量插入失败: {e}")
        # 降级为单条插入，进一步隔离异常
        success_count = 0
        for record in records:
            try:
                single_values = [record[col] for col in columns]
                cursor.execute(sql, single_values)
                conn.commit()
                success_count += 1
            except Exception as e2:
                logging.error(f"单条插入失败 id={record.get('id')}: {e2}")
        return success_count

    finally:
        cursor.close()
        conn.close()

import json
import os
from datetime import datetime

CHECKPOINT_FILE = 'migrate_checkpoint.json'

def load_checkpoint():
    """加载断点，支持任务中断后从上次位置继续"""
    if os.path.exists(CHECKPOINT_FILE):
        with open(CHECKPOINT_FILE, 'r', encoding='utf-8') as f:
            return json.load(f)
    return {'last_processed_id': 0, 'total_success': 0, 'total_failed': 0}

def save_checkpoint(last_id, success_count, failed_count):
    """保存断点"""
    with open(CHECKPOINT_FILE, 'w', encoding='utf-8') as f:
        json.dump({
            'last_processed_id': last_id,
            'total_success': success_count,
            'total_failed': failed_count,
            'updated_at': datetime.now().isoformat()
        }, f, ensure_ascii=False, indent=2)

def main():
    pool = create_connection_pool()
    checkpoint = load_checkpoint()

    total_success = checkpoint['total_success']
    total_failed = checkpoint['total_failed']
    last_id = checkpoint['last_processed_id']

    logging.info(f"从断点恢复: last_id={last_id}, 已处理成功={total_success}")

    batch_count = 0
    for batch in fetch_in_batches(pool, batch_size=5000):
        batch_count += 1

        # 跳过已处理的批次（简单实现：基于ID过滤）
        if batch and batch[0]['id'] <= last_id:
            continue

        # 清洗
        success_records, failed_records = process_batch(batch)

        # 写入
        columns = ['id', 'order_no', 'phone', 'province', 'amount', 'create_time']
        inserted = batch_insert(pool, 'orders_summary', success_records, columns)

        total_success += inserted
        total_failed += len(failed_records)
        last_id = batch[-1]['id']

        # 每10个批次保存一次断点
        if batch_count % 10 == 0:
            save_checkpoint(last_id, total_success, total_failed)
            logging.info(f"进度: 批次={batch_count}, 成功={total_success}, 失败={total_failed}")

    # 最终保存
    save_checkpoint(last_id, total_success, total_failed)
    logging.info(f"任务完成: 总计成功={total_success}, 失败={total_failed}")

    # 失败记录单独导出，供人工复核
    if failed_records:
        with open('failed_records.json', 'w', encoding='utf-8') as f:
            json.dump(failed_records, f, ensure_ascii=False, indent=2)

if __name__ == '__main__':
    main()

[定时触发/API触发] → [连接数据库] → [分批查询] → [数据清洗] → [批量写入] → [发送邮件]
         ↓                  ↓              ↓              ↓              ↓
    [定时任务调度]     [异常重试]    [脏数据隔离]    [事务提交]    [生成报表]