guke
40 lines
1.6 KiB
Python
40 lines
1.6 KiB
Python
"""手机号换绑台账(phone_rebind_log)的查询与写入。见 M2 spec §4.1。"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.phone_rebind_log import PhoneRebindLog
|
|
|
|
|
|
def rebound_within_days(db: Session, phone: str, days: int) -> bool:
|
|
"""该手机号在最近 days 天内是否换绑过(命中 → 禁止再次换绑)。"""
|
|
since = datetime.now(timezone.utc) - timedelta(days=days)
|
|
stmt = (
|
|
select(PhoneRebindLog.id)
|
|
.where(PhoneRebindLog.phone == phone, PhoneRebindLog.rebound_at >= since)
|
|
.limit(1)
|
|
)
|
|
return db.execute(stmt).first() is not None
|
|
|
|
|
|
def remaining_block_days(db: Session, phone: str, days: int) -> int:
|
|
"""距离该手机号可再次换绑还剩几天(向上取整;无记录返回 0)。"""
|
|
last = db.execute(
|
|
select(func.max(PhoneRebindLog.rebound_at)).where(PhoneRebindLog.phone == phone)
|
|
).scalar_one_or_none()
|
|
if last is None:
|
|
return 0
|
|
if last.tzinfo is None: # SQLite 取回 naive datetime,按 UTC 归一
|
|
last = last.replace(tzinfo=timezone.utc)
|
|
remaining = (last + timedelta(days=days) - datetime.now(timezone.utc)).total_seconds()
|
|
return max(0, math.ceil(remaining / 86400))
|
|
|
|
|
|
def add_rebind_log(db: Session, *, phone: str, old_user_id: int | None, new_user_id: int, source: str) -> None:
|
|
"""写一条换绑台账(**不 commit**,交给调用方 rebind_account 的单事务)。"""
|
|
db.add(PhoneRebindLog(phone=phone, old_user_id=old_user_id, new_user_id=new_user_id, source=source))
|