#Django用户认证系统 - 安全登录注册与权限管理

"""
认证(Authentication) vs 授权(Authorization):

认证(Authentication):
- 验证用户身份
- 证明"你是谁"
- 如：用户名密码验证
- Django内置认证系统

授权(Authorization):
- 验证用户权限
- 证明"你能做什么"
- 如：角色权限检查
- Django权限系统

认证流程：
1. 用户提供凭据（用户名/密码）
2. 系统验证凭据有效性
3. 创建用户会话
4. 标记用户为已认证

授权流程：
1. 检查用户认证状态
2. 验证用户权限
3. 允许或拒绝访问
4. 记录访问日志
"""

"""
Django认证系统组成部分：

1. User模型 (django.contrib.auth.models.User)
   - 用户基本信息存储
   - 密码哈希存储
   - 权限字段

2. Authentication Backend (django.contrib.auth.backends)
   - 用户验证逻辑
   - 支持多种认证方式

3. Permission System
   - 用户权限管理
   - 内容级权限控制

4. Group System
   - 用户分组管理
   - 批量权限分配

5. Login/Logout Views
   - 内置登录/登出视图
   - 会话管理

6. Decorators & Mixins
   - 视图保护装饰器
   - 类视图权限检查
"""

# Django认证核心组件
from django.contrib.auth import authenticate, login, logout
from django.contrib.auth.models import User, Group, Permission
from django.contrib.auth.backends import ModelBackend
from django.contrib.auth.decorators import login_required, permission_required
from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin

"""
主要功能模块：

1. 用户管理
   - 创建、修改、删除用户
   - 密码管理
   - 激活/禁用用户

2. 权限管理
   - 模型级权限
   - 对象级权限
   - 自定义权限

3. 组管理
   - 用户分组
   - 权限批量分配
   - 角色管理

4. 会话管理
   - 登录/登出
   - 会话安全
   - 多设备登录
"""

# 认证中间件配置
from django.contrib.auth.middleware import AuthenticationMiddleware
from django.contrib.sessions.middleware import SessionMiddleware

class CustomAuthenticationMiddleware(AuthenticationMiddleware):
    """自定义认证中间件"""
    
    def process_request(self, request):
        """处理请求 - 认证用户"""
        super().process_request(request)
        
        # 额外的认证逻辑
        if hasattr(request, 'user') and request.user.is_authenticated:
            # 更新最后活动时间
            self.update_last_activity(request.user)
            
            # 检查账户状态
            if not self.is_account_valid(request.user):
                # 账户无效，强制登出
                logout(request)
    
    def update_last_activity(self, user):
        """更新用户最后活动时间"""
        from django.utils import timezone
        user.last_login = timezone.now()
        user.save(update_fields=['last_login'])
    
    def is_account_valid(self, user):
        """检查账户是否有效"""
        # 检查账户是否被禁用
        if not user.is_active:
            return False
        
        # 检查账户是否过期
        if hasattr(user, 'expiry_date') and user.expiry_date:
            from django.utils import timezone
            if user.expiry_date < timezone.now():
                return False
        
        return True

# 认证后端接口
from django.contrib.auth.backends import BaseBackend

class CustomAuthBackend(BaseBackend):
    """自定义认证后端"""
    
    def authenticate(self, request, username=None, password=None, **kwargs):
        """认证用户"""
        if username is None or password is None:
            return None
        
        try:
            # 自定义查询逻辑
            user = self.get_user_by_username(username)
            
            if user and self.check_password(password, user.password):
                return user
        except Exception:
            # 记录认证失败日志
            import logging
            logger = logging.getLogger('auth.failed')
            logger.warning(f"Authentication failed for username: {username}")
        
        return None
    
    def get_user_by_username(self, username):
        """根据用户名获取用户"""
        try:
            from django.contrib.auth.models import User
            return User.objects.get(username=username)
        except User.DoesNotExist:
            return None
    
    def check_password(self, raw_password, hashed_password):
        """检查密码"""
        from django.contrib.auth.hashers import check_password
        return check_password(raw_password, hashed_password)
    
    def get_user(self, user_id):
        """获取用户"""
        try:
            from django.contrib.auth.models import User
            return User.objects.get(pk=user_id)
        except User.DoesNotExist:
            return None

# 认证处理器
class AuthHandler:
    """认证处理器"""
    
    def __init__(self):
        self.backends = self.get_backends()
    
    def get_backends(self):
        """获取认证后端"""
        from django.conf import settings
        from django.utils.module_loading import import_string
        
        backends = []
        for backend_path in settings.AUTHENTICATION_BACKENDS:
            backend_cls = import_string(backend_path)
            backends.append(backend_cls())
        
        return backends
    
    def authenticate_user(self, **credentials):
        """认证用户"""
        for backend in self.backends:
            user = backend.authenticate(None, **credentials)
            if user:
                # 设置后端信息
                user.backend = f"{backend.__class__.__module__}.{backend.__class__.__name__}"
                return user
        return None

# 使用示例
def custom_authenticate(username, password):
    """自定义认证函数"""
    handler = AuthHandler()
    return handler.authenticate_user(username=username, password=password)

# 用户会话管理
from django.contrib.auth import login, logout
from django.contrib.sessions.models import Session
from django.contrib.auth.models import User
from django.utils import timezone

class UserSessionManager:
    """用户会话管理器"""
    
    @staticmethod
    def create_user_session(request, user, remember_me=False):
        """创建用户会话"""
        # 登录用户
        login(request, user)
        
        # 设置会话过期时间
        if not remember_me:
            request.session.set_expiry(0)  # 浏览器关闭时过期
        else:
            request.session.set_expiry(1209600)  # 2周后过期
        
        # 记录登录信息
        UserSessionManager.record_login_info(user, request)
    
    @staticmethod
    def record_login_info(user, request):
        """记录登录信息"""
        # 更新用户最后登录时间
        user.last_login = timezone.now()
        user.save(update_fields=['last_login'])
        
        # 记录登录日志
        login_log = UserLoginLog.objects.create(
            user=user,
            ip_address=request.META.get('REMOTE_ADDR'),
            user_agent=request.META.get('HTTP_USER_AGENT', ''),
            login_time=timezone.now()
        )
        
        # 清除之前的失败登录记录
        UserFailedLoginAttempt.objects.filter(
            user=user,
            attempt_time__lt=timezone.now() - timezone.timedelta(hours=1)
        ).delete()
    
    @staticmethod
    def logout_user(request):
        """登出用户"""
        if request.user.is_authenticated:
            # 记录登出信息
            UserSessionManager.record_logout_info(request.user, request)
        
        # 执行登出
        logout(request)
    
    @staticmethod
    def record_logout_info(user, request):
        """记录登出信息"""
        UserLogoutLog.objects.create(
            user=user,
            ip_address=request.META.get('REMOTE_ADDR'),
            logout_time=timezone.now()
        )

