如何在Django中实现Leads到Contacts模型的数据迁移（含数据合并）？

两种主流实现方式 #

1. 自定义数据迁移文件（推荐） #

Django的迁移系统是处理模型变更和数据迁移的官方标准方案，适配这类需要持久化的操作，步骤如下：

• 第一步：生成空迁移文件
  在对应app目录下执行命令：

  python manage.py makemigrations --empty your_app_name
  

  这会生成一个空的迁移文件，格式类似000X_auto_xxxxxx_xxxx.py。

• 第二步：编写数据迁移逻辑
  打开生成的迁移文件，修改operations列表，添加RunPython操作来处理数据转换。示例代码如下：

  from django.db import migrations, models
  
  def migrate_leads_to_contacts(apps, schema_editor):
      # 通过apps获取模型，避免直接导入导致的依赖问题
      Leads = apps.get_model('your_app_name', 'Leads')
      Contacts = apps.get_model('your_app_name', 'Contacts')
  
      # 按name和phone分组，合并重复记录
      grouped_leads = Leads.objects.values('name', 'phone').annotate(
          # 这里用Max取最新的记录值，可根据需求替换为拼接、取第一个等逻辑
          latest_email=models.Max('email'),
          latest_address=models.Max('address')
      )
  
      # 批量创建Contact记录，提升效率
      contacts_to_create = []
      for group in grouped_leads:
          contacts_to_create.append(Contacts(
              name=group['name'],
              phone=group['phone'],
              email=group['latest_email'],
              address=group['latest_address']
          ))
      Contacts.objects.bulk_create(contacts_to_create)
  
  class Migration(migrations.Migration):
      dependencies = [
          # 依赖Contacts模型的迁移文件，替换为实际的迁移编号
          ('your_app_name', '000Y_contacts'),
      ]
  
      operations = [
          migrations.RunPython(migrate_leads_to_contacts),
      ]
  

  注：合并逻辑可按需调整，比如需要合并多个邮箱/地址，可改用', '.join(...)拼接对应字段的所有值。

• 第三步：执行迁移
  运行命令完成数据迁移：

  python manage.py migrate
  

  这种方式的核心优势：

  • 符合Django最佳实践，迁移操作可追溯、支持回滚（编写反向函数即可）
  • 批量操作效率高，适配大数据量场景
  • 操作记录在django_migrations表中，团队协作时不会重复执行或遗漏

2. 独立脚本/Shell命令（临时场景） #

如果只是一次性临时处理，不需要持久化迁移记录，可采用这种方式：

• 方式一：Django Shell
  运行python manage.py shell，然后执行类似代码：

  from your_app.models import Leads, Contacts
  from django.db.models import Max
  
  grouped = Leads.objects.values('name', 'phone').annotate(latest_email=Max('email'), latest_address=Max('address'))
  for g in grouped:
      Contacts.objects.get_or_create(name=g['name'], phone=g['phone'], defaults={'email': g['latest_email'], 'address': g['latest_address']})
  

• 方式二：独立脚本
  编写脚本（如migrate_data.py），开头初始化Django环境：

  import os
  import django
  
  os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
  django.setup()
  
  from your_app.models import Leads, Contacts
  from django.db.models import Max
  
  grouped_leads = Leads.objects.values('name', 'phone').annotate(
      latest_email=Max('email'),
      latest_address=Max('address')
  )
  
  contacts_to_create = []
  for group in grouped_leads:
      contacts_to_create.append(Contacts(
          name=group['name'],
          phone=group['phone'],
          email=group['latest_email'],
          address=group['latest_address']
      ))
  Contacts.objects.bulk_create(contacts_to_create)
  

  执行脚本：python migrate_data.py

  这种方式的优势：

  • 快速简单，无需创建迁移文件
  • 适合临时测试或小数据量场景

哪种方式更优？ #

优先选择自定义数据迁移文件，原因如下：

• 迁移操作可审计，团队成员执行migrate时会自动同步数据，避免遗漏或重复执行
• 支持反向迁移，便于数据回滚
• 批量操作性能更优，迁移系统会自动处理数据库事务
• 符合Django生态规范，后续维护更便捷

仅在临时处理少量数据、不需要持久化迁移记录的场景下，才考虑使用独立脚本或Shell命令。

内容的提问来源于stack exchange，提问作者tarp20