import polars as pl
from itertools import combinations, product
import random
import numpy as np
import sys, io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
from dupehell import generate
def get_content_cols(df):
exclude = {"record_id", "master_id", "entity_type", "domain"}
return [c for c in df.columns if c.lower() not in exclude]
def similarity_score(va, vb):
if va is None and vb is None:
return 1.0
if va is None or vb is None:
return 0.0
try:
return 1.0 if va == vb else 0.0
except:
return 1.0 if str(va) == str(vb) else 0.0
print("=" * 70)
print("DEEP DIVE: kyc medium hard_neg similarity")
print("=" * 70)
result = generate('kyc', size=2000, seed=42, difficulty='medium', output_dir='hn_quality_output', output_format='ipc')
df = pl.read_ipc(result.dataset)
gt = pl.read_ipc(result.ground_truth)
content_cols = get_content_cols(df)
hn_rids = gt.filter(pl.col('match_type') == 'hard_neg')['record_id'].to_list()
hn_df = df.filter(pl.col('record_id').is_in(hn_rids))
mid_map = {}
for row in hn_df.iter_rows(named=True):
mid_map[row['record_id']] = row['master_id']
by_mid = {}
for rid, mid in mid_map.items():
by_mid.setdefault(mid, []).append(rid)
mids = list(by_mid.keys())
random.seed(42)
record_map = {}
for row in df.iter_rows(named=True):
record_map[row['record_id']] = row
print(f"\nHard neg records: {len(hn_rids)}, masters: {len(mids)}")
print(f"Content columns: {len(content_cols)}")
sample_pairs = []
for i in range(min(3, len(mids))):
for j in range(i+1, min(3, len(mids))):
for a in by_mid[mids[i]]:
for b in by_mid[mids[j]]:
sample_pairs.append((a, b))
print(f"\nSample hard_neg pairs and their per-field similarity:\n")
for rid1, rid2 in sample_pairs[:3]:
r1, r2 = record_map[rid1], record_map[rid2]
match_count = 0
total = len(content_cols)
print(f"Pair: {rid1} (master={r1['master_id']}) vs {rid2} (master={r2['master_id']})")
for col in content_cols:
v1, v2 = r1[col], r2[col]
sim = similarity_score(v1, v2)
if sim == 1.0:
match_count += 1
else:
print(f" DIFF: {col}: \"{v1}\" vs \"{v2}\"")
print(f" Overall: {match_count}/{total} fields match = {match_count/total:.4f}\n")
print("=" * 70)
print("DEEP DIVE: hell difficulty — exact_dup master counts")
print("=" * 70)
result_hell = generate('kyc', size=2000, seed=42, difficulty='hell', output_dir='hn_quality_output', output_format='ipc')
df_h = pl.read_ipc(result_hell.dataset)
gt_h = pl.read_ipc(result_hell.ground_truth)
exact_rids_h = set(gt_h.filter(pl.col('match_type') == 'exact_dup')['record_id'].to_list())
exact_df_h = df_h.filter(pl.col('record_id').is_in(exact_rids_h))
per_master = exact_df_h.group_by('master_id').agg(pl.len()).sort('len', descending=True)
print("\nExact_dup records per master_id (top 20):")
print(per_master.head(20))
print(f"\nMax records for any master: {per_master['len'].max()}")
print(f"Masters with 1 record: {per_master.filter(pl.col('len') == 1).height}")
print(f"Masters with 2+ records: {per_master.filter(pl.col('len') >= 2).height}")
print("\nMaster_id duplicates across ALL match_types in hell:")
dupe_masters = df_h.group_by('master_id').agg(pl.len(), pl.col('match_type').first()).filter(pl.col('len') >= 2)
print(f"Masters with 2+ records (all types): {dupe_masters.height}")
if dupe_masters.height > 0:
print(dupe_masters.head(10))
else:
print("NONE — every record has a unique master_id in hell difficulty")
print("\n" + "=" * 70)
print("DEEP DIVE: light difficulty — exact_dup per master")
print("=" * 70)
result_light = generate('kyc', size=2000, seed=42, difficulty='light', output_dir='hn_quality_output', output_format='ipc')
df_l = pl.read_ipc(result_light.dataset)
gt_l = pl.read_ipc(result_light.ground_truth)
exact_rids_l = set(gt_l.filter(pl.col('match_type') == 'exact_dup')['record_id'].to_list())
exact_df_l = df_l.filter(pl.col('record_id').is_in(exact_rids_l))
per_master_l = exact_df_l.group_by('master_id').agg(pl.len()).sort('len', descending=True)
print("\nLight: Exact_dup records per master_id:")
print(per_master_l.head(20))
print(f"Masters with 1 record: {per_master_l.filter(pl.col('len') == 1).height}")
print(f"Masters with 2+ records: {per_master_l.filter(pl.col('len') >= 2).height}")
print("\n" + "=" * 70)
print("Cross-domain: avg differing fields in hard_neg pairs")
print("=" * 70)
for domain in ['kyc', 'healthcare', 'fintech', 'publishing', 'ecommerce']:
r = generate(domain, size=2000, seed=42, difficulty='medium', output_dir='hn_quality_output', output_format='ipc')
ddf = pl.read_ipc(r.dataset)
ggt = pl.read_ipc(r.ground_truth)
ccols = get_content_cols(ddf)
hn_rids = set(ggt.filter(pl.col('match_type') == 'hard_neg')['record_id'].to_list())
hn_ddf = ddf.filter(pl.col('record_id').is_in(hn_rids))
rmap = {}
midmap = {}
for row in hn_ddf.iter_rows(named=True):
rmap[row['record_id']] = row
midmap[row['record_id']] = row['master_id']
by_mid = {}
for rid, mid in midmap.items():
by_mid.setdefault(mid, []).append(rid)
mids = list(by_mid.keys())
diffs_list = []
for i in range(min(10, len(mids))):
for j in range(i+1, min(10, len(mids))):
for a in by_mid[mids[i]][:2]:
for b in by_mid[mids[j]][:2]:
if a == b: continue
r1, r2 = rmap[a], rmap[b]
diffs = sum(1 for c in ccols if similarity_score(r1[c], r2[c]) == 0.0)
diffs_list.append(diffs)
avg_diffs = np.mean(diffs_list) if diffs_list else 0
total_cols = len(ccols)
print(f"{domain:>14}: avg {avg_diffs:.1f}/{total_cols} fields differ = {(1-avg_diffs/total_cols)*100:.1f}% similarity")