90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
"""MinHash 指纹与去重。
|
||
|
||
纯 Python 实现 MinHash（字符 3-gram shingles + 稳定哈希 + 随机置换
|
||
模拟），生成稳定十六进制指纹。用于相同 (host, service, message_fingerprint)
|
||
短窗口去重（architecture.md 5.3.2，Redis `hms:log:fp:{fingerprint}`）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import random
|
||
import re
|
||
import struct
|
||
import zlib
|
||
from typing import Dict, Iterable, Optional, Set
|
||
|
||
_MAX_HASH = (1 << 32) - 1
|
||
|
||
|
||
def _h(token: str) -> int:
|
||
return zlib.crc32(token.encode("utf-8")) & _MAX_HASH
|
||
|
||
|
||
def shingles(text: str, k: int = 3) -> Set[str]:
|
||
"""将文本切分为 k-gram 集合（字符级，跨空白压缩）。"""
|
||
cleaned = re.sub(r"\s+", " ", text or "").strip()
|
||
if len(cleaned) < k:
|
||
# 短文本退化为整串
|
||
return {cleaned} if cleaned else set()
|
||
return {cleaned[i : i + k] for i in range(len(cleaned) - k + 1)}
|
||
|
||
|
||
class MinHash:
|
||
"""MinHash 指纹器，相同（或高度相似）文本产生相同/相近指纹。"""
|
||
|
||
def __init__(self, num_perm: int = 64, seed: int = 42):
|
||
self.num_perm = num_perm
|
||
rng = random.Random(seed)
|
||
self.a = [rng.randint(1, _MAX_HASH) for _ in range(num_perm)]
|
||
self.b = [rng.randint(0, _MAX_HASH) for _ in range(num_perm)]
|
||
|
||
def signature(self, tokens: Iterable[str]) -> list:
|
||
hashes = [_h(t) for t in tokens]
|
||
if not hashes:
|
||
return [0] * self.num_perm
|
||
sig = []
|
||
for ai, bi in zip(self.a, self.b):
|
||
sig.append(min(((ai * x + bi) & _MAX_HASH) for x in hashes))
|
||
return sig
|
||
|
||
def fingerprint(self, text: str) -> str:
|
||
"""返回文本的 16 字符十六进制指纹。"""
|
||
tokens = shingles(text)
|
||
if not tokens:
|
||
return "0" * 16
|
||
sig = self.signature(tokens)
|
||
digest = hashlib.sha256()
|
||
for v in sig:
|
||
digest.update(struct.pack(">I", v))
|
||
return digest.hexdigest()[:16]
|
||
|
||
|
||
class Deduplicator:
|
||
"""短窗口去重器。生产可替换为 Redis SETNX，测试用进程内实现。"""
|
||
|
||
def __init__(self, window_seconds: int = 300, clock=None):
|
||
self.window_seconds = window_seconds
|
||
self._clock = clock or _time_now
|
||
self._seen: Dict[str, float] = {}
|
||
|
||
def is_duplicate(self, fingerprint: str) -> bool:
|
||
now = self._clock()
|
||
last = self._seen.get(fingerprint)
|
||
if last is not None and (now - last) <= self.window_seconds:
|
||
return True
|
||
self._seen[fingerprint] = now
|
||
return False
|
||
|
||
def mark(self, fingerprint: str) -> None:
|
||
self._seen[fingerprint] = self._clock()
|
||
|
||
def clear(self) -> None:
|
||
self._seen.clear()
|
||
|
||
|
||
def _time_now() -> float:
|
||
import time
|
||
|
||
return time.time()
|