pip install pymysql sqlalchemy sshtunnel  # 基本依賴
pip install pandas mysql-connector-python  # 高級功能可選

SOURCE_DB = {
    'host': 'source_host',
    'user': 'username',
    'password': 'password',
    'database': 'db_name',
    'port': 3306,
    'charset': 'utf8mb4'
}

TARGET_DB = {
    'host': 'target_host',
    'user': 'username',
    'password': 'password',
    'database': 'db_name',
    'port': 3306
}

import pymysql
from db_config import SOURCE_DB, TARGET_DB

def full_sync(source_config, target_config):
    try:
        # 連接源數(shù)據(jù)庫
        source_conn = pymysql.connect(**source_config)
        with source_conn.cursor() as src_cursor:
            src_cursor.execute("SHOW TABLES")
            tables = src_cursor.fetchall()
            
            # 連接目標(biāo)數(shù)據(jù)庫
            target_conn = pymysql.connect(**target_config)
            
            for (table,) in tables:
                print(f"同步表: {table}")
                
                # 獲取表結(jié)構(gòu)
                src_cursor.execute(f"SHOW CREATE TABLE {table}")
                create_table_sql = src_cursor.fetchone()[1]
                
                # 在目標(biāo)庫重建表（先刪除舊表）
                with target_conn.cursor() as tgt_cursor:
                    tgt_cursor.execute(f"DROP TABLE IF EXISTS {table}")
                    tgt_cursor.execute(create_table_sql)
                
                # 獲取數(shù)據(jù)并插入
                src_cursor.execute(f"SELECT * FROM {table}")
                rows = src_cursor.fetchall()
                if rows:
                    columns = [desc[0] for desc in src_cursor.description]
                    placeholders = ', '.join(['%s'] * len(columns))
                    insert_sql = f"INSERT INTO {table} ({', '.join(columns)}) VALUES ({placeholders})"
                    
                    with target_conn.cursor() as tgt_cursor:
                        tgt_cursor.executemany(insert_sql, rows)
                    target_conn.commit()
                    
    except Exception as e:
        print(f"同步失敗: {str(e)}")
    finally:
        source_conn.close() if 'source_conn' in locals() else None
        target_conn.close() if 'target_conn' in locals() else None

# 執(zhí)行全量同步
full_sync(SOURCE_DB, TARGET_DB)

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker
from db_config import SOURCE_DB, TARGET_DB

def orm_sync():
    # 創(chuàng)建引擎
    source_engine = create_engine(
        f"mysql+pymysql://{SOURCE_DB['user']}:{SOURCE_DB['password']}@"
        f"{SOURCE_DB['host']}:{SOURCE_DB['port']}/{SOURCE_DB['database']}"
    )
    target_engine = create_engine(
        f"mysql+pymysql://{TARGET_DB['user']}:{TARGET_DB['password']}@"
        f"{TARGET_DB['host']}:{TARGET_DB['port']}/{TARGET_DB['database']}"
    )
    
    # 獲取源庫元數(shù)據(jù)
    source_meta = MetaData(bind=source_engine)
    source_meta.reflect()
    
    # 創(chuàng)建目標(biāo)會話
    TargetSession = sessionmaker(bind=target_engine)
    target_session = TargetSession()
    
    try:
        for table_name, table in source_meta.tables.items():
            print(f"處理表: {table_name}")
            
            # 清空目標(biāo)表（生產(chǎn)環(huán)境應(yīng)考慮更安全的策略）
            target_session.execute(f"TRUNCATE TABLE {table_name}")
            
            # 查詢源數(shù)據(jù)
            result = source_engine.execute(table.select())
            rows = result.fetchall()
            
            if rows:
                # 批量插入
                insert_stmt = table.insert().values(rows)
                target_session.execute(insert_stmt)
                target_session.commit()
                
    except Exception as e:
        target_session.rollback()
        print(f"同步錯誤: {str(e)}")
    finally:
        target_session.close()

