数据去重与污染检测：从 n-gram 到语义候选的最小复现

"""Controlled dedup audit: no training, no benchmark-performance claims. Python 3.9+."""
import argparse
from collections import Counter
import hashlib
import itertools
import json
import math
from pathlib import Path
import platform
import random
import re
import unicodedata

SEED = 64
# English fixtures deliberately make tokenizer assumptions inspectable.
DOCS = [
    ('t0', 'train', 'The small cat sleeps on the warm mat.'),
    ('t1', 'train', '  THE small cat sleeps on the warm mat!'),
    ('t2', 'train', 'A tiny feline rests on a cozy rug.'),
    ('t3', 'train', 'The small cat never sleeps on the warm mat.'),
    ('t4', 'train', 'The red telescope observes a distant galaxy.'),
    ('t5', 'train', 'The archive states the small cat sleeps on the warm mat beside a lengthy unrelated appendix about planetary orbits.'),
    ('b0', 'benchmark', 'The small cat sleeps on the warm mat.'),
    ('b1', 'benchmark', 'A tiny feline rests on a cozy rug.'),
    ('b2', 'benchmark', 'The ocean wave breaks along a rocky shore.'),
]
# A tiny hand-written synonym mapping, NOT a pretrained semantic encoder.
SYNONYMS = dict(zip('tiny feline rests cozy rug'.split(), 'small cat sleeps warm mat'.split()))
STOP = {'a', 'the', 'on'}
GOLD = {frozenset(p) for p in [('t0','t1'), ('t0','t2'), ('t1','t2')]}


def normalize(text):
    return ' '.join(re.findall(r'\w+', unicodedata.normalize('NFKC', text).casefold()))


def grams(text, n):
    tokens = normalize(text).split()
    return {tuple(tokens[i:i+n]) for i in range(len(tokens)-n+1)}


def jaccard(a, b):
    return len(a & b) / len(a | b) if a or b else 0.0


def containment(query, doc):
    return len(query & doc) / len(query) if query else 0.0


def toy_vectors(texts):
    bags = [Counter(SYNONYMS.get(w, w) for w in normalize(t).split() if w not in STOP) for t in texts]
    vocab = sorted(set().union(*(b.keys() for b in bags)))
    rows = []
    for bag in bags:
        row = [float(bag[w]) for w in vocab]
        norm = math.sqrt(sum(v*v for v in row))
        rows.append([v/norm if norm else 0.0 for v in row])
    return rows


def components(ids, edges):
    parent = {i:i for i in ids}
    def root(i):
        while parent[i] != i:
            parent[i] = parent[parent[i]]
            i = parent[i]
        return i
    for a,b in edges:
        ra,rb = root(a),root(b)
        parent[max(ra,rb)] = min(ra,rb)
    groups = {}
    for i in ids:
        groups.setdefault(root(i), []).append(i)
    return sorted(sorted(g) for g in groups.values())


