90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
"""三层去重的指纹算法。
|
||
|
||
核心:
|
||
- normalize_content: 把 content 折叠成纯净文本，用于跨源比对
|
||
- content_hash: normalize 后 SHA1[:16]
|
||
- simhash64: 字符 3-gram + md5 加权累加，产出 64 位无符号整数
|
||
- hamming: 两个 SimHash 的汉明距离
|
||
|
||
设计取舍:
|
||
SimHash 的"分词"用字符 3-gram 而非英文分词库。理由:
|
||
1. 字符 3-gram 对英文和中文同样有效，无需外部 NLP 依赖
|
||
2. 英文字符级 3-gram 天然捕获词根、前缀、后缀信息
|
||
3. 跨语言场景（英文源可能引用中文/日文公司名）字符级更鲁棒
|
||
"""
|
||
|
||
import hashlib
|
||
import unicodedata
|
||
|
||
# 64 位 SimHash 位宽
|
||
SIMHASH_BITS = 64
|
||
SIMHASH_MASK = (1 << SIMHASH_BITS) - 1
|
||
|
||
# 默认 SimHash 汉明距离阈值（≤ 此值视为重复）
|
||
DEFAULT_HAMMING_THRESHOLD = 3
|
||
|
||
# 字符 n-gram 长度
|
||
NGRAM_SIZE = 3
|
||
|
||
|
||
def normalize_content(text: str) -> str:
|
||
"""把 content 折叠成"无空白无标点"形式，用于 L2 内容 hash 与 SimHash 输入。
|
||
|
||
使用 Unicode 类别判断:
|
||
- P* Punctuation（所有中英文标点）
|
||
- Z* Separator（空格 / 行 / 段分隔符）
|
||
- C* Control（NUL / 换行控制等）
|
||
保留 L*（字母）、N*（数字）、S*（符号，如 +/-、% 等），以及 CJK 字符。
|
||
"""
|
||
if not text:
|
||
return ""
|
||
return "".join(
|
||
ch for ch in text if unicodedata.category(ch)[0] not in ("P", "Z", "C")
|
||
)
|
||
|
||
|
||
def content_hash(text: str) -> str:
|
||
"""对 normalize_content(text) 做 SHA1，取前 16 hex 字符。"""
|
||
norm = normalize_content(text)
|
||
return hashlib.sha1(norm.encode("utf-8")).hexdigest()[:16]
|
||
|
||
|
||
def _ngrams(text: str, n: int = NGRAM_SIZE) -> list[str]:
|
||
"""字符级 n-gram。文本短于 n 时，直接整体作为单个 token。"""
|
||
if len(text) < n:
|
||
return [text] if text else []
|
||
return [text[i : i + n] for i in range(len(text) - n + 1)]
|
||
|
||
|
||
def simhash64(text: str) -> int:
|
||
"""64 位 SimHash。返回无符号整数，空文本返回 0。"""
|
||
norm = normalize_content(text)
|
||
if not norm:
|
||
return 0
|
||
|
||
grams = _ngrams(norm)
|
||
if not grams:
|
||
return 0
|
||
|
||
v = [0] * SIMHASH_BITS
|
||
for gram in grams:
|
||
h = int(hashlib.md5(gram.encode("utf-8"), usedforsecurity=False).hexdigest(), 16)
|
||
# 取低 64 位
|
||
h64 = h & SIMHASH_MASK
|
||
for i in range(SIMHASH_BITS):
|
||
if (h64 >> i) & 1:
|
||
v[i] += 1
|
||
else:
|
||
v[i] -= 1
|
||
|
||
fp = 0
|
||
for i in range(SIMHASH_BITS):
|
||
if v[i] > 0:
|
||
fp |= 1 << i
|
||
return fp
|
||
|
||
|
||
def hamming(a: int, b: int) -> int:
|
||
"""两个 SimHash 的汉明距离。"""
|
||
return bin((a ^ b) & SIMHASH_MASK).count("1")
|