simple-dibang/migrate_sqlite_to_mysql.py

"""
SQLite数据迁移到MySQL脚本
将SQLite数据库中的所有数据导出并导入到MySQL数据库
"""
import sqlite3
import sys
import os
from datetime import datetime

# 尝试导入MySQL相关模块
try:
    import pymysql
    PYMySQL_AVAILABLE = True
except ImportError:
    PYMySQL_AVAILABLE = False
    print("错误: 请先安装pymysql: pip install pymysql")
    sys.exit(1)

# 尝试导入配置模块
try:
    import config
    MYSQL_CONFIG = getattr(config, 'MYSQL_CONFIG', {})
except ImportError:
    print("错误: 找不到config.py配置文件")
    sys.exit(1)


def get_base_path():
    """获取程序运行的基础路径"""
    if getattr(sys, 'frozen', False):
        return os.path.dirname(sys.executable)
    else:
        return os.path.dirname(os.path.abspath(__file__))


def get_sqlite_db_path():
    """获取SQLite数据库路径"""
    base_path = get_base_path()
    return os.path.join(base_path, "dibang.db")


def connect_mysql():
    """连接MySQL数据库"""
    try:
        conn = pymysql.connect(
            host=MYSQL_CONFIG.get('host', 'localhost'),
            port=MYSQL_CONFIG.get('port', 3306),
            user=MYSQL_CONFIG.get('user', 'root'),
            password=MYSQL_CONFIG.get('password', ''),
            database=MYSQL_CONFIG.get('database', 'dibang'),
            charset=MYSQL_CONFIG.get('charset', 'utf8mb4')
        )
        return conn
    except Exception as e:
        print(f"MySQL连接失败: {str(e)}")
        return None


