39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
|
"""客服对话中的重复兴趣主题计数。"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
|
||
|
|
class InterestTopicTracker:
|
||
|
|
"""使用 Redis 统计客户当天对同一主题的重复关注次数。"""
|
||
|
|
|
||
|
|
KEY_PREFIX = "customer:memory:interest"
|
||
|
|
WINDOW_SECONDS = 24 * 60 * 60
|
||
|
|
DEFAULT_THRESHOLD = 3
|
||
|
|
|
||
|
|
def __init__(self, redis, *, threshold: int = DEFAULT_THRESHOLD):
|
||
|
|
self.redis = redis
|
||
|
|
self.threshold = threshold
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def key(cls, customer_id: int, tag: str, now: datetime | None = None) -> str:
|
||
|
|
"""生成按客户、自然日和主题隔离的 Redis 计数 Key。"""
|
||
|
|
day = (now or datetime.now()).strftime("%Y%m%d")
|
||
|
|
digest = hashlib.sha256(tag.strip().lower().encode("utf-8")).hexdigest()[:16]
|
||
|
|
return f"{cls.KEY_PREFIX}:{customer_id}:{day}:{digest}"
|
||
|
|
|
||
|
|
async def record(self, customer_id: int, tag: str) -> tuple[int, bool]:
|
||
|
|
"""记录一次主题关注，返回累计次数和是否达到长期记忆阈值。"""
|
||
|
|
if not tag or not tag.strip():
|
||
|
|
raise ValueError("兴趣主题不能为空")
|
||
|
|
key = self.key(customer_id, tag)
|
||
|
|
count = await self.redis.incr(key)
|
||
|
|
if count == 1:
|
||
|
|
await self.redis.expire(key, self.WINDOW_SECONDS)
|
||
|
|
return int(count), int(count) >= self.threshold
|
||
|
|
|
||
|
|
|
||
|
|
__all__ = ["InterestTopicTracker"]
|