92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.card_key_types import CARD_TYPE_DURATION, CARD_TYPE_POINTS
|
|
from app.models.card_key import CardKey
|
|
from app.models.user import User
|
|
from app.schemas.card_key import CardKeyActivateResponse
|
|
|
|
|
|
class CardKeyActivateError(Exception):
|
|
def __init__(self, message: str) -> None:
|
|
self.message = message
|
|
super().__init__(message)
|
|
|
|
|
|
def _utc_now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _ensure_utc(value: datetime) -> datetime:
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value
|
|
|
|
|
|
def _extend_vip_end_time(current: datetime | None, days: int) -> datetime:
|
|
now = _utc_now()
|
|
if current is not None:
|
|
end = _ensure_utc(current)
|
|
base = end if end > now else now
|
|
else:
|
|
base = now
|
|
return base + timedelta(days=days)
|
|
|
|
|
|
async def activate_card_key(
|
|
db: AsyncSession,
|
|
*,
|
|
username: str,
|
|
serial_number: str,
|
|
) -> CardKeyActivateResponse:
|
|
user = await db.scalar(select(User).where(User.username == username))
|
|
if user is None:
|
|
raise CardKeyActivateError("用户不存在")
|
|
|
|
result = await db.execute(
|
|
select(CardKey)
|
|
.where(CardKey.serial_number == serial_number)
|
|
.with_for_update()
|
|
)
|
|
card = result.scalar_one_or_none()
|
|
if card is None:
|
|
raise CardKeyActivateError("卡密不存在")
|
|
|
|
if card.activated_at is not None:
|
|
raise CardKeyActivateError("卡密已被使用")
|
|
|
|
now = _utc_now()
|
|
card.activated_at = now
|
|
card.user_id = user.id
|
|
card.username = user.username
|
|
|
|
if card.card_type == CARD_TYPE_DURATION:
|
|
if card.duration_days < 1:
|
|
raise CardKeyActivateError("卡密配置无效")
|
|
user.vip_end_time = _extend_vip_end_time(user.vip_end_time, card.duration_days)
|
|
await db.flush()
|
|
await db.refresh(user)
|
|
return CardKeyActivateResponse(
|
|
card_type=card.card_type,
|
|
vip_end_time=user.vip_end_time,
|
|
points=user.points,
|
|
added_days=card.duration_days,
|
|
)
|
|
|
|
if card.card_type == CARD_TYPE_POINTS:
|
|
if card.points_amount is None or card.points_amount <= 0:
|
|
raise CardKeyActivateError("卡密配置无效")
|
|
user.points = (user.points or 0.0) + card.points_amount
|
|
await db.flush()
|
|
await db.refresh(user)
|
|
return CardKeyActivateResponse(
|
|
card_type=card.card_type,
|
|
vip_end_time=user.vip_end_time,
|
|
points=user.points,
|
|
added_points=card.points_amount,
|
|
)
|
|
|
|
raise CardKeyActivateError("卡密类型无效")
|