410 lines
9.8 KiB
Python
410 lines
9.8 KiB
Python
"""
|
||
兑换码模型
|
||
|
||
定义余额兑换码相关的数据表结构。
|
||
|
||
设计说明：
|
||
- 兑换码支持批量生成、导入导出
|
||
- 记录完整的使用日志
|
||
- 支持设置有效期和使用限制
|
||
"""
|
||
|
||
from datetime import datetime, timezone
|
||
from enum import Enum
|
||
from typing import TYPE_CHECKING
|
||
from uuid import uuid4
|
||
import secrets
|
||
import string
|
||
|
||
from sqlalchemy import (
|
||
BigInteger,
|
||
Boolean,
|
||
DateTime,
|
||
Enum as SQLEnum,
|
||
ForeignKey,
|
||
Index,
|
||
Integer,
|
||
String,
|
||
Text,
|
||
func,
|
||
)
|
||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||
|
||
from app.database import Base
|
||
|
||
if TYPE_CHECKING:
|
||
from app.models.user import User
|
||
|
||
|
||
def generate_uuid() -> str:
|
||
"""生成 UUID 字符串"""
|
||
return str(uuid4())
|
||
|
||
|
||
def utc_now() -> datetime:
|
||
"""获取当前 UTC 时间"""
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def generate_redeem_code(length: int = 16) -> str:
|
||
"""
|
||
生成兑换码
|
||
|
||
格式：大写字母和数字，分段显示（如 XXXX-XXXX-XXXX-XXXX）
|
||
排除容易混淆的字符：0/O, 1/I/L
|
||
"""
|
||
# 可用字符集（排除易混淆字符）
|
||
alphabet = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
|
||
code = "".join(secrets.choice(alphabet) for _ in range(length))
|
||
# 每 4 个字符用连字符分隔
|
||
return "-".join(code[i:i + 4] for i in range(0, len(code), 4))
|
||
|
||
|
||
class RedeemCodeStatus(str, Enum):
|
||
"""兑换码状态枚举"""
|
||
|
||
ACTIVE = "active" # 可用
|
||
USED = "used" # 已使用
|
||
DISABLED = "disabled" # 已禁用
|
||
EXPIRED = "expired" # 已过期
|
||
|
||
|
||
class RedeemCodeBatch(Base):
|
||
"""
|
||
兑换码批次模型
|
||
|
||
用于管理批量生成的兑换码
|
||
"""
|
||
|
||
__tablename__ = "redeem_code_batches"
|
||
|
||
# 主键
|
||
id: Mapped[str] = mapped_column(
|
||
String(36),
|
||
primary_key=True,
|
||
default=generate_uuid,
|
||
comment="批次唯一标识",
|
||
)
|
||
|
||
# 批次信息
|
||
name: Mapped[str] = mapped_column(
|
||
String(128),
|
||
nullable=False,
|
||
comment="批次名称",
|
||
)
|
||
description: Mapped[str | None] = mapped_column(
|
||
Text,
|
||
nullable=True,
|
||
comment="批次描述",
|
||
)
|
||
|
||
# 面值（单位额度）
|
||
face_value: Mapped[int] = mapped_column(
|
||
BigInteger,
|
||
nullable=False,
|
||
comment="面值（单位额度，1000 = 1.00）",
|
||
)
|
||
|
||
# 数量统计
|
||
total_count: Mapped[int] = mapped_column(
|
||
Integer,
|
||
default=0,
|
||
nullable=False,
|
||
comment="生成总数",
|
||
)
|
||
used_count: Mapped[int] = mapped_column(
|
||
Integer,
|
||
default=0,
|
||
nullable=False,
|
||
comment="已使用数量",
|
||
)
|
||
|
||
# 创建者
|
||
created_by: Mapped[str] = mapped_column(
|
||
String(36),
|
||
ForeignKey("users.id", ondelete="SET NULL"),
|
||
nullable=True,
|
||
comment="创建者 ID",
|
||
)
|
||
|
||
# 时间戳
|
||
created_at: Mapped[datetime] = mapped_column(
|
||
DateTime(timezone=True),
|
||
default=utc_now,
|
||
server_default=func.now(),
|
||
nullable=False,
|
||
comment="创建时间",
|
||
)
|
||
|
||
# 关系
|
||
codes: Mapped[list["RedeemCode"]] = relationship(
|
||
"RedeemCode",
|
||
back_populates="batch",
|
||
lazy="selectin",
|
||
)
|
||
creator: Mapped["User"] = relationship(
|
||
"User",
|
||
lazy="selectin",
|
||
)
|
||
|
||
@property
|
||
def display_face_value(self) -> str:
|
||
"""显示面值（2 位小数）"""
|
||
return f"{self.face_value / 1000:.2f}"
|
||
|
||
def __repr__(self) -> str:
|
||
return f"<RedeemCodeBatch(id={self.id!r}, name={self.name!r})>"
|
||
|
||
|
||
class RedeemCode(Base):
|
||
"""
|
||
兑换码模型
|
||
|
||
单个兑换码记录
|
||
"""
|
||
|
||
__tablename__ = "redeem_codes"
|
||
__table_args__ = (
|
||
Index("ix_redeem_codes_status_expires", "status", "expires_at"),
|
||
)
|
||
|
||
# 主键
|
||
id: Mapped[str] = mapped_column(
|
||
String(36),
|
||
primary_key=True,
|
||
default=generate_uuid,
|
||
comment="兑换码记录唯一标识",
|
||
)
|
||
|
||
# 兑换码（唯一索引）
|
||
code: Mapped[str] = mapped_column(
|
||
String(32),
|
||
unique=True,
|
||
nullable=False,
|
||
index=True,
|
||
default=generate_redeem_code,
|
||
comment="兑换码",
|
||
)
|
||
|
||
# 批次关联（可选）
|
||
batch_id: Mapped[str | None] = mapped_column(
|
||
String(36),
|
||
ForeignKey("redeem_code_batches.id", ondelete="SET NULL"),
|
||
nullable=True,
|
||
index=True,
|
||
comment="关联批次 ID",
|
||
)
|
||
|
||
# 面值（单位额度）
|
||
face_value: Mapped[int] = mapped_column(
|
||
BigInteger,
|
||
nullable=False,
|
||
comment="面值（单位额度，1000 = 1.00）",
|
||
)
|
||
|
||
# 状态
|
||
status: Mapped[RedeemCodeStatus] = mapped_column(
|
||
SQLEnum(RedeemCodeStatus),
|
||
default=RedeemCodeStatus.ACTIVE,
|
||
nullable=False,
|
||
index=True,
|
||
comment="兑换码状态",
|
||
)
|
||
|
||
# 使用限制
|
||
max_uses: Mapped[int] = mapped_column(
|
||
Integer,
|
||
default=1,
|
||
nullable=False,
|
||
comment="最大使用次数",
|
||
)
|
||
used_count: Mapped[int] = mapped_column(
|
||
Integer,
|
||
default=0,
|
||
nullable=False,
|
||
comment="已使用次数",
|
||
)
|
||
|
||
# 有效期
|
||
expires_at: Mapped[datetime | None] = mapped_column(
|
||
DateTime(timezone=True),
|
||
nullable=True,
|
||
comment="过期时间",
|
||
)
|
||
|
||
# 使用信息
|
||
used_by: Mapped[str | None] = mapped_column(
|
||
String(36),
|
||
ForeignKey("users.id", ondelete="SET NULL"),
|
||
nullable=True,
|
||
comment="使用者 ID（最后使用）",
|
||
)
|
||
used_at: Mapped[datetime | None] = mapped_column(
|
||
DateTime(timezone=True),
|
||
nullable=True,
|
||
comment="使用时间（最后使用）",
|
||
)
|
||
|
||
# 备注
|
||
remark: Mapped[str | None] = mapped_column(
|
||
Text,
|
||
nullable=True,
|
||
comment="备注",
|
||
)
|
||
|
||
# 创建者
|
||
created_by: Mapped[str | None] = mapped_column(
|
||
String(36),
|
||
ForeignKey("users.id", ondelete="SET NULL"),
|
||
nullable=True,
|
||
comment="创建者 ID",
|
||
)
|
||
|
||
# 时间戳
|
||
created_at: Mapped[datetime] = mapped_column(
|
||
DateTime(timezone=True),
|
||
default=utc_now,
|
||
server_default=func.now(),
|
||
nullable=False,
|
||
comment="创建时间",
|
||
)
|
||
updated_at: Mapped[datetime] = mapped_column(
|
||
DateTime(timezone=True),
|
||
default=utc_now,
|
||
onupdate=utc_now,
|
||
server_default=func.now(),
|
||
nullable=False,
|
||
comment="更新时间",
|
||
)
|
||
|
||
# 关系
|
||
batch: Mapped["RedeemCodeBatch | None"] = relationship(
|
||
"RedeemCodeBatch",
|
||
back_populates="codes",
|
||
lazy="selectin",
|
||
)
|
||
usage_logs: Mapped[list["RedeemCodeUsageLog"]] = relationship(
|
||
"RedeemCodeUsageLog",
|
||
back_populates="redeem_code",
|
||
lazy="selectin",
|
||
order_by="desc(RedeemCodeUsageLog.created_at)",
|
||
)
|
||
|
||
@property
|
||
def display_face_value(self) -> str:
|
||
"""显示面值（2 位小数）"""
|
||
return f"{self.face_value / 1000:.2f}"
|
||
|
||
@property
|
||
def is_valid(self) -> bool:
|
||
"""检查兑换码是否有效"""
|
||
if self.status != RedeemCodeStatus.ACTIVE:
|
||
return False
|
||
if self.used_count >= self.max_uses:
|
||
return False
|
||
if self.expires_at and self.expires_at < utc_now():
|
||
return False
|
||
return True
|
||
|
||
def __repr__(self) -> str:
|
||
return f"<RedeemCode(code={self.code!r}, status={self.status.value})>"
|
||
|
||
|
||
class RedeemCodeUsageLog(Base):
|
||
"""
|
||
兑换码使用日志模型
|
||
|
||
记录每次兑换的详细信息
|
||
"""
|
||
|
||
__tablename__ = "redeem_code_usage_logs"
|
||
__table_args__ = (
|
||
Index("ix_redeem_usage_user_created", "user_id", "created_at"),
|
||
)
|
||
|
||
# 主键
|
||
id: Mapped[str] = mapped_column(
|
||
String(36),
|
||
primary_key=True,
|
||
default=generate_uuid,
|
||
comment="日志唯一标识",
|
||
)
|
||
|
||
# 关联
|
||
redeem_code_id: Mapped[str] = mapped_column(
|
||
String(36),
|
||
ForeignKey("redeem_codes.id", ondelete="CASCADE"),
|
||
nullable=False,
|
||
index=True,
|
||
comment="关联兑换码 ID",
|
||
)
|
||
user_id: Mapped[str] = mapped_column(
|
||
String(36),
|
||
ForeignKey("users.id", ondelete="CASCADE"),
|
||
nullable=False,
|
||
index=True,
|
||
comment="使用者 ID",
|
||
)
|
||
transaction_id: Mapped[str | None] = mapped_column(
|
||
String(36),
|
||
ForeignKey("balance_transactions.id", ondelete="SET NULL"),
|
||
nullable=True,
|
||
comment="关联交易记录 ID",
|
||
)
|
||
|
||
# 兑换信息快照
|
||
code_snapshot: Mapped[str] = mapped_column(
|
||
String(32),
|
||
nullable=False,
|
||
comment="兑换码快照",
|
||
)
|
||
face_value: Mapped[int] = mapped_column(
|
||
BigInteger,
|
||
nullable=False,
|
||
comment="面值（单位额度）",
|
||
)
|
||
|
||
# 客户端信息
|
||
ip_address: Mapped[str | None] = mapped_column(
|
||
String(64),
|
||
nullable=True,
|
||
comment="客户端 IP",
|
||
)
|
||
user_agent: Mapped[str | None] = mapped_column(
|
||
String(512),
|
||
nullable=True,
|
||
comment="User Agent",
|
||
)
|
||
|
||
# 时间戳
|
||
created_at: Mapped[datetime] = mapped_column(
|
||
DateTime(timezone=True),
|
||
default=utc_now,
|
||
server_default=func.now(),
|
||
nullable=False,
|
||
comment="使用时间",
|
||
)
|
||
|
||
# 关系
|
||
redeem_code: Mapped["RedeemCode"] = relationship(
|
||
"RedeemCode",
|
||
back_populates="usage_logs",
|
||
lazy="selectin",
|
||
)
|
||
user: Mapped["User"] = relationship(
|
||
"User",
|
||
lazy="selectin",
|
||
)
|
||
|
||
@property
|
||
def display_face_value(self) -> str:
|
||
"""显示面值（2 位小数）"""
|
||
return f"{self.face_value / 1000:.2f}"
|
||
|
||
def __repr__(self) -> str:
|
||
return (
|
||
f"<RedeemCodeUsageLog(code={self.code_snapshot!r}, "
|
||
f"user_id={self.user_id!r})>"
|
||
)
|
||
|