def incremental_sync(last_sync_time):
    try:
        source_conn = pymysql.connect(**SOURCE_DB)
        target_conn = pymysql.connect(**TARGET_DB)
        
        with source_conn.cursor() as src_cursor, target_conn.cursor() as tgt_cursor:
            # 假設(shè)所有表都有update_time字段
            src_cursor.execute("SHOW TABLES")
            tables = [table[0] for table in src_cursor.fetchall()]
            
            for table in tables:
                # 查詢增量數(shù)據(jù)
                query = f"""
                SELECT * FROM {table} 
                WHERE update_time > '{last_sync_time}'
                """
                src_cursor.execute(query)
                new_rows = src_cursor.fetchall()
                
                if new_rows:
                    columns = [desc[0] for desc in src_cursor.description]
                    placeholders = ', '.join(['%s'] * len(columns))
                    insert_sql = f"""
                    INSERT INTO {table} ({', '.join(columns)}) 
                    VALUES ({placeholders})
                    ON DUPLICATE KEY UPDATE
                    """ + ', '.join([f"{col}=VALUES({col})" for col in columns[1:]])
                    
                    tgt_cursor.executemany(insert_sql, new_rows)
                    target_conn.commit()
            
            # 更新最后同步時間（實際應(yīng)持久化存儲）
            current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            
    except Exception as e:
        print(f"增量同步失敗: {str(e)}")
    finally:
        source_conn.close()
        target_conn.close()

from pymysqlreplication import BinLogStreamReader
import pymysql

def binlog_sync():
    mysql_settings = {
        'host': SOURCE_DB['host'],
        'port': SOURCE_DB['port'],
        'user': SOURCE_DB['user'],
        'passwd': SOURCE_DB['password']
    }
    
    target_conn = pymysql.connect(**TARGET_DB)
    
    stream = BinLogStreamReader(
        mysql_settings,
        server_id=100,
        blocking=True,
        only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent]
    )
    
    try:
        for binlogevent in stream:
            binlogevent.dump()
            for row in binlogevent.rows:
                table = binlogevent.table
                event_type = binlogevent.__class__.__name__
                
                # 根據(jù)事件類型處理數(shù)據(jù)
                if event_type == "WriteRowsEvent":
                    # 處理插入
                    pass
                elif event_type == "UpdateRowsEvent":
                    # 處理更新
                    pass
                elif event_type == "DeleteRowsEvent":
                    # 處理刪除
                    pass
                    
    except KeyboardInterrupt:
        print("手動停止同步")
    finally:
        stream.close()
        target_conn.close()

from concurrent.futures import ThreadPoolExecutor
import pymysql

def sync_table(table_name, source_config, target_config):
    try:
        source_conn = pymysql.connect(**source_config)
        target_conn = pymysql.connect(**target_config)
        
        with source_conn.cursor() as src_cursor, target_conn.cursor() as tgt_cursor:
            # 實現(xiàn)單表同步邏輯...
            
    except Exception as e:
        print(f"表{table_name}同步失敗: {str(e)}")

def parallel_sync():
    source_conn = pymysql.connect(**SOURCE_DB)
    with source_conn.cursor() as cursor:
        cursor.execute("SHOW TABLES")
        tables = [table[0] for table in cursor.fetchall()]
    
    with ThreadPoolExecutor(max_workers=4) as executor:
        for table in tables:
            executor.submit(sync_table, table, SOURCE_DB, TARGET_DB)

def verify_sync(source_config, target_config):
    source_conn = pymysql.connect(**source_config)
    target_conn = pymysql.connect(**target_config)
    
    mismatches = []
    
    with source_conn.cursor() as src_cursor, target_conn.cursor() as tgt_cursor:
        src_cursor.execute("SHOW TABLES")
        tables = [table[0] for table in src_cursor.fetchall()]
        
        for table in tables:
            # 計算源表記錄數(shù)
            src_cursor.execute(f"SELECT COUNT(*) FROM {table}")
            src_count = src_cursor.fetchone()[0]
            
            # 計算目標(biāo)表記錄數(shù)
            tgt_cursor.execute(f"SELECT COUNT(*) FROM {table}")
            tgt_count = tgt_cursor.fetchone()[0]
            
            if src_count != tgt_count:
                mismatches.append((table, "記錄數(shù)不匹配", src_count, tgt_count))
            
            # 可選：抽樣校驗數(shù)據(jù)內(nèi)容...
    
    if mismatches:
        print("發(fā)現(xiàn)數(shù)據(jù)不一致:")
        for item in mismatches:
            print(item)
        return False
    return True

mysql_sync/
├── config/
│   ├── db_config.py       # 數(shù)據(jù)庫配置
│   └── logger_config.py   # 日志配置
├── core/
│   ├── sync_engine.py     # 核心同步邏輯
│   ├── verifier.py        # 數(shù)據(jù)校驗
│   └── utils.py           # 工具函數(shù)
├── scripts/
│   ├── full_sync.py       # 全量同步腳本
│   └── incremental.py     # 增量同步腳本
└── tests/
    └── test_sync.py        # 單元測試