import time from typing import Dict class PhoneRateLimiter: def __init__(self, window_seconds: int = 60) -> None: """手机号限流器 Args: window_seconds: 限流窗口秒数 Returns: None """ self.window = window_seconds self.last_sent: Dict[str, float] = {} def allow(self, phone: str) -> bool: """校验是否允许发送 Args: phone: 手机号 Returns: True 表示允许发送,False 表示命中限流 """ now = time.time() ts = self.last_sent.get(phone, 0) if now - ts < self.window: return False self.last_sent[phone] = now return True def next_allowed_at(self, phone: str) -> float: """返回下一次允许发送的时间戳 Args: phone: 手机号 Returns: 时间戳(秒) """ ts = self.last_sent.get(phone, 0) return ts + self.window def reset(self) -> None: """重置限流状态 Returns: None """ self.last_sent.clear()