DataOps-platform/fix_duplicate_records.py

#!/usr/bin/env python3
"""
修复 duplicate_business_cards 表中 main_card_id 为 null 的损坏记录

使用方法:
    python fix_duplicate_records.py

描述:
    此脚本用于修复在处理重复记录时出现的数据完整性问题。
    当 main_card_id 字段为 null 时，会违反数据库的非空约束。
    
注意:
    - 此操作会永久删除损坏的记录
    - 建议在执行前备份数据库
    - 仅在确认数据损坏时使用
"""

import sys
import os

# 添加项目根目录到 Python 路径
project_root = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, project_root)

try:
    from app import create_app, db
    from app.core.data_parse.parse import DuplicateBusinessCard
    import logging
    from datetime import datetime
    
    # 设置日志
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
    logger = logging.getLogger(__name__)
    
    def fix_broken_records():
        """修复损坏的重复记录"""
        try:
            logger.info("开始修复损坏的重复记录...")
            
            # 查找所有 main_card_id 为 null 的记录
            broken_records = DuplicateBusinessCard.query.filter(
                DuplicateBusinessCard.main_card_id.is_(None)
            ).all()
            
            if not broken_records:
                logger.info("没有发现需要修复的损坏记录")
                return True
            
            logger.info(f"发现 {len(broken_records)} 条损坏记录")
            
            # 记录要删除的记录信息
            for i, record in enumerate(broken_records, 1):
                logger.info(f"损坏记录 {i}:")
                logger.info(f"  - ID: {record.id}")
                logger.info(f"  - 重复原因: {record.duplicate_reason}")
                logger.info(f"  - 处理状态: {record.processing_status}")
                logger.info(f"  - 创建时间: {record.created_at}")
                logger.info(f"  - 处理时间: {record.processed_at}")
            
            # 询问用户确认
            print(f"\n发现 {len(broken_records)} 条损坏的重复记录")
            print("这些记录的 main_card_id 字段为 null，违反了数据库约束")
            print("是否要删除这些损坏的记录？ (y/N): ", end="")
            
            confirm = input().strip().lower()
            if confirm not in ['y', 'yes']:
                logger.info("用户取消操作")
                return False
            
            # 删除损坏的记录
            for record in broken_records:
                db.session.delete(record)
            
            # 提交事务
            db.session.commit()
            
            logger.info(f"成功删除了 {len(broken_records)} 条损坏的重复记录")
            return True
            
        except Exception as e:
            logger.error(f"修复操作失败: {str(e)}")
            db.session.rollback()
            return False
    
    def main():
        """主函数"""
        print("=== 重复记录修复工具 ===")
        print("此工具用于修复 duplicate_business_cards 表中的损坏记录")
        print("=" * 50)
        
        # 创建应用上下文
        app = create_app()
        
        with app.app_context():
            success = fix_broken_records()
            
            if success:
                print("\n✅ 修复操作完成")
            else:
                print("\n❌ 修复操作失败或被取消")
                sys.exit(1)
    
    if __name__ == "__main__":
        main()
        
except ImportError as e:
    print(f"导入错误: {e}")
    print("请确保从项目根目录运行此脚本")
    sys.exit(1)
except Exception as e:
    print(f"执行错误: {e}")
    sys.exit(1)