使用Python自制数据库备份工具实现数据定时覆盖

pyinstaller --onefile --icon=my_icon.ico your_script.py

import schedule
import time
import pandas as pd
from sqlalchemy import create_engine, text

# 生产数据库配置
prod_db_config = {
    'username': 'root',
    'password': 'root',
    'ip': '127.0.0.1',
    'database': 'test1',
    'port': 3306
}

# 测试数据库配置
test_db_config = {
    'username': 'root',
    'password': 'root',
    'ip': '127.0.0.1',
    'database': 'test2',
    'port': 3306
}


def update_test_db():
    try:
        # 构建测试数据库连接字符串
        test_db_url = f"mysql://{test_db_config['username']}:{test_db_config['password']}@{test_db_config['ip']}:{test_db_config['port']}/{test_db_config['database']}?charset=utf8mb4"

        # 创建测试数据库引擎
        test_engine = create_engine(test_db_url)

        # 测试连接是否成功
        with test_engine.connect() as conn:
            conn.execute(text("select 1"))
        print("成功连接到测试数据库.")

        # 清除测试数据库中的所有表
        with test_engine.connect() as conn:
            tables = conn.execute("SHOW TABLES").fetchall()
            if tables:
                for table in tables:
                    conn.execute(f"DROP TABLE IF EXISTS {table[0]}")

        # 获取生产数据库中的所有表
        prod_db_url = f"mysql://{prod_db_config['username']}:{prod_db_config['password']}@{prod_db_config['ip']}:{prod_db_config['port']}/{prod_db_config['database']}?charset=utf8mb4"
        prod_engine = create_engine(prod_db_url)

        with prod_engine.connect() as prod_conn:
            tables = prod_conn.execute("SHOW TABLES").fetchall()
            print("从生产数据库中获取表完成.")

            # 将生产数据库中的所有表数据备份到测试数据库
            for table in tables:
                table_name = table[0]
                print(f"备份表 {table_name} 数据...")
                data = pd.read_sql_table(table_name, prod_conn)
                data.to_sql(table_name, test_engine, index=False, if_exists='replace')
                print(f"表 {table_name} 数据备份完成.")

        print("测试数据库更新成功.")
    except Exception as e:
        print("更新测试数据库时出错:", e)

if __name__ == "__main__":
    # 将任务调度在每天特定时间执行
    schedule.every().day.at("23:22").do(update_test_db)

    # 运行定时任务
    while True:
        schedule.run_pending()
        time.sleep(60)