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

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

# 生產(chǎn)數(shù)據(jù)庫配置
prod_db_config = {
    'username': 'root',
    'password': 'root',
    'ip': '127.0.0.1',
    'database': 'test1',
    'port': 3306
}

# 測試數(shù)據(jù)庫配置
test_db_config = {
    'username': 'root',
    'password': 'root',
    'ip': '127.0.0.1',
    'database': 'test2',
    'port': 3306
}


def update_test_db():
    try:
        # 構(gòu)建測試數(shù)據(jù)庫連接字符串
        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"

        # 創(chuàng)建測試數(shù)據(jù)庫引擎
        test_engine = create_engine(test_db_url)

        # 測試連接是否成功
        with test_engine.connect() as conn:
            conn.execute(text("select 1"))
        print("成功連接到測試數(shù)據(jù)庫.")

        # 清除測試數(shù)據(jù)庫中的所有表
        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]}")

        # 獲取生產(chǎn)數(shù)據(jù)庫中的所有表
        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("從生產(chǎn)數(shù)據(jù)庫中獲取表完成.")

            # 將生產(chǎn)數(shù)據(jù)庫中的所有表數(shù)據(jù)備份到測試數(shù)據(jù)庫
            for table in tables:
                table_name = table[0]
                print(f"備份表 {table_name} 數(shù)據(jù)...")
                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} 數(shù)據(jù)備份完成.")

        print("測試數(shù)據(jù)庫更新成功.")
    except Exception as e:
        print("更新測試數(shù)據(jù)庫時出錯:", e)

if __name__ == "__main__":
    # 將任務(wù)調(diào)度在每天特定時間執(zhí)行
    schedule.every().day.at("23:22").do(update_test_db)

    # 運行定時任務(wù)
    while True:
        schedule.run_pending()
        time.sleep(60)