def export_data_to_sql_file(sqlite_path, output_file='sqlite_export.sql'):
    """
    导出SQLite数据到SQL文件
    生成可以直接在MySQL中执行的INSERT语句
    """
    if not os.path.exists(sqlite_path):
        print(f"错误: SQLite数据库文件不存在: {sqlite_path}")
        return False
    
    conn_sqlite = sqlite3.connect(sqlite_path)
    conn_sqlite.row_factory = sqlite3.Row
    cursor_sqlite = conn_sqlite.cursor()
    
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write("-- SQLite数据导出到MySQL\n")
        f.write(f"-- 导出时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
        f.write("-- 禁用外键检查（如果需要）\n")
        f.write("SET FOREIGN_KEY_CHECKS = 0;\n\n")
        f.write("-- 清空表数据（可选，谨慎使用）\n")
        f.write("-- TRUNCATE TABLE companies;\n")
        f.write("-- TRUNCATE TABLE cargos;\n")
        f.write("-- TRUNCATE TABLE purchase_plans;\n")
        f.write("-- TRUNCATE TABLE weighing_slips;\n\n")
        
        # 导出公司表
        print("正在导出公司表数据...")
        cursor_sqlite.execute("SELECT * FROM companies ORDER BY id")
        companies = cursor_sqlite.fetchall()
        f.write("-- 公司表数据\n")
        for row in companies:
            name = str(row['name']).replace("'", "\\'").replace("\\", "\\\\")
            created_at = str(row['created_at'])
            f.write(f"INSERT INTO companies (name, created_at) VALUES ('{name}', '{created_at}');\n")
        f.write(f"-- 共 {len(companies)} 条公司记录\n\n")
        print(f"  导出了 {len(companies)} 条公司记录")
        
        # 导出货物表
        print("正在导出货物表数据...")
        cursor_sqlite.execute("SELECT * FROM cargos ORDER BY id")
        cargos = cursor_sqlite.fetchall()
        f.write("-- 货物表数据\n")
        for row in cargos:
            name = str(row['name']).replace("'", "\\'").replace("\\", "\\\\")
            created_at = str(row['created_at'])
            f.write(f"INSERT INTO cargos (name, created_at) VALUES ('{name}', '{created_at}');\n")
        f.write(f"-- 共 {len(cargos)} 条货物记录\n\n")
        print(f"  导出了 {len(cargos)} 条货物记录")
        
        # 导出采购计划表
        print("正在导出采购计划表数据...")
        cursor_sqlite.execute("SELECT * FROM purchase_plans ORDER BY id")
        plans = cursor_sqlite.fetchall()
        f.write("-- 采购计划表数据\n")
        for row in plans:
            plan_date = str(row['plan_date'])
            business_type = str(row['business_type']).replace("'", "\\'").replace("\\", "\\\\")
            product = str(row['product']).replace("'", "\\'").replace("\\", "\\\\")
            specification = str(row.get('specification', '') or '').replace("'", "\\'").replace("\\", "\\\\")
            planned_quantity = row['planned_quantity']
            vehicle_no = str(row['vehicle_no']).replace("'", "\\'").replace("\\", "\\\\")
            customer_name = str(row.get('customer_name', '') or '').replace("'", "\\'").replace("\\", "\\\\")
            remarks = str(row.get('remarks', '') or '').replace("'", "\\'").replace("\\", "\\\\")
            created_at = str(row['created_at'])
            
            f.write(f"INSERT INTO purchase_plans (plan_date, business_type, product, specification, "
                   f"planned_quantity, vehicle_no, customer_name, remarks, created_at) VALUES "
                   f"('{plan_date}', '{business_type}', '{product}', '{specification}', "
                   f"{planned_quantity}, '{vehicle_no}', '{customer_name}', '{remarks}', '{created_at}');\n")
        f.write(f"-- 共 {len(plans)} 条采购计划记录\n\n")
        print(f"  导出了 {len(plans)} 条采购计划记录")
        
        # 导出地磅单表
        print("正在导出地磅单表数据...")
        cursor_sqlite.execute("SELECT * FROM weighing_slips ORDER BY id")
        slips = cursor_sqlite.fetchall()
        f.write("-- 地磅单表数据\n")
        for row in slips:
            vehicle_no = str(row['vehicle_no']).replace("'", "\\'").replace("\\", "\\\\")
            plan_no = str(row.get('plan_no', '') or '').replace("'", "\\'").replace("\\", "\\\\")
            cargo_name = str(row['cargo_name']).replace("'", "\\'").replace("\\", "\\\\")
            business_type = str(row['business_type']).replace("'", "\\'").replace("\\", "\\\\")
            batch_no = str(row.get('batch_no', '') or '').replace("'", "\\'").replace("\\", "\\\\")
            supplier = str(row['supplier']).replace("'", "\\'").replace("\\", "\\\\")
            consignee = str(row['consignee']).replace("'", "\\'").replace("\\", "\\\\")
            gross_weight = row['gross_weight']
            gross_weight_time = str(row['gross_weight_time'])
            tare_weight = row['tare_weight']
            tare_weight_time = str(row['tare_weight_time'])
            deduction = row.get('deduction', 0) or 0
            net_weight = row['net_weight']
            planned_quantity = row.get('planned_quantity') if row.get('planned_quantity') is not None else 'NULL'
            executed_quantity = row.get('executed_quantity') if row.get('executed_quantity') is not None else 'NULL'
            weighing_point = str(row.get('weighing_point', '山峡地磅') or '山峡地磅').replace("'", "\\'").replace("\\", "\\\\")
            remarks = str(row.get('remarks', '') or '').replace("'", "\\'").replace("\\", "\\\\")
            is_voided = row.get('is_voided', 0) or 0
            created_at = str(row['created_at'])
            
            planned_qty = f"'{planned_quantity}'" if planned_quantity != 'NULL' else 'NULL'
            executed_qty = f"'{executed_quantity}'" if executed_quantity != 'NULL' else 'NULL'
            
            f.write(f"INSERT INTO weighing_slips (vehicle_no, plan_no, cargo_name, business_type, batch_no, "
                   f"supplier, consignee, gross_weight, gross_weight_time, tare_weight, tare_weight_time, "
                   f"deduction, net_weight, planned_quantity, executed_quantity, weighing_point, remarks, "
                   f"is_voided, created_at) VALUES "
                   f"('{vehicle_no}', '{plan_no}', '{cargo_name}', '{business_type}', '{batch_no}', "
                   f"'{supplier}', '{consignee}', {gross_weight}, '{gross_weight_time}', {tare_weight}, "
                   f"'{tare_weight_time}', {deduction}, {net_weight}, {planned_qty}, {executed_qty}, "
                   f"'{weighing_point}', '{remarks}', {is_voided}, '{created_at}');\n")
        f.write(f"-- 共 {len(slips)} 条地磅单记录\n\n")
        print(f"  导出了 {len(slips)} 条地磅单记录")
        
        f.write("-- 启用外键检查\n")
        f.write("SET FOREIGN_KEY_CHECKS = 1;\n")
    
    conn_sqlite.close()
    print(f"\n数据已导出到文件: {output_file}")
    return True


def migrate_data_directly(sqlite_path):
    """
    直接将SQLite数据迁移到MySQL
    注意：这会清空MySQL中的现有数据并重新导入
    """
    if not os.path.exists(sqlite_path):
        print(f"错误: SQLite数据库文件不存在: {sqlite_path}")
        return False
    
    # 连接MySQL
    conn_mysql = connect_mysql()
    if conn_mysql is None:
        return False
    
    # 连接SQLite
    conn_sqlite = sqlite3.connect(sqlite_path)
    conn_sqlite.row_factory = sqlite3.Row
    cursor_sqlite = conn_sqlite.cursor()
    cursor_mysql = conn_mysql.cursor()
    
    try:
        print("\n开始迁移数据到MySQL...")
        print("警告: 这将清空MySQL中的所有数据！")
        
        # 确认
        confirm = input("是否继续? (yes/no): ")
        if confirm.lower() != 'yes':
            print("已取消迁移")
            return False
        
        # 清空MySQL表（注意顺序，因为有外键依赖）
        print("\n正在清空MySQL表...")
        cursor_mysql.execute("SET FOREIGN_KEY_CHECKS = 0")
        cursor_mysql.execute("TRUNCATE TABLE weighing_slips")
        cursor_mysql.execute("TRUNCATE TABLE purchase_plans")
        cursor_mysql.execute("TRUNCATE TABLE companies")
        cursor_mysql.execute("TRUNCATE TABLE cargos")
        cursor_mysql.execute("SET FOREIGN_KEY_CHECKS = 1")
        conn_mysql.commit()
        print("MySQL表已清空")
        
        # 迁移公司表
        print("\n正在迁移公司表数据...")
        cursor_sqlite.execute("SELECT * FROM companies ORDER BY id")
        companies = cursor_sqlite.fetchall()
        for row in companies:
            cursor_mysql.execute(
                "INSERT INTO companies (name, created_at) VALUES (%s, %s)",
                (row['name'], row['created_at'])
            )
        conn_mysql.commit()
        print(f"  迁移了 {len(companies)} 条公司记录")
        
        # 迁移货物表
        print("正在迁移货物表数据...")
        cursor_sqlite.execute("SELECT * FROM cargos ORDER BY id")
        cargos = cursor_sqlite.fetchall()
        for row in cargos:
            cursor_mysql.execute(
                "INSERT INTO cargos (name, created_at) VALUES (%s, %s)",
                (row['name'], row['created_at'])
            )
        conn_mysql.commit()
        print(f"  迁移了 {len(cargos)} 条货物记录")
        
        # 迁移采购计划表
        print("正在迁移采购计划表数据...")
        cursor_sqlite.execute("SELECT * FROM purchase_plans ORDER BY id")
        plans = cursor_sqlite.fetchall()
        for row in plans:
            cursor_mysql.execute(
                """INSERT INTO purchase_plans (plan_date, business_type, product, specification,
                   planned_quantity, vehicle_no, customer_name, remarks, created_at)
                   VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)""",
                (
                    row['plan_date'], row['business_type'], row['product'],
                    row.get('specification', '') or '', row['planned_quantity'],
                    row['vehicle_no'], row.get('customer_name', '') or '',
                    row.get('remarks', '') or '', row['created_at']
                )
            )
        conn_mysql.commit()
        print(f"  迁移了 {len(plans)} 条采购计划记录")
        
        # 迁移地磅单表
        print("正在迁移地磅单表数据...")
        cursor_sqlite.execute("SELECT * FROM weighing_slips ORDER BY id")
        slips = cursor_sqlite.fetchall()
        for row in slips:
            cursor_mysql.execute(
                """INSERT INTO weighing_slips (vehicle_no, plan_no, cargo_name, business_type, batch_no,
                   supplier, consignee, gross_weight, gross_weight_time, tare_weight, tare_weight_time,
                   deduction, net_weight, planned_quantity, executed_quantity, weighing_point, remarks,
                   is_voided, created_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
                (
                    row['vehicle_no'], row.get('plan_no', '') or '', row['cargo_name'],
                    row['business_type'], row.get('batch_no', '') or '', row['supplier'],
                    row['consignee'], row['gross_weight'], row['gross_weight_time'],
                    row['tare_weight'], row['tare_weight_time'], row.get('deduction', 0) or 0,
                    row['net_weight'], row.get('planned_quantity'), row.get('executed_quantity'),
                    row.get('weighing_point', '山峡地磅') or '山峡地磅',
                    row.get('remarks', '') or '', row.get('is_voided', 0) or 0,
                    row['created_at']
                )
            )
        conn_mysql.commit()
        print(f"  迁移了 {len(slips)} 条地磅单记录")
        
        print("\n✅ 数据迁移完成！")
        print(f"  公司: {len(companies)} 条")
        print(f"  货物: {len(cargos)} 条")
        print(f"  采购计划: {len(plans)} 条")
        print(f"  地磅单: {len(slips)} 条")
        
        conn_sqlite.close()
        conn_mysql.close()
        return True
        
    except Exception as e:
        print(f"\n❌ 迁移失败: {str(e)}")
        conn_mysql.rollback()
        conn_sqlite.close()
        conn_mysql.close()
        return False


def main():
    """主函数"""
    sqlite_path = get_sqlite_db_path()
    
    print("=" * 60)
    print("SQLite数据迁移到MySQL工具")
    print("=" * 60)
    print(f"SQLite数据库: {sqlite_path}")
    print(f"MySQL数据库: {MYSQL_CONFIG.get('host')}:{MYSQL_CONFIG.get('port')}/{MYSQL_CONFIG.get('database')}")
    print("=" * 60)
    print("\n请选择操作:")
    print("1. 导出SQLite数据为SQL文件（推荐，可手动检查后再导入）")
    print("2. 直接将SQLite数据迁移到MySQL（会清空MySQL现有数据）")
    print("3. 退出")
    
    choice = input("\n请输入选项 (1/2/3): ").strip()
    
    if choice == '1':
        output_file = input("\n请输入导出文件名（默认: sqlite_export.sql）: ").strip()
        if not output_file:
            output_file = 'sqlite_export.sql'
        if export_data_to_sql_file(sqlite_path, output_file):
            print(f"\n✅ 导出成功！")
            print(f"你可以使用以下命令导入MySQL:")
            print(f"  mysql -h {MYSQL_CONFIG.get('host')} -u {MYSQL_CONFIG.get('user')} -p {MYSQL_CONFIG.get('database')} < {output_file}")
        else:
            print("\n❌ 导出失败！")
    
    elif choice == '2':
        if migrate_data_directly(sqlite_path):
            print("\n迁移完成！")
        else:
            print("\n迁移失败！")
    
    elif choice == '3':
        print("已退出")
    else:
        print("无效选项")


if __name__ == '__main__':
    main()