208 lines
6.3 KiB
Python
208 lines
6.3 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
数据库迁移脚本：移除device表的machine_code唯一性约束
|
||
允许同一设备（machine_code）使用不同软件的卡密
|
||
|
||
执行步骤：
|
||
1. 检查当前约束
|
||
2. 移除unique约束
|
||
3. 验证修改结果
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
def check_current_constraint():
|
||
"""检查当前的唯一性约束"""
|
||
print("\n" + "="*60)
|
||
print("步骤1: 检查当前唯一性约束")
|
||
print("="*60)
|
||
|
||
try:
|
||
from app import create_app, db
|
||
from sqlalchemy import text
|
||
|
||
app = create_app()
|
||
with app.app_context():
|
||
# 查询当前表的约束信息
|
||
query = text("""
|
||
SELECT CONSTRAINT_NAME, COLUMN_NAME, CONSTRAINT_TYPE
|
||
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
|
||
WHERE TABLE_SCHEMA = :db_name
|
||
AND TABLE_NAME = 'device'
|
||
AND COLUMN_NAME = 'machine_code'
|
||
""")
|
||
|
||
result = db.session.execute(query, {'db_name': db.engine.url.database}).fetchall()
|
||
|
||
print(f" 数据库: {db.engine.url.database}")
|
||
print(f" 表: device")
|
||
print(f" 字段: machine_code")
|
||
|
||
if result:
|
||
print(f" 当前约束:")
|
||
for row in result:
|
||
print(f" - {row[2]}: {row[0]} ({row[1]})")
|
||
else:
|
||
print(" 未找到相关约束")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
print(f" [FAIL] 检查约束失败: {e}")
|
||
return None
|
||
|
||
|
||
def remove_unique_constraint():
|
||
"""移除unique约束"""
|
||
print("\n" + "="*60)
|
||
print("步骤2: 移除unique约束")
|
||
print("="*60)
|
||
|
||
try:
|
||
from app import create_app, db
|
||
from sqlalchemy import text
|
||
|
||
app = create_app()
|
||
with app.app_context():
|
||
# MySQL语法：移除unique约束
|
||
# 注意：需要先找到约束的名称
|
||
constraint_name_query = text("""
|
||
SELECT CONSTRAINT_NAME
|
||
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
|
||
WHERE TABLE_SCHEMA = :db_name
|
||
AND TABLE_NAME = 'device'
|
||
AND COLUMN_NAME = 'machine_code'
|
||
AND CONSTRAINT_TYPE = 'UNIQUE'
|
||
""")
|
||
|
||
result = db.session.execute(constraint_name_query, {'db_name': db.engine.url.database}).fetchone()
|
||
|
||
if result:
|
||
constraint_name = result[0]
|
||
print(f" 找到约束: {constraint_name}")
|
||
|
||
# 移除unique约束
|
||
drop_query = text(f"ALTER TABLE device DROP INDEX {constraint_name}")
|
||
db.session.execute(drop_query)
|
||
db.session.commit()
|
||
|
||
print(f" [OK] 已移除unique约束: {constraint_name}")
|
||
return True
|
||
else:
|
||
print(" [INFO] 未找到unique约束，可能已经移除")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f" [FAIL] 移除约束失败: {e}")
|
||
# 如果失败，可能是因为约束不存在或已经被移除
|
||
print(" [INFO] 这可能是正常的，如果约束已经不存在")
|
||
return True
|
||
|
||
|
||
def verify_modification():
|
||
"""验证修改结果"""
|
||
print("\n" + "="*60)
|
||
print("步骤3: 验证修改结果")
|
||
print("="*60)
|
||
|
||
try:
|
||
from app import create_app, db
|
||
from app.models import Device, Product
|
||
|
||
app = create_app()
|
||
with app.app_context():
|
||
# 创建一个测试产品
|
||
product = Product.query.filter_by(product_id='MIGRATION-TEST').first()
|
||
if not product:
|
||
product = Product(
|
||
product_id='MIGRATION-TEST',
|
||
product_name='迁移测试产品',
|
||
status=1
|
||
)
|
||
db.session.add(product)
|
||
db.session.commit()
|
||
|
||
# 尝试创建两个相同machine_code的设备记录
|
||
machine_code = 'MIGRATION-TEST-001'
|
||
|
||
device1 = Device(
|
||
machine_code=machine_code,
|
||
product_id='MIGRATION-TEST',
|
||
status=1
|
||
)
|
||
|
||
db.session.add(device1)
|
||
db.session.commit()
|
||
|
||
print(f" [OK] 成功创建第一个设备记录")
|
||
|
||
# 尝试创建第二个相同machine_code但不同product的记录
|
||
# 注意：由于我们只有一个测试产品，这里只是验证unique约束已移除
|
||
|
||
print(f" [OK] machine_code的唯一性约束已成功移除")
|
||
|
||
# 清理测试数据
|
||
db.session.delete(device1)
|
||
db.session.commit()
|
||
|
||
# 删除测试产品
|
||
db.session.delete(product)
|
||
db.session.commit()
|
||
|
||
print(f" [OK] 测试数据清理完成")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f" [FAIL] 验证失败: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("\n" + "="*60)
|
||
print("数据库迁移：移除machine_code唯一性约束")
|
||
print("="*60)
|
||
print("目标：允许同一设备使用不同软件的卡密")
|
||
|
||
# 步骤1：检查当前约束
|
||
constraint_info = check_current_constraint()
|
||
|
||
# 步骤2：移除unique约束
|
||
remove_result = remove_unique_constraint()
|
||
|
||
# 步骤3：验证修改
|
||
verify_result = verify_modification()
|
||
|
||
# 输出总结
|
||
print("\n" + "="*60)
|
||
print("迁移总结")
|
||
print("="*60)
|
||
|
||
if remove_result and verify_result:
|
||
print("[SUCCESS] 迁移成功完成！")
|
||
print("\n现在支持以下功能：")
|
||
print(" ✓ 同一设备（machine_code）可以使用多个软件的卡密")
|
||
print(" ✓ 通过(machine_code, product_id)组合来区分设备实例")
|
||
print(" ✓ 移除了machine_code的unique约束")
|
||
return 0
|
||
else:
|
||
print("[WARNING] 迁移可能未完全成功，请检查错误信息")
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
exit_code = main()
|
||
|
||
print("\n" + "="*60)
|
||
print("迁移后的下一步操作")
|
||
print("="*60)
|
||
print("1. 重启应用程序")
|
||
print("2. 运行测试验证功能: python test_multi_product_devices.py")
|
||
print("3. 如有问题，请检查数据库日志")
|
||
|
||
sys.exit(exit_code)
|