完整数据导入方案

1. 项目结构准备
   bash
   复制
   your_project/
   ├── your_app/
   │ ├── management/
   │ │ └── commands/
   │ │ └── import_members.py # 自定义管理命令
   │ ├── models.py
   │ └── fixtures/
   │ └── members.json # 生成的JSON数据
   ├── db.sqlite3
   └── manage.py
2. 模型定义（models.py）
   python
   复制
   from django.db import models
   from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin

class Member(AbstractBaseUser, PermissionsMixin):
    username = models.CharField(max_length=255, unique=True)
    password = models.CharField(max_length=255)
    email = models.EmailField(max_length=50, unique=True, blank=True, null=True)
    phone = models.CharField(max_length=50, unique=True, blank=True, null=True)
    address = models.CharField(max_length=50, blank=True, null=True)
    realname = models.CharField(max_length=50, blank=True, null=True)
    nickname = models.CharField(max_length=50, blank=True, null=True)
    wx_openid = models.CharField(max_length=50, blank=True, null=True)
    avatar = models.CharField(max_length=100, blank=True, null=True)
    coordinate = models.CharField(max_length=50, blank=True, null=True)
    gender = models.CharField(max_length=50, blank=True, null=True)
    birthday = models.DateField(null=True, blank=True)
    mark = models.CharField(max_length=100, blank=True, null=True)
    is_active = models.BooleanField(default=True)
    is_staff = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    USERNAME_FIELD = 'username'
    REQUIRED_FIELDS = []

3. 自定义管理命令（import_members.py）

from django.core.management.base import BaseCommand
from django.contrib.auth.hashers import make_password
from your_app.models import Member
import json
from datetime import datetime
from django.db import transaction

class Command(BaseCommand):
    help = 'Import members from JSON file'

    def add_arguments(self, parser):
        parser.add_argument('file_path', type=str, help='Path to JSON file')

    @transaction.atomic
    def handle(self, *args, **options):
        file_path = options['file_path']
        processed = 0
        skipped = 0
        
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
                
                for item in data:
                    # 处理日期字段
                    try:
                        birthday = datetime.strptime(item['birthday'], '%Y-%m-%d').date() if item['birthday'] else None
                    except:
                        birthday = None
                        
                    # 检查唯一约束
                    if Member.objects.filter(username=item['username']).exists():
                        self.stdout.write(self.style.WARNING(
                            f"跳过重复用户: {item['username']}"
                        ))
                        skipped += 1
                        continue

                    # 创建记录
                    Member.objects.create(
                        username=item['username'],
                        password=make_password(item['password']),
                        phone=item['phone'],
                        email=item['email'] or None,  # 空字符串转为None
                        address=item['address'],
                        realname=item['realname'],
                        nickname=item['nickname'],
                        wx_openid=item['wx_openid'],
                        gender=item['gender'],
                        birthday=birthday,
                        mark=item['mark'],
                        is_active=item['is_active'],
                        is_staff=item['is_staff'],
                        created_at=datetime.fromisoformat(item['created_at'].replace('Z', '')),
                        # updated_at自动处理
                    )
                    processed += 1

            self.stdout.write(self.style.SUCCESS(
                f"成功导入 {processed} 条记录，跳过 {skipped} 条重复记录"
            ))

        except Exception as e:
            self.stdout.write(self.style.ERROR(f"导入失败: {str(e)}"))

4. 操作流程
   (1) 生成JSON数据
   使用之前改进的Python脚本生成members.json

(2) 放置数据文件
将生成的JSON文件放到项目目录：

bash
复制
cp members.json your_project/your_app/fixtures/
(3) 运行迁移
bash
复制
python manage.py makemigrations
python manage.py migrate
(4) 执行导入命令
bash
复制
python manage.py import_members your_app/fixtures/members.json
5. 关键增强功能
事务处理：

使用@transaction.atomic装饰器确保数据完整性

导入失败时自动回滚

数据清洗：

python
复制

处理空值

email=item[‘email’] or None # 将空字符串转为NULL

处理日期

birthday=datetime.strptime(…).date() if item[‘birthday’] else None
密码加密：

password=make_password(item['password'])

唯一性检查：

if Member.objects.filter(username=item['username']).exists():
    continue

错误处理：

捕获所有异常并输出错误信息

详细的日志输出

6. 高级配置建议
   批量创建优化：

# 使用bulk_create提升性能
batch_size = 100
objs = []
for item in data:
    objs.append(Member(...))
    if len(objs) >= batch_size:
        Member.objects.bulk_create(objs)
        objs = []
if objs:
    Member.objects.bulk_create(objs)
    ```
进度显示：

```python
from tqdm import tqdm
for item in tqdm(data, desc='处理进度'):
    # 处理逻辑

邮件通知：

from django.core.mail import send_mail

在导入完成后发送通知邮件

日志记录：

import logging
logger = logging.getLogger(__name__)
try:
    # 导入逻辑
except Exception as e:
    logger.error(f"导入失败: {str(e)}")
    ```
7. 验证导入结果
使用Django shell检查数据：

```bash
python manage.py shell

from your_app.models import Member
# 检查记录数量
print(Member.objects.count())

# 检查示例记录
member = Member.objects.first()
print(member.username, member.realname)

检查密码加密状态：

print(member.password)  # 应该显示加密后的哈希值

8. 常见问题处理
   日期格式错误：

在JSON生成阶段确保日期格式为ISO格式

添加更详细的日期解析错误提示

唯一约束冲突：

使用update_or_create方法处理重复数据

Member.objects.update_or_create(
    username=item['username'],
    defaults={...}
)

内存优化：

使用ijson库处理大文件

import ijson
for item in ijson.items(f, 'item'):

逐条处理

该方案提供了从数据准备到数据库导入的完整流程，包含错误处理、性能优化和安全措施，可直接用于生产环境。根据实际需求调整批量处理大小、日志级别和通知机制即可。