"""文本去重策略工厂 支持三种策略: - none:关闭去重,每次都跑完整流水线 - sha256:精确匹配(hash 完全相同才算重复) - simhash:近似匹配(海明距离 <= 阈值视为重复) 每种策略实现 DedupStrategy 协议:lookup(text) -> dict|None、record(text, result) -> None。 工厂 get_dedup_strategy() 根据 runtime_settings.dedup 选择实现。 所有策略共享同一 Redis key 前缀,但 key 后缀按策略区分避免冲突。 """ from __future__ import annotations import hashlib from typing import Any, Protocol, runtime_checkable import structlog from app.core.runtime_settings import get_runtime_settings from app.services.redis import RedisCache logger = structlog.get_logger() # Redis 去重 key 前缀,按策略分桶 _DEDUP_KEY_PREFIX = "dedup:" @runtime_checkable class DedupStrategy(Protocol): """去重策略协议""" async def lookup(self, text: str) -> dict[str, Any] | None: """查询文本是否已入库;命中返回旧 IngestionResult(dict),未命中返回 None""" ... async def record(self, text: str, result_dict: dict[str, Any]) -> None: """记录文本已入库,供后续命中复用""" ... class NoopDedupStrategy: """关闭去重:永远不命中,也不记录""" async def lookup(self, text: str) -> dict[str, Any] | None: return None async def record(self, text: str, result_dict: dict[str, Any]) -> None: return None class Sha256DedupStrategy: """SHA256 精确去重:完全相同文本才算重复""" def __init__(self, redis: RedisCache, ttl_seconds: int) -> None: self._redis = redis self._ttl = ttl_seconds def _key(self, text: str) -> str: h = hashlib.sha256(text.encode("utf-8")).hexdigest() return f"{_DEDUP_KEY_PREFIX}sha256:{h}" async def lookup(self, text: str) -> dict[str, Any] | None: try: return await self._redis.get_json(self._key(text)) except Exception: logger.warning("sha256 去重查询失败,降级未命中", exc_info=True) return None async def record(self, text: str, result_dict: dict[str, Any]) -> None: try: await self._redis.set_json(self._key(text), result_dict, ttl=self._ttl) except Exception: logger.warning("sha256 去重记录写入失败", exc_info=True) # SimHash 实现(64 位) _MASK_64 = (1 << 64) - 1 def _simhash(text: str, token_size: int = 4) -> int: """计算文本 64 位 simhash 简化实现:按 token_size 字符滑窗分词,每段 md5 → 128 位 → 取低 64 位作为 hash, 逐位加权(hash 该位为 1 则 +1,为 0 则 -1),最终符号位定 1/0。 """ if not text: return 0 tokens = [text[i : i + token_size] for i in range(0, len(text), token_size)] weights = [1] * 64 vec = [0] * 64 for token in tokens: h = int(hashlib.md5(token.encode("utf-8")).hexdigest(), 16) & _MASK_64 for i in range(64): bit = (h >> i) & 1 vec[i] += weights[i] if bit else -weights[i] fingerprint = 0 for i in range(64): if vec[i] > 0: fingerprint |= (1 << i) return fingerprint def _hamming_distance(a: int, b: int) -> int: return bin((a ^ b) & _MASK_64).count("1") class SimhashDedupStrategy: """SimHash 近似去重:海明距离 <= 阈值视为重复 Redis 中存所有已入库文本的 simhash(value 为 IngestionResult + simhash)。 lookup 时遍历所有候选 simhash 算海明距离,找最近的一个 <= 阈值则命中。 注:本实现为简化版,全表扫描,适合中小规模知识库;超大规模需换 LSH 索引。 """ INDEX_KEY = f"{_DEDUP_KEY_PREFIX}simhash:index" # list 形式存所有 simhash+key def __init__(self, redis: RedisCache, ttl_seconds: int, threshold: int = 3) -> None: self._redis = redis self._ttl = ttl_seconds self._threshold = max(0, min(64, threshold)) async def lookup(self, text: str) -> dict[str, Any] | None: try: fingerprint = _simhash(text) # 简化:扫描所有 simhash 记录找最近的 # 用 set:dedup:simhash:fps 存所有 fingerprint,每个 fp 对应一个 record key # 这里用 list key 简化(适合小规模) index = await self._redis.get_json(self.INDEX_KEY) or {"entries": []} entries = index.get("entries", []) best_dist = self._threshold + 1 best_key: str | None = None for entry in entries: fp = entry.get("fingerprint") key = entry.get("key") if fp is None or key is None: continue dist = _hamming_distance(fingerprint, int(fp)) if dist <= self._threshold and dist < best_dist: best_dist = dist best_key = key if best_key is None: return None return await self._redis.get_json(best_key) except Exception: logger.warning("simhash 去重查询失败,降级未命中", exc_info=True) return None async def record(self, text: str, result_dict: dict[str, Any]) -> None: try: fingerprint = _simhash(text) key = f"{_DEDUP_KEY_PREFIX}simhash:{fingerprint:016x}" # 1. 写记录 await self._redis.set_json(key, result_dict, ttl=self._ttl) # 2. 更新索引 index = await self._redis.get_json(self.INDEX_KEY) or {"entries": []} entries = index.get("entries", []) entries.append({"fingerprint": fingerprint, "key": key}) await self._redis.set_json(self.INDEX_KEY, {"entries": entries}, ttl=self._ttl) except Exception: logger.warning("simhash 去重记录写入失败", exc_info=True) # ---------------------------------------------------------------------------- # # 工厂 # ---------------------------------------------------------------------------- # _strategy_cache: DedupStrategy | None = None _strategy_signature: tuple[str, int, int] | None = None # (strategy, ttl, threshold) def get_dedup_strategy(redis: RedisCache | None) -> DedupStrategy: """按 runtime_settings.dedup 构造去重策略 Redis 不可用时降级为 NoopDedupStrategy(关闭去重),不影响主流程。 策略配置变更时自动重建单例。 """ global _strategy_cache, _strategy_signature rt = get_runtime_settings() sig = (rt.dedup.strategy, rt.dedup.ttl_seconds, rt.dedup.simhash_threshold) if _strategy_cache is not None and _strategy_signature == sig: return _strategy_cache if redis is None or rt.dedup.strategy == "none": strategy: DedupStrategy = NoopDedupStrategy() elif rt.dedup.strategy == "sha256": strategy = Sha256DedupStrategy(redis=redis, ttl_seconds=rt.dedup.ttl_seconds) elif rt.dedup.strategy == "simhash": strategy = SimhashDedupStrategy( redis=redis, ttl_seconds=rt.dedup.ttl_seconds, threshold=rt.dedup.simhash_threshold ) else: logger.warning("未知去重策略,降级为 noop", requested=rt.dedup.strategy) strategy = NoopDedupStrategy() _strategy_cache = strategy _strategy_signature = sig return strategy def invalidate_dedup_strategy_cache() -> None: """清除策略缓存(runtime_settings 更新后调用)""" global _strategy_cache, _strategy_signature _strategy_cache = None _strategy_signature = None def compute_text_hash(text: str) -> str: """计算文本哈希(兼容旧接口,sha256 策略下与 dedup key 一致)""" return hashlib.sha256(text.encode("utf-8")).hexdigest() # 兼容旧测试与外部引用:保留 DEDUP_KEY_PREFIX 导出 DEDUP_KEY_PREFIX = _DEDUP_KEY_PREFIX