# 登录日志模型
from django.db import models

class UserLoginLog(models.Model):
    """用户登录日志"""
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    ip_address = models.GenericIPAddressField()
    user_agent = models.TextField(blank=True)
    login_time = models.DateTimeField(auto_now_add=True)
    success = models.BooleanField(default=True)
    
    class Meta:
        db_table = 'user_login_log'
        ordering = ['-login_time']

class UserLogoutLog(models.Model):
    """用户登出日志"""
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    ip_address = models.GenericIPAddressField()
    logout_time = models.DateTimeField(auto_now_add=True)
    
    class Meta:
        db_table = 'user_logout_log'
        ordering = ['-logout_time']

class UserFailedLoginAttempt(models.Model):
    """用户失败登录尝试"""
    username = models.CharField(max_length=150)
    ip_address = models.GenericIPAddressField()
    attempt_time = models.DateTimeField(auto_now_add=True)
    user_agent = models.TextField(blank=True)
    
    class Meta:
        db_table = 'user_failed_login_attempts'
        ordering = ['-attempt_time']

# 会话安全检查
class SessionSecurityChecker:
    """会话安全检查器"""
    
    @staticmethod
    def check_brute_force_attack(username, ip_address, max_attempts=5):
        """检查暴力破解攻击"""
        from django.utils import timezone
        from django.conf import settings
        
        # 检查相同IP的失败尝试
        recent_attempts = UserFailedLoginAttempt.objects.filter(
            ip_address=ip_address,
            attempt_time__gte=timezone.now() - timezone.timedelta(
                seconds=getattr(settings, 'BRUTE_FORCE_TIME_WINDOW', 900)  # 15分钟
            )
        ).count()
        
        if recent_attempts >= getattr(settings, 'MAX_LOGIN_ATTEMPTS_PER_IP', 10):
            return True, "IP地址被临时封禁"
        
        # 检查特定用户的失败尝试
        user_attempts = UserFailedLoginAttempt.objects.filter(
            username=username,
            attempt_time__gte=timezone.now() - timezone.timedelta(
                seconds=getattr(settings, 'BRUTE_FORCE_TIME_WINDOW', 900)
            )
        ).count()
        
        if user_attempts >= max_attempts:
            return True, "账户被临时锁定"
        
        return False, ""
    
    @staticmethod
    def log_failed_attempt(username, ip_address, user_agent):
        """记录失败登录尝试"""
        UserFailedLoginAttempt.objects.create(
            username=username,
            ip_address=ip_address,
            user_agent=user_agent
        )

# 使用示例
def secure_login_view(request):
    """安全登录视图"""
    if request.method == 'POST':
        username = request.POST.get('username')
        password = request.POST.get('password')
        remember_me = request.POST.get('remember_me', False)
        
        # 检查暴力破解
        ip_address = request.META.get('REMOTE_ADDR')
        user_agent = request.META.get('HTTP_USER_AGENT', '')
        
        is_attack, attack_msg = SessionSecurityChecker.check_brute_force_attack(
            username, ip_address
        )
        
        if is_attack:
            return {'error': attack_msg, 'blocked': True}
        
        # 尝试认证
        user = authenticate(request, username=username, password=password)
        if user:
            if user.is_active:
                # 创建安全会话
                UserSessionManager.create_user_session(
                    request, user, remember_me
                )
                return {'success': True, 'redirect': '/dashboard/'}
            else:
                return {'error': '账户已被禁用'}
        else:
            # 记录失败尝试
            SessionSecurityChecker.log_failed_attempt(
                username, ip_address, user_agent
            )
            return {'error': '用户名或密码错误'}
    
    return {'success': False}

# 认证装饰器
from functools import wraps
from django.contrib.auth.decorators import login_required, permission_required
from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin
from django.http import HttpResponseForbidden, HttpResponseRedirect
from django.urls import reverse
from django.contrib import messages

def custom_login_required(function=None, redirect_field_name='next', login_url=None):
    """自定义登录要求装饰器"""
    def decorator(view_func):
        @wraps(view_func)
        def _wrapped_view(request, *args, **kwargs):
            if not request.user.is_authenticated:
                # 检查是否为AJAX请求
                if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
                    return HttpResponseForbidden("Authentication required")
                
                # 重定向到登录页面
                login_url = login_url or reverse('login')
                next_url = request.get_full_path()
                redirect_url = f"{login_url}?{redirect_field_name}={next_url}"
                return HttpResponseRedirect(redirect_url)
            
            # 检查用户账户状态
            if hasattr(request.user, 'is_account_active') and not request.user.is_account_active():
                messages.error(request, '您的账户已被禁用')
                logout(request)
                return HttpResponseRedirect(reverse('login'))
            
            return view_func(request, *args, **kwargs)
        return _wrapped_view
    
    if function:
        return decorator(function)
    return decorator

def role_required(roles, login_url=None, redirect_field_name='next'):
    """角色要求装饰器"""
    def decorator(view_func):
        @wraps(view_func)
        def _wrapped_view(request, *args, **kwargs):
            if not request.user.is_authenticated:
                login_url = login_url or reverse('login')
                next_url = request.get_full_path()
                redirect_url = f"{login_url}?{redirect_field_name}={next_url}"
                return HttpResponseRedirect(redirect_url)
            
            # 检查用户角色
            user_roles = set(request.user.groups.values_list('name', flat=True))
            required_roles = set(roles) if isinstance(roles, (list, tuple)) else {roles}
            
            if not (user_roles & required_roles):
                messages.error(request, '您没有权限访问此页面')
                return HttpResponseForbidden("Access denied")
            
            return view_func(request, *args, **kwargs)
        return _wrapped_view
    return decorator

