68 lines
3.5 KiB
Python
68 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Dict, List
|
|
|
|
|
|
class CooldownManager:
|
|
def __init__(self, config: Dict):
|
|
self.config = config or {}
|
|
self.last_reply_at: Dict[str, float] = {}
|
|
self.at_mention_history: Dict[str, List[float]] = {}
|
|
self.user_reply_history: Dict[str, List[float]] = {}
|
|
|
|
def pass_cooldown(self, room_id: str, sender: str, trigger: Dict) -> bool:
|
|
current_ts = time.time()
|
|
room_cd = int(self.config.get("group_reply_cooldown_sec", 45))
|
|
user_cd = int(self.config.get("same_user_followup_cooldown_sec", 10))
|
|
at_min_interval = int(self.config.get("at_mention_min_interval_sec", 8))
|
|
at_burst_window = int(self.config.get("at_mention_burst_window_sec", 90))
|
|
at_burst_limit = int(self.config.get("at_mention_burst_limit", 4))
|
|
at_silent_sec = int(self.config.get("at_mention_silent_sec", 180))
|
|
directed_burst_window = int(self.config.get("directed_burst_window_sec", 240))
|
|
directed_burst_limit = int(self.config.get("directed_burst_limit", 4))
|
|
directed_silent_sec = int(self.config.get("directed_burst_silent_sec", 480))
|
|
last_room_reply = self.last_reply_at.get(room_id, 0.0)
|
|
user_key = f"{room_id}:{sender}"
|
|
user_history = [ts for ts in self.user_reply_history.get(user_key, []) if current_ts - ts <= directed_burst_window]
|
|
self.user_reply_history[user_key] = user_history
|
|
|
|
if trigger.get("is_at") or trigger.get("is_followup") or trigger.get("is_directed"):
|
|
if user_history and (current_ts - user_history[-1]) < user_cd:
|
|
trigger["_cooldown_reason"] = "same_user_directed_cooldown"
|
|
return False
|
|
if len(user_history) >= directed_burst_limit and (current_ts - user_history[-1]) < directed_silent_sec:
|
|
trigger["_cooldown_reason"] = "same_user_directed_silent"
|
|
return False
|
|
|
|
if trigger.get("trigger_type") == "at_trigger":
|
|
history = [ts for ts in self.at_mention_history.get(room_id, []) if current_ts - ts <= at_burst_window]
|
|
self.at_mention_history[room_id] = history
|
|
if history and (current_ts - history[-1]) < at_min_interval:
|
|
trigger["_cooldown_reason"] = "at_min_interval"
|
|
return False
|
|
if len(history) >= at_burst_limit:
|
|
if (current_ts - history[-1]) < at_silent_sec:
|
|
trigger["_cooldown_reason"] = "at_burst_silent"
|
|
return False
|
|
self.at_mention_history[room_id] = []
|
|
self.at_mention_history.setdefault(room_id, []).append(current_ts)
|
|
self.user_reply_history.setdefault(user_key, []).append(current_ts)
|
|
return True
|
|
|
|
if trigger.get("is_question") or trigger.get("is_followup"):
|
|
trigger["_cooldown_reason"] = "followup_cooldown"
|
|
allowed = (current_ts - last_room_reply) >= user_cd
|
|
if allowed and (trigger.get("is_directed") or trigger.get("is_followup")):
|
|
self.user_reply_history.setdefault(user_key, []).append(current_ts)
|
|
return allowed
|
|
|
|
trigger["_cooldown_reason"] = "group_cooldown"
|
|
allowed = (current_ts - last_room_reply) >= room_cd
|
|
if allowed and trigger.get("is_directed"):
|
|
self.user_reply_history.setdefault(user_key, []).append(current_ts)
|
|
return allowed
|
|
|
|
def note_reply(self, room_id: str) -> None:
|
|
self.last_reply_at[room_id] = time.time()
|