120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
"""近重复（抄袭）JD 检测：纯 Python MinHash（64 哈希、5-gram shingles）+ LSH banding。
|
||
|
||
招聘网站之间存在大量"同文不同公司"的抄袭 JD，直接入图会导致技能频次虚高。
|
||
本模块把近重复 JD 聚簇：簇内保留最早来源，其余标记 is_duplicate 并降权。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import logging
|
||
import random
|
||
|
||
from app.models import JDRecord
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
NUM_HASHES = 256 # MinHash 哈希函数个数（越高估计越稳；128 时边沿样本偏差可达 ±0.1 误杀植入簇）
|
||
SHINGLE_N = 5 # 字符 5-gram shingles
|
||
LSH_BANDS = 16 # 16 bands × 8 rows（严格分带，抑制模板化文本误配）
|
||
_PRIME = (1 << 61) - 1 # 梅森素数，用于 (a*x+b) mod p 哈希族
|
||
_MAXINT = (1 << 32) - 1
|
||
|
||
|
||
def shingles(text: str, n: int = SHINGLE_N) -> set[int]:
|
||
"""把文本切成字符 n-gram shingles，并映射为 32 位整数集合。"""
|
||
text = "".join(text.split()) # 去掉空白，抵抗排版差异
|
||
if len(text) < n:
|
||
text = text + " " * (n - len(text))
|
||
out = set()
|
||
for i in range(len(text) - n + 1):
|
||
h = hashlib.md5(text[i:i + n].encode("utf-8")).digest()
|
||
out.add(int.from_bytes(h[:4], "little"))
|
||
return out
|
||
|
||
|
||
class MinHash:
|
||
"""(a*x+b) mod p 置换哈希族实现的 MinHash 签名。"""
|
||
|
||
def __init__(self, num_hashes: int = NUM_HASHES, seed: int = 42):
|
||
self.num_hashes = num_hashes
|
||
# 确定性大随机系数，保证 (a*x+b) mod p 是真置换哈希且可复现
|
||
rng = random.Random(seed)
|
||
self._a = [rng.randrange(1, _PRIME) for _ in range(num_hashes)]
|
||
self._b = [rng.randrange(0, _PRIME) for _ in range(num_hashes)]
|
||
|
||
def signature(self, shingle_set: set[int]) -> list[int]:
|
||
return [min((a * x + b) % _PRIME for x in shingle_set)
|
||
for a, b in zip(self._a, self._b)]
|
||
|
||
def estimate(self, s1: list[int], s2: list[int]) -> float:
|
||
return sum(1 for x, y in zip(s1, s2) if x == y) / self.num_hashes
|
||
|
||
|
||
def _lsh_candidates(sig: list[int], bands: int, buckets: dict) -> None:
|
||
"""把签名按 band 切段放入哈希桶，同桶者成为候选对。"""
|
||
rows = len(sig) // bands
|
||
for b in range(bands):
|
||
key = (b, tuple(sig[b * rows:(b + 1) * rows]))
|
||
buckets.setdefault(key, []).append(sig)
|
||
|
||
|
||
def find_near_duplicates(
|
||
records: list[JDRecord],
|
||
threshold: float = 0.85,
|
||
num_hashes: int = NUM_HASHES,
|
||
bands: int = LSH_BANDS,
|
||
) -> list[list[str]]:
|
||
"""对全部 JD 做近重复聚簇，返回簇列表（每簇为 jid 列表，保留者排第一）。
|
||
|
||
簇内按 (publish_date, jid) 排序，最早来源为保留者；其余记录被标记
|
||
``is_duplicate=True``、``dup_of=<保留者 jid>`` 并降权 0.2。
|
||
"""
|
||
mh = MinHash(num_hashes)
|
||
sigs: dict[str, list[int]] = {}
|
||
for r in records:
|
||
sigs[r.jid] = mh.signature(shingles(r.text))
|
||
|
||
buckets: dict = {}
|
||
for sig in sigs.values():
|
||
_lsh_candidates(sig, bands, buckets)
|
||
|
||
# 并查集聚簇：候选对需通过完整签名相似度复核
|
||
parent = {r.jid: r.jid for r in records}
|
||
|
||
def find(x: str) -> str:
|
||
while parent[x] != x:
|
||
parent[x] = parent[parent[x]]
|
||
x = parent[x]
|
||
return x
|
||
|
||
def union(x: str, y: str) -> None:
|
||
parent[find(x)] = find(y)
|
||
|
||
jid_of_sig = {id(sig): jid for jid, sig in sigs.items()}
|
||
for members in buckets.values():
|
||
for i in range(len(members)):
|
||
for j in range(i + 1, len(members)):
|
||
if mh.estimate(members[i], members[j]) >= threshold:
|
||
union(jid_of_sig[id(members[i])], jid_of_sig[id(members[j])])
|
||
|
||
groups: dict[str, list[JDRecord]] = {}
|
||
by_jid = {r.jid: r for r in records}
|
||
for r in records:
|
||
groups.setdefault(find(r.jid), []).append(r)
|
||
|
||
clusters: list[list[str]] = []
|
||
for members in groups.values():
|
||
if len(members) < 2:
|
||
continue
|
||
members.sort(key=lambda r: (r.publish_date, r.jid))
|
||
keeper = members[0]
|
||
cluster = [keeper.jid]
|
||
for dup in members[1:]:
|
||
dup.is_duplicate = True
|
||
dup.dup_of = keeper.jid
|
||
dup.weight = round(dup.weight * 0.2, 4) # 抄袭来源降权
|
||
cluster.append(dup.jid)
|
||
clusters.append(cluster)
|
||
logger.info("检出近重复簇：保留 %s，降权 %d 条", keeper.jid, len(members) - 1)
|
||
return clusters
|