- Invitation 模型:team/code(TEAM-XXXX-XXXX唯一)/role/email/月额度/created_by/expires_at(7天)/ used_by/used_at/status(pending/used/revoked/expired);迁移 accounts/0003(未 apply 远程库,rule 3) - RegisterSerializer 加 invite_code:有效码→幂等 get_or_create 成员+标used+跳过额度发放; 无码→开新团队当 owner+试用额度(原逻辑);邀请 role 不许 OWNER 降级 ADMIN;不存在/已用/过期→400 - 端点 /api/auth/team/invitations/(GET列/POST生成)+ <id>/revoke/;权限复用 can_manage_team(超管/团管) 验证: django check 0;apps.accounts 测试 5/5 OK —— 含「凭码加入同团队/角色额度对/不重复发额度/ 未知码400/普通成员生成403」。认证定调:username 为主标识、不用邮箱注册。 ⚠️ Invitation 迁移需用户低峰 apply 到远程 MySQL,前端生成码 e2e 才能跑(逻辑已被测试证)。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
148 lines
5.5 KiB
Python
148 lines
5.5 KiB
Python
import uuid
|
|
|
|
from django.contrib.auth.models import AbstractUser
|
|
from django.db import models
|
|
|
|
from apps.common.models import TimeStampedModel
|
|
|
|
|
|
class User(AbstractUser):
|
|
class Status(models.TextChoices):
|
|
ACTIVE = "active", "Active"
|
|
DISABLED = "disabled", "Disabled"
|
|
|
|
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
status = models.CharField(max_length=24, choices=Status.choices, default=Status.ACTIVE)
|
|
phone = models.CharField(max_length=32, blank=True)
|
|
avatar_url = models.URLField(blank=True)
|
|
|
|
@property
|
|
def is_disabled(self) -> bool:
|
|
return self.status == self.Status.DISABLED
|
|
|
|
|
|
class Team(TimeStampedModel):
|
|
class Status(models.TextChoices):
|
|
ACTIVE = "active", "Active"
|
|
DISABLED = "disabled", "Disabled"
|
|
|
|
name = models.CharField(max_length=128)
|
|
status = models.CharField(max_length=24, choices=Status.choices, default=Status.ACTIVE)
|
|
owner = models.ForeignKey(User, on_delete=models.PROTECT, related_name="owned_teams")
|
|
|
|
def __str__(self) -> str:
|
|
return self.name
|
|
|
|
|
|
def _default_notify() -> dict:
|
|
return {"n-export": True, "n-fail": True, "n-quota": True, "n-login": True}
|
|
|
|
|
|
def _default_creation() -> dict:
|
|
return {"template": "pain", "duration": "60", "subtitle": "big-variety", "bgm": "kapian", "transition": "fade"}
|
|
|
|
|
|
def _default_display() -> dict:
|
|
return {"appearance": "system", "language": "zh", "density": "standard"}
|
|
|
|
|
|
class UserPreference(TimeStampedModel):
|
|
"""用户设置:通知策略 / 两步验证 / 创作默认 / 显示偏好。服务端持久化(替代前端 localStorage)。"""
|
|
|
|
user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="preference")
|
|
notify = models.JSONField(default=_default_notify, blank=True)
|
|
two_factor_enabled = models.BooleanField(default=False)
|
|
creation_defaults = models.JSONField(default=_default_creation, blank=True)
|
|
display = models.JSONField(default=_default_display, blank=True)
|
|
|
|
def __str__(self) -> str:
|
|
return f"prefs/{self.user}"
|
|
|
|
|
|
class LoginSession(TimeStampedModel):
|
|
"""登录会话记录:每次登录写一条(设备 UA / IP / 时间),供设置页「在用设备」展示与下线。"""
|
|
|
|
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="login_sessions")
|
|
user_agent = models.CharField(max_length=400, blank=True)
|
|
ip_address = models.GenericIPAddressField(null=True, blank=True)
|
|
last_seen_at = models.DateTimeField(auto_now=True)
|
|
revoked_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ["-last_seen_at"]
|
|
|
|
|
|
class TeamMember(TimeStampedModel):
|
|
class Role(models.TextChoices):
|
|
OWNER = "owner", "Owner"
|
|
ADMIN = "admin", "Admin"
|
|
MEMBER = "member", "Member"
|
|
VIEWER = "viewer", "Viewer"
|
|
|
|
class Status(models.TextChoices):
|
|
ACTIVE = "active", "Active"
|
|
INVITED = "invited", "Invited"
|
|
DISABLED = "disabled", "Disabled"
|
|
|
|
team = models.ForeignKey(Team, on_delete=models.CASCADE, related_name="members")
|
|
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="team_memberships")
|
|
role = models.CharField(max_length=24, choices=Role.choices, default=Role.MEMBER)
|
|
status = models.CharField(max_length=24, choices=Status.choices, default=Status.ACTIVE)
|
|
monthly_credit_limit = models.DecimalField(max_digits=12, decimal_places=2, default=0)
|
|
|
|
class Meta:
|
|
unique_together = [("team", "user")]
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.team} / {self.user} / {self.role}"
|
|
|
|
|
|
def _gen_invite_code() -> str:
|
|
import secrets
|
|
import string
|
|
|
|
alphabet = "".join(c for c in (string.ascii_uppercase + string.digits) if c not in "O0I1")
|
|
block = lambda: "".join(secrets.choice(alphabet) for _ in range(4)) # noqa: E731
|
|
return f"TEAM-{block()}-{block()}"
|
|
|
|
|
|
def _default_invite_expiry():
|
|
from datetime import timedelta
|
|
|
|
from django.utils import timezone
|
|
|
|
return timezone.now() + timedelta(days=7)
|
|
|
|
|
|
class Invitation(TimeStampedModel):
|
|
"""团队邀请码:超管/团管生成,新用户凭码注册即加入该团队(幂等加成员、不重复发额度)。"""
|
|
|
|
class Status(models.TextChoices):
|
|
PENDING = "pending", "Pending"
|
|
USED = "used", "Used"
|
|
REVOKED = "revoked", "Revoked"
|
|
EXPIRED = "expired", "Expired"
|
|
|
|
team = models.ForeignKey(Team, on_delete=models.CASCADE, related_name="invitations")
|
|
code = models.CharField(max_length=32, unique=True, default=_gen_invite_code)
|
|
role = models.CharField(max_length=24, choices=TeamMember.Role.choices, default=TeamMember.Role.MEMBER)
|
|
email = models.EmailField(blank=True)
|
|
monthly_credit_limit = models.DecimalField(max_digits=12, decimal_places=2, default=0)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name="created_invitations")
|
|
expires_at = models.DateTimeField(default=_default_invite_expiry)
|
|
used_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name="used_invitations")
|
|
used_at = models.DateTimeField(null=True, blank=True)
|
|
status = models.CharField(max_length=16, choices=Status.choices, default=Status.PENDING)
|
|
|
|
class Meta:
|
|
indexes = [models.Index(fields=["team", "status"])]
|
|
|
|
@property
|
|
def is_valid(self) -> bool:
|
|
from django.utils import timezone
|
|
|
|
return self.status == self.Status.PENDING and self.expires_at > timezone.now()
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.team} / {self.code} / {self.status}"
|