def metrics(pred):
    tp = len(pred & GOLD)
    return {'tp':tp, 'fp':len(pred-GOLD), 'fn':len(GOLD-pred),
            'precision':tp/len(pred) if pred else None, 'recall':tp/len(GOLD)}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--out', type=Path, default=Path(__file__).with_name('results.json'))
    parser.add_argument('--model-dir', type=Path, help='Optional local SentenceTransformer snapshot; never auto-download')
    args = parser.parse_args()
    random.seed(SEED)
    texts = [d[2] for d in DOCS]
    engine = 'handwritten-synonym-bag; NOT pretrained embedding'
    model_files = None
    if args.model_dir:
        import torch
        from sentence_transformers import SentenceTransformer
        torch.manual_seed(SEED)
        torch.set_num_threads(1)
        torch.use_deterministic_algorithms(True)
        model = SentenceTransformer(str(args.model_dir.resolve()), device='cpu', local_files_only=True)
        vectors = model.encode([normalize(t) for t in texts], normalize_embeddings=True, show_progress_bar=False).tolist()
        engine = str(args.model_dir.resolve())
        model_files = {}
        for p in sorted(args.model_dir.rglob('*')):
            if p.is_file():
                h = hashlib.sha256()
                with p.open('rb') as f:
                    for chunk in iter(lambda:f.read(1024*1024), b''): h.update(chunk)
                model_files[str(p.relative_to(args.model_dir))] = h.hexdigest()
    else:
        vectors = toy_vectors(texts)
    by_id = {d[0]:i for i,d in enumerate(DOCS)}
    train = [d[0] for d in DOCS if d[1]=='train']
    bench = [d[0] for d in DOCS if d[1]=='benchmark']
    def compare(a,b):
        i,j = by_id[a],by_id[b]
        ga,gb = grams(texts[i],3),grams(texts[j],3)
        return {'a':a,'b':b, 'exact':normalize(texts[i])==normalize(texts[j]),
                'jaccard':jaccard(ga,gb), 'query_containment':containment(ga,gb),
                'cosine':sum(x*y for x,y in zip(vectors[i],vectors[j]))}
    pairs = [compare(a,b) for a,b in itertools.combinations(train,2)]
    cross = [compare(a,b) for a in bench for b in train]
    predictions = {
        'exact': {frozenset((p['a'],p['b'])) for p in pairs if p['exact']},
        'jaccard_0.5': {frozenset((p['a'],p['b'])) for p in pairs if p['jaccard'] >= 0.5},
        'cosine_0.9': {frozenset((p['a'],p['b'])) for p in pairs if p['cosine'] >= 0.9},
    }
    # Audit first: only normalized exact matches are automatically quarantined.
    seen, keep, quarantine = {}, [], []
    for ident in train:
        value = normalize(texts[by_id[ident]])
        digest = hashlib.sha256(value.encode()).hexdigest()
        prior = seen.get(digest)
        if prior is not None and value == normalize(texts[by_id[prior]]):
            quarantine.append({'id':ident, 'representative':prior, 'reason':'normalized_exact'})
        else:
            seen[digest] = ident
            keep.append(ident)
    # Boundary assertions: exclude diagonal; short/empty text isn't a match;
    # directed containment detects insertion; near-similarity is non-transitive.
    assert len(pairs)==15 and len(cross)==18
    assert grams('a',3)==set() and jaccard(set(),set())==0
    assert containment(set(),set())==0
    assert [x['id'] for x in quarantine]==['t1'] and 't3' in keep
    assert compare('b0','t5')['query_containment']==1.0
    assert compare('b0','t5')['jaccard'] < 0.5
    angles = [math.radians(x) for x in [0,20,40]]
    chain = [[math.cos(a),math.sin(a)] for a in angles]
    edges = [(str(i),str(j)) for i,j in itertools.combinations(range(3),2)
             if sum(x*y for x,y in zip(chain[i],chain[j]))>=0.9]
    assert edges==[('0','1'),('1','2')]
    assert components(['0','1','2'],edges)==[['0','1','2']]
    if not args.model_dir:
        assert compare('t0','t2')['jaccard']==0
        assert compare('t0','t3')['cosine']>=0.9
    report = {'seed':SEED,'python':platform.python_version(), 'engine':engine,
              'model_sha256':model_files, 'embedding_shape':[len(vectors),len(vectors[0])],
              'source_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
              'fixtures_sha256':hashlib.sha256(json.dumps(DOCS).encode()).hexdigest(),
              'fixtures':DOCS, 'pair_definition':'equivalent full-document meaning; containment alone is negative',
              'gold_positive_pairs':sorted(sorted(p) for p in GOLD),
              'metrics':{k:metrics(v) for k,v in predictions.items()},
              'threshold_sweep':[{'cosine_threshold':t, **metrics({frozenset((p['a'],p['b'])) for p in pairs if p['cosine']>=t})} for t in [0.85,0.9,0.95,0.99]],
              'train_pairs':pairs,'cross_split_pairs':cross,'keep':keep,'quarantine':quarantine,
              'cross_split_lexical_candidates':[p for p in cross if p['query_containment']>=0.8],
              'chain_edges':edges,'checks':'passed',
              'limitations':'Controlled fixtures only. Similarity candidates do not establish historical training exposure or model memorization.'}
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
    print(json.dumps({k:report[k] for k in ['seed','python','engine','embedding_shape','metrics','keep','quarantine','checks']},indent=2))
    print('cross_split_lexical_candidates=',len(report['cross_split_lexical_candidates']))
    print('report=',args.out)

if __name__=='__main__':
    main()