def multi_permission_required(permissions, logical='AND', login_url=None, redirect_field_name='next'):
    """多权限要求装饰器"""
    def decorator(view_func):
        @wraps(view_func)
        def _wrapped_view(request, *args, **kwargs):
            if not request.user.is_authenticated:
                login_url = login_url or reverse('login')
                next_url = request.get_full_path()
                redirect_url = f"{login_url}?{redirect_field_name}={next_url}"
                return HttpResponseRedirect(redirect_url)
            
            # 检查权限
            has_permissions = []
            for perm in permissions:
                if '.' in perm:
                    app_label, codename = perm.split('.', 1)
                    has_permissions.append(request.user.has_perm(perm))
                else:
                    # 直接检查权限代码
                    has_permissions.append(request.user.has_perm(perm))
            
            # 根据逻辑操作符检查权限
            if logical.upper() == 'AND':
                has_access = all(has_permissions)
            elif logical.upper() == 'OR':
                has_access = any(has_permissions)
            else:
                has_access = all(has_permissions)  # 默认AND
            
            if not has_access:
                messages.error(request, '您没有足够的权限访问此页面')
                return HttpResponseForbidden("Access denied")
            
            return view_func(request, *args, **kwargs)
        return _wrapped_view
    return decorator

# 自定义混入类
from django.contrib.auth.mixins import AccessMixin

class RoleRequiredMixin(AccessMixin):
    """角色要求混入类"""
    role_required = None
    raise_exception = False
    permission_denied_message = '您没有权限访问此页面'
    
    def dispatch(self, request, *args, **kwargs):
        if not request.user.is_authenticated:
            return self.handle_no_permission()
        
        if not self.has_role():
            return self.handle_no_permission()
        
        return super().dispatch(request, *args, **kwargs)
    
    def has_role(self):
        """检查用户角色"""
        if self.role_required is None:
            return True
        
        user_roles = set(self.request.user.groups.values_list('name', flat=True))
        required_roles = set(self.role_required) if isinstance(self.role_required, (list, tuple)) else {self.role_required}
        
        return bool(user_roles & required_roles)
    
    def handle_no_permission(self):
        """处理无权限情况"""
        if self.raise_exception:
            from django.core.exceptions import PermissionDenied
            raise PermissionDenied(self.permission_denied_message)
        
        messages.error(self.request, self.permission_denied_message)
        return HttpResponseForbidden(self.permission_denied_message)

class MultiPermissionRequiredMixin(AccessMixin):
    """多权限要求混入类"""
    permissions_required = None
    permissions_logical = 'AND'  # 'AND' or 'OR'
    raise_exception = False
    permission_denied_message = '您没有足够的权限访问此页面'
    
    def dispatch(self, request, *args, **kwargs):
        if not request.user.is_authenticated:
            return self.handle_no_permission()
        
        if not self.has_permissions():
            return self.handle_no_permission()
        
        return super().dispatch(request, *args, **kwargs)
    
    def has_permissions(self):
        """检查用户权限"""
        if self.permissions_required is None:
            return True
        
        has_perms = []
        for perm in self.permissions_required:
            if '.' in perm:
                has_perms.append(self.request.user.has_perm(perm))
            else:
                has_perms.append(self.request.user.has_perm(perm))
        
        if self.permissions_logical.upper() == 'AND':
            return all(has_perms)
        elif self.permissions_logical.upper() == 'OR':
            return any(has_perms)
        else:
            return all(has_perms)
    
    def handle_no_permission(self):
        """处理无权限情况"""
        if self.raise_exception:
            from django.core.exceptions import PermissionDenied
            raise PermissionDenied(self.permission_denied_message)
        
        messages.error(self.request, self.permission_denied_message)
        return HttpResponseForbidden(self.permission_denied_message)

# 使用示例
@custom_login_required
@role_required(['admin', 'manager'])
@multi_permission_required(['app.view_report', 'app.edit_data'], logical='OR')
def admin_dashboard_view(request):
    """管理员仪表板视图"""
    return {'user': request.user, 'permissions': request.user.get_all_permissions()}

class AdminDashboardView(LoginRequiredMixin, RoleRequiredMixin, MultiPermissionRequiredMixin):
    """管理员仪表板类视图"""
    role_required = ['admin', 'manager']
    permissions_required = ['app.view_report', 'app.edit_data']
    permissions_logical = 'OR'
    
    def get(self, request):
        return {'user': request.user, 'permissions': request.user.get_all_permissions()}
}

## 自定义用户模型 \{#自定义用户模型}

### 扩展内置用户模型

