155 lines
5.2 KiB
Python
155 lines
5.2 KiB
Python
"""
|
||
User models for authentication and profile management.
|
||
"""
|
||
import secrets
|
||
import string
|
||
from django.db import models
|
||
from django.contrib.auth.models import AbstractBaseUser, BaseUserManager, PermissionsMixin
|
||
|
||
|
||
class UserManager(BaseUserManager):
|
||
"""Custom user manager."""
|
||
|
||
def create_user(self, open_id, password=None, **extra_fields):
|
||
"""Create and save a regular user."""
|
||
if not open_id:
|
||
raise ValueError('The open_id must be set')
|
||
user = self.model(open_id=open_id, **extra_fields)
|
||
if password:
|
||
user.set_password(password)
|
||
else:
|
||
user.set_unusable_password()
|
||
user.save(using=self._db)
|
||
return user
|
||
|
||
def create_superuser(self, open_id, password=None, **extra_fields):
|
||
"""Create and save a superuser."""
|
||
extra_fields.setdefault('is_staff', True)
|
||
extra_fields.setdefault('is_superuser', True)
|
||
extra_fields.setdefault('role', 'admin')
|
||
return self.create_user(open_id, password=password, **extra_fields)
|
||
|
||
|
||
class User(AbstractBaseUser, PermissionsMixin):
|
||
"""Custom user model matching the original schema."""
|
||
|
||
class Role(models.TextChoices):
|
||
USER = 'user', '普通用户'
|
||
ADMIN = 'admin', '管理员'
|
||
|
||
id = models.AutoField(primary_key=True)
|
||
open_id = models.CharField('OpenID', max_length=64, unique=True)
|
||
user_id = models.CharField(
|
||
'用户ID',
|
||
max_length=16,
|
||
unique=True,
|
||
blank=True,
|
||
null=True,
|
||
help_text='可分享的用户ID，用于好友搜索、商品搜索等',
|
||
)
|
||
name = models.CharField('用户名', max_length=255, blank=True, null=True)
|
||
email = models.EmailField('邮箱', max_length=320, blank=True, null=True)
|
||
avatar = models.TextField('头像', blank=True, null=True)
|
||
login_method = models.CharField('登录方式', max_length=64, blank=True, null=True)
|
||
role = models.CharField(
|
||
'角色',
|
||
max_length=10,
|
||
choices=Role.choices,
|
||
default=Role.USER
|
||
)
|
||
stripe_customer_id = models.CharField('Stripe客户ID', max_length=255, blank=True, null=True)
|
||
stripe_account_id = models.CharField('Stripe账户ID', max_length=255, blank=True, null=True)
|
||
|
||
# Django auth fields
|
||
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)
|
||
last_signed_in = models.DateTimeField('最后登录时间', auto_now=True)
|
||
|
||
objects = UserManager()
|
||
|
||
USERNAME_FIELD = 'open_id'
|
||
REQUIRED_FIELDS = []
|
||
|
||
class Meta:
|
||
db_table = 'users'
|
||
verbose_name = '用户'
|
||
verbose_name_plural = '用户'
|
||
indexes = [
|
||
models.Index(fields=["role", "is_active"]),
|
||
models.Index(fields=["created_at"]),
|
||
models.Index(fields=["user_id"]),
|
||
]
|
||
|
||
def __str__(self):
|
||
return self.name or self.open_id
|
||
|
||
def _generate_user_id(self):
|
||
"""Generate a unique shareable user_id (e.g. U8x3k2m1)."""
|
||
alphabet = string.ascii_lowercase + string.digits
|
||
for _ in range(10):
|
||
candidate = "U" + "".join(secrets.choice(alphabet) for _ in range(8))
|
||
if not User.objects.filter(user_id=candidate).exists():
|
||
return candidate
|
||
return "U" + secrets.token_hex(4)
|
||
|
||
def save(self, *args, **kwargs):
|
||
if not self.user_id:
|
||
self.user_id = self._generate_user_id()
|
||
super().save(*args, **kwargs)
|
||
|
||
@property
|
||
def is_admin(self):
|
||
return self.role == self.Role.ADMIN
|
||
|
||
|
||
class FriendRequest(models.Model):
|
||
"""Friend request and relationship status."""
|
||
|
||
class Status(models.TextChoices):
|
||
PENDING = "pending", "待处理"
|
||
ACCEPTED = "accepted", "已通过"
|
||
REJECTED = "rejected", "已拒绝"
|
||
CANCELED = "canceled", "已取消"
|
||
|
||
requester = models.ForeignKey(
|
||
User,
|
||
related_name="sent_friend_requests",
|
||
on_delete=models.CASCADE,
|
||
)
|
||
receiver = models.ForeignKey(
|
||
User,
|
||
related_name="received_friend_requests",
|
||
on_delete=models.CASCADE,
|
||
)
|
||
status = models.CharField(
|
||
"状态",
|
||
max_length=10,
|
||
choices=Status.choices,
|
||
default=Status.PENDING,
|
||
)
|
||
accepted_at = models.DateTimeField("通过时间", blank=True, null=True)
|
||
created_at = models.DateTimeField("创建时间", auto_now_add=True)
|
||
updated_at = models.DateTimeField("更新时间", auto_now=True)
|
||
|
||
class Meta:
|
||
db_table = "friend_requests"
|
||
verbose_name = "好友请求"
|
||
verbose_name_plural = "好友请求"
|
||
unique_together = ("requester", "receiver")
|
||
constraints = [
|
||
models.CheckConstraint(
|
||
check=~models.Q(requester=models.F("receiver")),
|
||
name="no_self_friend_request",
|
||
)
|
||
]
|
||
indexes = [
|
||
models.Index(fields=["receiver", "status"]),
|
||
models.Index(fields=["requester", "status"]),
|
||
]
|
||
|
||
def __str__(self):
|
||
return f"{self.requester_id}->{self.receiver_id} ({self.status})"
|