```python
# 扩展内置用户模型
from django.contrib.auth.models import AbstractUser
from django.db import models
from django.utils import timezone

class CustomUser(AbstractUser):
    """自定义用户模型"""
    # 基础扩展字段
    phone_number = models.CharField(max_length=20, blank=True, null=True, unique=True)
    birth_date = models.DateField(blank=True, null=True)
    avatar = models.ImageField(upload_to='avatars/', blank=True, null=True)
    bio = models.TextField(max_length=500, blank=True)
    location = models.CharField(max_length=30, blank=True)
    website = models.URLField(blank=True)
    
    # 账户状态字段
    is_verified = models.BooleanField(default=False)
    is_premium = models.BooleanField(default=False)
    expiry_date = models.DateTimeField(blank=True, null=True)
    
    # 审计字段
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    
    # 自定义字段
    preferred_language = models.CharField(
        max_length=10, 
        choices=[
            ('en', 'English'),
            ('zh', '中文'),
            ('ja', '日本語'),
            ('ko', '한국어')
        ], 
        default='en'
    )
    timezone = models.CharField(max_length=50, default='UTC')
    
    # 索引优化
    class Meta:
        db_table = 'custom_user'
        indexes = [
            models.Index(fields=['phone_number']),
            models.Index(fields=['is_verified']),
            models.Index(fields=['is_premium']),
            models.Index(fields=['created_at']),
        ]
    
    def __str__(self):
        return f"{self.username} ({self.email})"
    
    def get_full_name(self):
        """获取完整姓名"""
        full_name = f"{self.first_name} {self.last_name}".strip()
        if not full_name:
            full_name = self.username
        return full_name
    
    def get_short_name(self):
        """获取简称"""
        return self.first_name or self.username
    
    def is_account_active(self):
        """检查账户是否活跃"""
        if not self.is_active:
            return False
        
        if self.expiry_date and self.expiry_date < timezone.now():
            return False
        
        return True
    
    def get_display_name(self):
        """获取显示名称"""
        if self.first_name and self.last_name:
            return f"{self.first_name} {self.last_name}"
        elif self.first_name:
            return self.first_name
        elif self.last_name:
            return self.last_name
        else:
            return self.username

# 用户配置文件模型
class UserProfile(models.Model):
    """用户配置文件"""
    user = models.OneToOneField(CustomUser, on_delete=models.CASCADE, related_name='profile')
    
    # 个人资料
    gender = models.CharField(
        max_length=10,
        choices=[
            ('male', '男'),
            ('female', '女'),
            ('other', '其他'),
            ('prefer_not_to_say', '不愿透露')
        ],
        blank=True
    )
    occupation = models.CharField(max_length=100, blank=True)
    company = models.CharField(max_length=100, blank=True)
    education = models.CharField(max_length=100, blank=True)
    
    # 社交信息
    wechat = models.CharField(max_length=50, blank=True)
    qq = models.CharField(max_length=20, blank=True)
    weibo = models.URLField(blank=True)
    
    # 通知设置
    email_notifications = models.BooleanField(default=True)
    sms_notifications = models.BooleanField(default=False)
    push_notifications = models.BooleanField(default=True)
    
    # 隐私设置
    profile_public = models.BooleanField(default=True)
    show_email = models.BooleanField(default=True)
    show_phone = models.BooleanField(default=False)
    
    # 统计信息
    login_count = models.PositiveIntegerField(default=0)
    post_count = models.PositiveIntegerField(default=0)
    follower_count = models.PositiveIntegerField(default=0)
    following_count = models.PositiveIntegerField(default=0)
    
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    
    class Meta:
        db_table = 'user_profile'
        verbose_name = '用户配置文件'
        verbose_name_plural = '用户配置文件'

# 信号处理器 - 自动创建用户配置文件
from django.db.models.signals import post_save
from django.dispatch import receiver

@receiver(post_save, sender=CustomUser)
def create_user_profile(sender, instance, created, **kwargs):
    """创建用户配置文件"""
    if created:
        UserProfile.objects.create(user=instance)

@receiver(post_save, sender=CustomUser)
def save_user_profile(sender, instance, **kwargs):
    """保存用户配置文件"""
    if hasattr(instance, 'profile'):
        instance.profile.save()

# 用户验证器
from django.core.exceptions import ValidationError
from django.contrib.auth.validators import UnicodeUsernameValidator
import re

class UserValidator:
    """用户验证器"""
    
    @staticmethod
    def validate_phone_number(phone_number):
        """验证手机号"""
        if phone_number:
            # 中国手机号验证
            pattern = r'^1[3-9]\d{9}$'
            if not re.match(pattern, phone_number):
                raise ValidationError('请输入有效的手机号码')
    
    @staticmethod
    def validate_username(username):
        """验证用户名"""
        if username:
            # 检查长度
            if len(username) < 3 or len(username) > 30:
                raise ValidationError('用户名长度应在3-30个字符之间')
            
            # 检查字符
            if not re.match(r'^[a-zA-Z0-9_]+$', username):
                raise ValidationError('用户名只能包含字母、数字和下划线')
    
    @staticmethod
    def validate_email(email):
        """验证邮箱"""
        if email:
            from django.core.validators import validate_email
            try:
                validate_email(email)
            except ValidationError:
                raise ValidationError('请输入有效的邮箱地址')

# 自定义用户管理器
from django.contrib.auth.models import BaseUserManager

class CustomUserManager(BaseUserManager):
    """自定义用户管理器"""
    
    def create_user(self, username, email=None, password=None, **extra_fields):
        """创建普通用户"""
        if not username:
            raise ValueError('用户名不能为空')
        
        email = self.normalize_email(email) if email else None
        user = self.model(username=username, email=email, **extra_fields)
        user.set_password(password)
        user.save(using=self._db)
        return user
    
    def create_superuser(self, username, email=None, password=None, **extra_fields):
        """创建超级用户"""
        extra_fields.setdefault('is_staff', True)
        extra_fields.setdefault('is_superuser', True)
        extra_fields.setdefault('is_verified', True)
        
        if extra_fields.get('is_staff') is not True:
            raise ValueError('超级用户必须设置is_staff=True')
        if extra_fields.get('is_superuser') is not True:
            raise ValueError('超级用户必须设置is_superuser=True')
        
        return self.create_user(username, email, password, **extra_fields)
    
    def get_by_phone(self, phone_number):
        """根据手机号获取用户"""
        return self.get(phone_number=phone_number)
    
    def get_active_users(self):
        """获取活跃用户"""
        return self.filter(
            is_active=True,
            expiry_date__isnull=True
        ).select_related('profile')
    
    def get_premium_users(self):
        """获取付费用户"""
        return self.filter(is_premium=True).select_related('profile')

# 更新模型使用自定义管理器
CustomUser.add_to_class('objects', CustomUserManager())

# 用户服务类
class UserService:
    """用户服务类"""
    
    @staticmethod
    def create_user_with_profile(username, email, password, **extra_fields):
        """创建用户并配置资料"""
        user = CustomUser.objects.create_user(
            username=username,
            email=email,
            password=password,
            **extra_fields
        )
        
        # 更新配置文件
        profile_data = extra_fields.get('profile_data', {})
        if profile_data:
            profile = user.profile
            for field, value in profile_data.items():
                if hasattr(profile, field):
                    setattr(profile, field, value)
            profile.save()
        
        return user
    
    @staticmethod
    def update_user_profile(user, **profile_data):
        """更新用户配置文件"""
        profile = user.profile
        for field, value in profile_data.items():
            if hasattr(profile, field):
                setattr(profile, field, value)
        profile.save()
        return profile
    
    @staticmethod
    def get_user_recommendations(user, limit=10):
        """获取用户推荐"""
        # 基于兴趣、地理位置等的推荐逻辑
        from django.db.models import Q
        
        recommendations = CustomUser.objects.filter(
            Q(profile__location=user.profile.location) &
            ~Q(id=user.id)  # 排除自己
        ).exclude(
            id__in=user.profile.following.all()  # 排除已关注的
        ).select_related('profile')[:limit]
        
        return recommendations

# 使用示例
def user_creation_example():
    """用户创建示例"""
    # 创建用户
    user = UserService.create_user_with_profile(
        username='john_doe',
        email='john@example.com',
        password='secure_password',
        first_name='John',
        last_name='Doe',
        phone_number='13812345678',
        profile_data={
            'gender': 'male',
            'occupation': 'Software Engineer',
            'company': 'Tech Corp'
        }
    )
    
    # 更新配置文件
    UserService.update_user_profile(
        user,
        email_notifications=False,
        show_email=False
    )
    
    return user

# 完全自定义用户模型（替代AbstractUser）
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin
from django.contrib.auth.base_user import BaseUserManager
from django.db import models

class EmailUserManager(BaseUserManager):
    """邮箱用户管理器"""
    
    def create_user(self, email, password=None, **extra_fields):
        """创建用户"""
        if not email:
            raise ValueError('邮箱地址不能为空')
        
        email = self.normalize_email(email)
        user = self.model(email=email, **extra_fields)
        user.set_password(password)
        user.save(using=self._db)
        return user
    
    def create_superuser(self, email, password=None, **extra_fields):
        """创建超级用户"""
        extra_fields.setdefault('is_staff', True)
        extra_fields.setdefault('is_superuser', True)
        extra_fields.setdefault('is_verified', True)
        
        if extra_fields.get('is_staff') is not True:
            raise ValueError('超级用户必须设置is_staff=True')
        if extra_fields.get('is_superuser') is not True:
            raise ValueError('超级用户必须设置is_superuser=True')
        
        return self.create_user(email, password, **extra_fields)

class EmailUser(AbstractBaseUser, PermissionsMixin):
    """邮箱用户模型"""
    email = models.EmailField(unique=True)
    username = models.CharField(max_length=150, blank=True)
    first_name = models.CharField(max_length=30, blank=True)
    last_name = models.CharField(max_length=150, blank=True)
    is_staff = models.BooleanField(default=False)
    is_active = models.BooleanField(default=True)
    is_verified = models.BooleanField(default=False)
    date_joined = models.DateTimeField(auto_now_add=True)
    
    # 自定义字段
    phone_number = models.CharField(max_length=20, blank=True)
    avatar = models.URLField(blank=True)
    bio = models.TextField(max_length=500, blank=True)
    
    USERNAME_FIELD = 'email'  # 使用邮箱作为登录字段
    REQUIRED_FIELDS = ['username']  # 创建超级用户时需要的字段
    
    objects = EmailUserManager()
    
    class Meta:
        db_table = 'email_user'
        verbose_name = '邮箱用户'
        verbose_name_plural = '邮箱用户'
    
    def __str__(self):
        return self.email
    
    def get_full_name(self):
        """获取完整姓名"""
        full_name = f"{self.first_name} {self.last_name}".strip()
        if not full_name:
            full_name = self.username or self.email
        return full_name
    
    def get_short_name(self):
        """获取简称"""
        return self.first_name or self.email.split('@')[0]

# 使用完全自定义用户模型的配置
"""
# settings.py 配置
AUTH_USER_MODEL = 'your_app.EmailUser'

# 注意：使用自定义用户模型需要在项目初期就设置，
# 一旦有数据后很难更改
"""

# 多因素认证后端
import pyotp
import qrcode
from io import BytesIO
import base64
from django.contrib.auth.backends import BaseBackend
from django.contrib.auth import get_user_model

User = get_user_model()

class TwoFactorAuthBackend(BaseBackend):
    """两因素认证后端"""
    
    def authenticate(self, request, username=None, password=None, otp_code=None, **kwargs):
        """两因素认证"""
        if username is None or password is None:
            return None
        
        # 首先验证用户名密码
        user = self.authenticate_basic(username, password)
        if not user:
            return None
        
        # 检查是否启用两因素认证
        if hasattr(user, 'profile') and user.profile.enable_2fa:
            if not otp_code:
                # 需要OTP码但未提供
                raise ValueError('需要提供两因素认证码')
            
            # 验证OTP码
            if not self.verify_otp(user, otp_code):
                return None
        
        return user
    
    def authenticate_basic(self, username, password):
        """基础认证"""
        try:
            user = User.objects.get(username=username)
            if user.check_password(password):
                return user
        except User.DoesNotExist:
            pass
        return None
    
    def verify_otp(self, user, otp_code):
        """验证OTP码"""
        if not hasattr(user, 'profile') or not user.profile.totp_secret:
            return False
        
        totp = pyotp.TOTP(user.profile.totp_secret)
        return totp.verify(otp_code, valid_window=1)  # 允许前后1个时间窗口
    
    def get_user(self, user_id):
        """获取用户"""
        try:
            return User.objects.get(pk=user_id)
        except User.DoesNotExist:
            return None

# TOTP服务类
class TOTPService:
    """TOTP服务"""
    
    @staticmethod
    def generate_secret():
        """生成TOTP密钥"""
        return pyotp.random_base32()
    
    @staticmethod
    def generate_qr_code(user, issuer_name="YourApp"):
        """生成QR码"""
        secret = user.profile.totp_secret
        totp_uri = pyotp.totp.TOTP(secret).provisioning_uri(
            name=user.email,
            issuer_name=issuer_name
        )
        
        qr = qrcode.QRCode(version=1, box_size=10, border=5)
        qr.add_data(totp_uri)
        qr.make(fit=True)
        
        img = qr.make_image(fill_color="black", back_color="white")
        buffer = BytesIO()
        img.save(buffer, format='PNG')
        img_str = base64.b64encode(buffer.getvalue()).decode()
        
        return img_str, totp_uri
    
    @staticmethod
    def verify_code(secret, code):
        """验证TOTP码"""
        totp = pyotp.TOTP(secret)
        return totp.verify(code, valid_window=1)

# 更新用户配置文件模型以支持2FA
UserProfile.add_to_class(
    'enable_2fa',
    models.BooleanField(default=False)
)
UserProfile.add_to_class(
    'totp_secret',
    models.CharField(max_length=32, blank=True)
)
UserProfile.add_to_class(
    'backup_codes',
    models.TextField(blank=True)  # JSON格式存储备用码
)

# 社交认证后端
import requests
from social_core.backends.oauth import BaseOAuth2

class WeChatOAuthBackend(BaseBackend):
    """微信OAuth认证后端"""
    
    def authenticate(self, request, code=None, **kwargs):
        """微信OAuth认证"""
        if not code:
            return None
        
        # 获取访问令牌
        token_url = 'https://api.weixin.qq.com/sns/oauth2/access_token'
        params = {
            'appid': kwargs.get('appid'),
            'secret': kwargs.get('secret'),
            'code': code,
            'grant_type': 'authorization_code'
        }
        
        response = requests.get(token_url, params=params)
        token_data = response.json()
        
        if 'errcode' in token_data:
            return None
        
        # 获取用户信息
        user_info_url = 'https://api.weixin.qq.com/sns/userinfo'
        user_params = {
            'access_token': token_data['access_token'],
            'openid': token_data['openid'],
            'lang': 'zh_CN'
        }
        
        user_response = requests.get(user_info_url, params=user_params)
        user_data = user_response.json()
        
        if 'errcode' in user_data:
            return None
        
        # 创建或获取用户
        user, created = User.objects.get_or_create(
            username=f"wechat_{user_data['openid']}",
            defaults={
                'first_name': user_data.get('nickname', ''),
                'avatar': user_data.get('headimgurl', ''),
                'is_verified': True
            }
        )
        
        return user
    
    def get_user(self, user_id):
        """获取用户"""
        try:
            return User.objects.get(pk=user_id)
        except User.DoesNotExist:
            return None

# LDAP认证后端（需要python-ldap）
"""
class LDAPAuthBackend(BaseBackend):
    # LDAP认证后端实现
    pass
"""

# 自定义认证后端配置
"""
# settings.py 配置
AUTHENTICATION_BACKENDS = [
    'django.contrib.auth.backends.ModelBackend',  # 默认后端
    'your_app.backends.TwoFactorAuthBackend',     # 两因素认证
    'your_app.backends.WeChatOAuthBackend',       # 微信OAuth
    # 'your_app.backends.LDAPAuthBackend',       # LDAP认证
]
"""

# 密码策略和安全
from django.contrib.auth.hashers import make_password, check_password
from django.core.exceptions import ValidationError
import re
from django.conf import settings

class PasswordPolicy:
    """密码策略"""
    
    def __init__(self):
        self.min_length = getattr(settings, 'PASSWORD_MIN_LENGTH', 8)
        self.require_uppercase = getattr(settings, 'PASSWORD_REQUIRE_UPPERCASE', True)
        self.require_lowercase = getattr(settings, 'PASSWORD_REQUIRE_LOWERCASE', True)
        self.require_numbers = getattr(settings, 'PASSWORD_REQUIRE_NUMBERS', True)
        self.require_special = getattr(settings, 'PASSWORD_REQUIRE_SPECIAL', False)
        self.special_chars = getattr(settings, 'PASSWORD_SPECIAL_CHARS', '!@#$%^&*()_+-=[]{}|;:,.<>?')
    
    def validate_password(self, password, user=None):
        """验证密码强度"""
        errors = []
        
        # 长度检查
        if len(password) < self.min_length:
            errors.append(f'密码长度至少需要{self.min_length}个字符')
        
        # 大写字母检查
        if self.require_uppercase and not re.search(r'[A-Z]', password):
            errors.append('密码必须包含大写字母')
        
        # 小写字母检查
        if self.require_lowercase and not re.search(r'[a-z]', password):
            errors.append('密码必须包含小写字母')
        
        # 数字检查
        if self.require_numbers and not re.search(r'\d', password):
            errors.append('密码必须包含数字')
        
        # 特殊字符检查
        if self.require_special and not re.search(f'[{re.escape(self.special_chars)}]', password):
            errors.append(f'密码必须包含特殊字符: {self.special_chars}')
        
        # 检查是否与用户信息相似
        if user:
            similar_checks = [
                user.username.lower() in password.lower(),
                user.first_name.lower() in password.lower(),
                user.last_name.lower() in password.lower(),
                user.email.lower() in password.lower() if user.email else False
            ]
            
            if any(similar_checks):
                errors.append('密码不能与您的个人信息相似')
        
        if errors:
            raise ValidationError(errors)
        
        return password

# 密码历史记录
class PasswordHistory(models.Model):
    """密码历史记录"""
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    password_hash = models.CharField(max_length=128)  # 存储哈希后的密码
    created_at = models.DateTimeField(auto_now_add=True)
    
    class Meta:
        db_table = 'password_history'
        ordering = ['-created_at']
    
    @classmethod
    def add_password(cls, user, raw_password):
        """添加密码到历史记录"""
        password_hash = make_password(raw_password)
        cls.objects.create(user=user, password_hash=password_hash)
        
        # 限制历史记录数量
        cls.trim_history(user)
    
    @classmethod
    def trim_history(cls, user, max_history=5):
        """限制密码历史记录数量"""
        history_count = cls.objects.filter(user=user).count()
        if history_count > max_history:
            oldest_records = cls.objects.filter(user=user).order_by('created_at')[max_history:]
            oldest_records.delete()
    
    @classmethod
    def is_password_used_before(cls, user, raw_password, check_count=5):
        """检查密码是否曾经使用过"""
        recent_history = cls.objects.filter(user=user).order_by('-created_at')[:check_count]
        for record in recent_history:
            if check_password(raw_password, record.password_hash):
                return True
        return False

# 密码重置令牌
from django.utils.crypto import get_random_string
from django.utils import timezone

class PasswordResetToken(models.Model):
    """密码重置令牌"""
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    token = models.CharField(max_length=32, unique=True)
    created_at = models.DateTimeField(auto_now_add=True)
    is_used = models.BooleanField(default=False)
    
    class Meta:
        db_table = 'password_reset_token'
    
    @classmethod
    def generate_token(cls, user):
        """生成重置令牌"""
        # 先删除旧令牌
        cls.objects.filter(user=user, is_used=False).delete()
        
        token = get_random_string(32)
        reset_token = cls.objects.create(user=user, token=token)
        return reset_token
    
    def is_valid(self):
        """检查令牌是否有效"""
        from django.conf import settings
        validity_period = getattr(settings, 'PASSWORD_RESET_TOKEN_VALIDITY', 3600)  # 1小时
        
        return (
            not self.is_used and
            (timezone.now() - self.created_at).seconds < validity_period
        )

# 密码服务
class PasswordService:
    """密码服务"""
    
    def __init__(self):
        self.policy = PasswordPolicy()
    
    def set_password(self, user, raw_password, check_history=True):
        """设置用户密码"""
        # 验证密码策略
        self.policy.validate_password(raw_password, user)
        
        # 检查密码历史
        if check_history and PasswordHistory.is_password_used_before(user, raw_password):
            raise ValidationError('新密码不能与最近使用的密码相同')
        
        # 设置密码
        user.set_password(raw_password)
        user.save()
        
        # 添加到历史记录
        PasswordHistory.add_password(user, raw_password)
    
    def change_password(self, user, old_password, new_password):
        """更改密码"""
        # 验证旧密码
        if not user.check_password(old_password):
            raise ValidationError('旧密码不正确')
        
        # 设置新密码
        self.set_password(user, new_password)
    
    def reset_password_with_token(self, token_str, new_password):
        """使用令牌重置密码"""
        try:
            reset_token = PasswordResetToken.objects.get(token=token_str)
        except PasswordResetToken.DoesNotExist:
            raise ValidationError('无效的重置令牌')
        
        if not reset_token.is_valid():
            raise ValidationError('重置令牌已过期或已被使用')
        
        # 设置新密码
        self.set_password(reset_token.user, new_password)
        
        # 标记令牌为已使用
        reset_token.is_used = True
        reset_token.save()
    
    def generate_password_reset_link(self, user):
        """生成密码重置链接"""
        token = PasswordResetToken.generate_token(user)
        
        # 这里应该生成实际的重置链接
        # 通常会发送邮件或短信
        reset_link = f"/reset-password/{token.token}/"
        return reset_link

# 密码强度指示器
class PasswordStrengthIndicator:
    """密码强度指示器"""
    
    @staticmethod
    def calculate_strength(password):
        """计算密码强度"""
        score = 0
        feedback = []
        
        # 长度分数
        if len(password) >= 8:
            score += 1
        else:
            feedback.append("密码至少需要8个字符")
        
        if len(password) >= 12:
            score += 1
        
        # 字符多样性
        if re.search(r'[a-z]', password):
            score += 1
        else:
            feedback.append("添加小写字母")
        
        if re.search(r'[A-Z]', password):
            score += 1
        else:
            feedback.append("添加大写字母")
        
        if re.search(r'\d', password):
            score += 1
        else:
            feedback.append("添加数字")
        
        if re.search(r'[!@#$%^&*(),.?":{}|<>]', password):
            score += 1
        else:
            feedback.append("添加特殊字符")
        
        # 重复字符检查
        if len(set(password)) / len(password) < 0.7:  # 70%唯一字符
            score -= 1
            feedback.append("避免重复字符")
        
        # 强度评级
        if score <= 2:
            strength = 'weak'
            level = '弱'
        elif score <= 4:
            strength = 'medium'
            level = '中'
        else:
            strength = 'strong'
            level = '强'
        
        return {
            'score': score,
            'strength': strength,
            'level': level,
            'feedback': feedback,
            'percentage': min(100, int((score / 6) * 100))
        }

# 使用示例
def password_management_example():
    """密码管理示例"""
    service = PasswordService()
    indicator = PasswordStrengthIndicator()
    
    # 检查密码强度
    password_strength = indicator.calculate_strength("MyPass123!")
    print(f"密码强度: {password_strength}")
    
    # 创建用户并设置密码
    user = CustomUser.objects.create_user(
        username='testuser',
        email='test@example.com'
    )
    
    try:
        # 设置符合策略的密码
        service.set_password(user, "SecurePass123!")
        print("密码设置成功")
    except ValidationError as e:
        print(f"密码验证失败: {e}")
    
    # 生成密码重置链接
    reset_link = service.generate_password_reset_link(user)
    print(f"重置链接: {reset_link}")
    
    return user

# 用户CRUD操作
from django.contrib.auth import get_user_model
from django.db import transaction
from django.core.exceptions import ValidationError

User = get_user_model()

class UserManager:
    """用户管理器"""
    
    @staticmethod
    def create_user_with_validation(username, email, password, **extra_fields):
        """创建用户（带验证）"""
        # 验证输入
        validator = UserValidator()
        
        if username:
            validator.validate_username(username)
        if email:
            validator.validate_email(email)
        
        # 检查唯一性
        if User.objects.filter(username=username).exists():
            raise ValidationError('用户名已存在')
        
        if email and User.objects.filter(email=email).exists():
            raise ValidationError('邮箱已被注册')
        
        # 创建用户
        with transaction.atomic():
            user = User.objects.create_user(
                username=username,
                email=email,
                password=password,
                **extra_fields
            )
        
        return user
    
    @staticmethod
    def update_user(user_id, **updates):
        """更新用户信息"""
        try:
            user = User.objects.get(id=user_id)
        except User.DoesNotExist:
            raise ValidationError('用户不存在')
        
        # 验证更新字段
        validator = UserValidator()
        
        if 'username' in updates:
            validator.validate_username(updates['username'])
            # 检查用户名唯一性（排除当前用户）
            if User.objects.filter(username=updates['username']).exclude(id=user_id).exists():
                raise ValidationError('用户名已被使用')
        
        if 'email' in updates:
            validator.validate_email(updates['email'])
            # 检查邮箱唯一性
            if User.objects.filter(email=updates['email']).exclude(id=user_id).exists():
                raise ValidationError('邮箱已被使用')
        
        # 更新字段
        for field, value in updates.items():
            if hasattr(user, field):
                setattr(user, field, value)
        
        user.save()
        return user
    
    @staticmethod
    def deactivate_user(user_id, reason=''):
        """停用用户"""
        try:
            user = User.objects.get(id=user_id)
            user.is_active = False
            user.save()
            
            # 记录停用原因
            UserDeactivationLog.objects.create(
                user=user,
                reason=reason,
                deactivated_by=None  # 可以记录操作人
            )
            
            return True
        except User.DoesNotExist:
            return False
    
    @staticmethod
    def activate_user(user_id):
        """激活用户"""
        try:
            user = User.objects.get(id=user_id)
            user.is_active = True
            user.save()
            return True
        except User.DoesNotExist:
            return False
    
    @staticmethod
    def delete_user(user_id, hard_delete=False):
        """删除用户"""
        try:
            user = User.objects.get(id=user_id)
            
            if hard_delete:
                # 硬删除
                user.delete()
            else:
                # 软删除 - 重命名并停用
                user.username = f"deleted_{user.id}_{user.username}"
                user.is_active = False
                user.save()
            
            return True
        except User.DoesNotExist:
            return False
    
    @staticmethod
    def bulk_create_users(user_data_list):
        """批量创建用户"""
        users = []
        errors = []
        
        for i, user_data in enumerate(user_data_list):
            try:
                user = UserManager.create_user_with_validation(**user_data)
                users.append(user)
            except ValidationError as e:
                errors.append(f"用户 {i+1}: {str(e)}")
        
        return users, errors

# 用户停用日志
class UserDeactivationLog(models.Model):
    """用户停用日志"""
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    reason = models.TextField()
    deactivated_by = models.ForeignKey(
        User, 
        on_delete=models.SET_NULL, 
        null=True, 
        related_name='deactivated_users'
    )
    deactivated_at = models.DateTimeField(auto_now_add=True)
    
    class Meta:
        db_table = 'user_deactivation_log'

# 用户搜索和筛选
class UserSearchFilter:
    """用户搜索和筛选"""
    
    @staticmethod
    def search_users(query, filters=None):
        """搜索用户"""
        from django.db.models import Q
        
        queryset = User.objects.all()
        
        # 文本搜索
        if query:
            queryset = queryset.filter(
                Q(username__icontains=query) |
                Q(first_name__icontains=query) |
                Q(last_name__icontains=query) |
                Q(email__icontains=query)
            )
        
        # 应用过滤器
        if filters:
            if 'is_active' in filters:
                queryset = queryset.filter(is_active=filters['is_active'])
            
            if 'is_verified' in filters:
                if hasattr(User, 'is_verified'):
                    queryset = queryset.filter(is_verified=filters['is_verified'])
            
            if 'date_joined_after' in filters:
                queryset = queryset.filter(date_joined__gte=filters['date_joined_after'])
            
            if 'date_joined_before' in filters:
                queryset = queryset.filter(date_joined__lte=filters['date_joined_before'])
            
            if 'groups' in filters:
                queryset = queryset.filter(groups__name__in=filters['groups'])
        
        return queryset.distinct()
    
    @staticmethod
    def get_user_statistics():
        """获取用户统计信息"""
        total_users = User.objects.count()
        active_users = User.objects.filter(is_active=True).count()
        verified_users = User.objects.filter(is_verified=True).count() if hasattr(User, 'is_verified') else 0
        premium_users = User.objects.filter(is_premium=True).count() if hasattr(User, 'is_premium') else 0
        
        # 按日期统计
        from django.db.models import Count
        from django.utils import timezone
        from datetime import timedelta
        
        today = timezone.now().date()
        last_week = today - timedelta(days=7)
        
        daily_signups = User.objects.filter(
            date_joined__date__range=[last_week, today]
        ).values('date_joined__date').annotate(count=Count('id')).order_by('date_joined__date')
        
        return {
            'total': total_users,
            'active': active_users,
            'verified': verified_users,
            'premium': premium_users,
            'active_percentage': (active_users / total_users * 100) if total_users > 0 else 0,
            'daily_signups': list(daily_signups)
        }

# 用户导入导出
import csv
import json
from django.http import HttpResponse

class UserImportExport:
    """用户导入导出"""
    
    @staticmethod
    def export_users_csv(users, fields=None):
        """导出用户为CSV"""
        if fields is None:
            fields = ['username', 'email', 'first_name', 'last_name', 'is_active', 'date_joined']
        
        response = HttpResponse(content_type='text/csv')
        response['Content-Disposition'] = 'attachment; filename="users.csv"'
        
        writer = csv.writer(response)
        
        # 写入表头
        writer.writerow(fields)
        
        # 写入数据
        for user in users:
            row = []
            for field in fields:
                value = getattr(user, field, '')
                if isinstance(value, bool):
                    value = 'Yes' if value else 'No'
                elif hasattr(value, 'strftime'):  # DateTime
                    value = value.strftime('%Y-%m-%d %H:%M:%S')
                row.append(str(value))
            writer.writerow(row)
        
        return response
    
    @staticmethod
    def import_users_from_csv(csv_file, required_fields=None):
        """从CSV导入用户"""
        if required_fields is None:
            required_fields = ['username', 'email']
        
        users_created = []
        errors = []
        
        decoded_file = csv_file.read().decode('utf-8').splitlines()
        reader = csv.DictReader(decoded_file)
        
        for row_num, row in enumerate(reader, start=2):  # 从第2行开始（跳过表头）
            try:
                # 验证必需字段
                for field in required_fields:
                    if not row.get(field):
                        raise ValidationError(f"第{row_num}行缺少必需字段: {field}")
                
                # 创建用户
                user_data = {
                    'username': row.get('username'),
                    'email': row.get('email', ''),
                    'first_name': row.get('first_name', ''),
                    'last_name': row.get('last_name', ''),
                    'is_active': row.get('is_active', 'Yes').lower() in ['yes', 'true', '1'],
                }
                
                # 设置默认密码（实际应用中应该发送密码重置邮件）
                user_data['password'] = 'TempPassword123!'
                
                user = UserManager.create_user_with_validation(**user_data)
                users_created.append(user)
                
            except ValidationError as e:
                errors.append(f"第{row_num}行错误: {str(e)}")
            except Exception as e:
                errors.append(f"第{row_num}行未知错误: {str(e)}")
        
        return users_created, errors

# 使用示例
def user_management_example():
    """用户管理示例"""
    # 创建用户
    user = UserManager.create_user_with_validation(
        username='newuser',
        email='newuser@example.com',
        password='SecurePass123!',
        first_name='New',
        last_name='User'
    )
    
    # 搜索用户
    search_results = UserSearchFilter.search_users('new', {
        'is_active': True
    })
    
    # 获取统计信息
    stats = UserSearchFilter.get_user_statistics()
    print(f"用户统计: {stats}")
    
    return user