import json, os, re, tempfile, shutil
from datetime import datetime
from dupehell import *
import polars as pl
TOLERANCE = 0.10
DATE_PATTERNS = [
(r'^\d{4}-\d{2}-\d{2}$', '%Y-%m-%d'),
(r'^\d{2}-\d{2}-\d{4}$', '%d-%m-%Y'),
(r'^\d{2}/\d{2}/\d{4}$', '%m/%d/%Y'),
(r'^\d{4}/\d{2}/\d{2}$', '%Y/%m/%d'),
(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}', '%Y-%m-%dT%H:%M:%S'),
]
DOMAINS_OF_INTEREST = [
'pharma', 'biotech', 'food_beverage', 'hospitality', 'travel',
'fashion', 'retail', 'sports', 'gaming', 'social_media', 'cybersecurity',
]
NON_NEGATIVE_HINTS = {
'age', 'count', 'quantity', 'capacity', 'enrollment_count',
'years_of_experience', 'number_of_employees', 'years_in_business',
'credit_score', 'score', 'rating', 'stock', 'inventory',
'years', 'months',
}
def check_date(val):
if isinstance(val, (datetime,)):
return True, None
v = str(val)
for pat, fmt in DATE_PATTERNS:
if re.match(pat, v):
try:
datetime.strptime(v, fmt)
return True, None
except ValueError:
continue
return False, v
results = {}
for domain in DOMAINS_OF_INTEREST:
print(f"\n{'='*60}")
print(f"[{domain}]")
print(f"{'='*60}")
with open(os.path.join('schemas', f'{domain}.json')) as f:
schema = json.load(f)
schema_cols = {}
for entity in schema['entities']:
for col in entity['columns']:
schema_cols[col['name']] = {
'type': col.get('type', 'string'),
'null_rate': col.get('null_rate_default', 0),
'entity': entity['name'],
'pool_name': col.get('pool_name'),
'template': col.get('template'),
}
outdir = tempfile.mkdtemp(prefix=f'dq_{domain}_')
try:
result = generate(domain=domain, size=1000, seed=42, output_dir=outdir)
ds_path = result.dataset
gt_path = result.ground_truth
df = pl.read_parquet(ds_path)
gt = pl.read_parquet(gt_path)
n = len(df)
print(f" Records: {n}, GT: {len(gt)}")
except Exception as e:
print(f" GENERATION ERROR: {e}")
shutil.rmtree(outdir, ignore_errors=True)
results[domain] = {'error': str(e)}
continue
dr = {'null_rates': {'pass': True, 'failures': []},
'emails': {'pass': True, 'failures': []},
'phones': {'pass': True, 'failures': []},
'dates': {'pass': True, 'failures': []},
'numerics': {'pass': True, 'failures': []}}
for cname, cinfo in schema_cols.items():
if cname not in df.columns:
continue
exp = cinfo['null_rate']
act = df[cname].null_count() / n
delta = abs(act - exp)
if exp == 0:
ok = act <= 0.05
elif exp == 1.0:
ok = act >= 0.95
else:
ok = delta <= TOLERANCE
if not ok:
dr['null_rates']['pass'] = False
dr['null_rates']['failures'].append({
'column': cname, 'entity': cinfo['entity'],
'expected': exp, 'actual': round(act, 4), 'delta': round(delta, 4)})
print(f" NR-FAIL: {cinfo['entity']}.{cname} exp={exp} act={act:.4f} delta={delta:.4f}")
email_cols = [c for c in df.columns if 'email' in c.lower()]
for ec in email_cols:
non_null = df.filter(pl.col(ec).is_not_null())[ec]
total = len(non_null)
if total == 0:
continue
viol = 0
samples = []
for v in non_null:
v = str(v)
issues = []
if '@' not in v:
issues.append('no @')
elif v.count('@') != 1:
issues.append(f'{v.count("@")} @ signs')
else:
local, dom = v.split('@', 1)
if '.' not in dom:
issues.append('no dot in domain')
if ' ' in v:
issues.append('contains spaces')
if issues:
viol += 1
if len(samples) < 5:
samples.append({'val': v, 'issues': issues})
if viol:
dr['emails']['pass'] = False
dr['emails']['failures'].append({
'column': ec, 'total': total, 'violations': viol,
'pct': round(viol/total*100, 2), 'samples': samples})
print(f" EMAIL-FAIL: {ec} — {viol}/{total} ({viol/total*100:.1f}%)")
phone_cols = [c for c in df.columns if any(k in c.lower() for k in ['phone', 'telephone', 'contact_number'])]
for pc in phone_cols:
non_null = df.filter(pl.col(pc).is_not_null())[pc]
total = len(non_null)
if total == 0:
continue
viol = 0
samples = []
for v in non_null:
v = str(v)
issues = []
allowed = set('0123456789 +-()')
bad = [c for c in v if c not in allowed]
if bad:
issues.append(f'bad chars: {bad[:5]}')
digits = sum(1 for c in v if c.isdigit())
if digits < 8:
issues.append(f'only {digits} digits')
if issues:
viol += 1
if len(samples) < 5:
samples.append({'val': v, 'issues': issues})
if viol:
dr['phones']['pass'] = False
dr['phones']['failures'].append({
'column': pc, 'total': total, 'violations': viol,
'pct': round(viol/total*100, 2), 'samples': samples})
print(f" PHONE-FAIL: {pc} — {viol}/{total} ({viol/total*100:.1f}%)")
date_cols = list(set(
[c for c in df.columns if any(k in c.lower() for k in ['date', 'birth', 'dob'])] +
[c for c, ci in schema_cols.items() if ci['type'] in ('date', 'datetime') and c in df.columns]
))
for dc in date_cols:
non_null = df.filter(pl.col(dc).is_not_null())[dc]
total = len(non_null)
if total == 0:
continue
viol = 0
samples = []
for v in non_null:
ok, err = check_date(v)
if not ok:
viol += 1
if len(samples) < 5:
samples.append(err)
if viol:
dr['dates']['pass'] = False
dr['dates']['failures'].append({
'column': dc, 'total': total, 'violations': viol,
'pct': round(viol/total*100, 2), 'samples': samples})
print(f" DATE-FAIL: {dc} — {viol}/{total} ({viol/total*100:.1f}%)")
for cname, cinfo in schema_cols.items():
if cname not in df.columns:
continue
if cinfo['type'] not in ('int', 'float', 'price', 'amount'):
continue
nn = df.filter(pl.col(cname).is_not_null())[cname]
if len(nn) == 0:
continue
negs = (nn < 0).sum()
if negs > 0 and any(nk in cname.lower() for nk in NON_NEGATIVE_HINTS):
dr['numerics']['pass'] = False
samps = nn.filter(nn < 0).head(5).to_list()
dr['numerics']['failures'].append({
'column': cname, 'entity': cinfo['entity'],
'ctype': cinfo['type'], 'neg_count': int(negs),
'total': len(nn), 'samples': samps})
print(f" NUM-FAIL: {cinfo['entity']}.{cname} — {negs}/{len(nn)} negative")
results[domain] = dr
shutil.rmtree(outdir, ignore_errors=True)
print("\n\n" + "="*100)
print("DATA QUALITY AUDIT PT.2 — SUMMARY")
print("="*100)
header = f"{'Domain':<20} {'NullRt':>8} {'Emails':>8} {'Phones':>8} {'Dates':>8} {'Numerics':>8} {'Overall':>8}"
print(header)
print("-"*100)
ok_count = 0
fail_count = 0
for domain in DOMAINS_OF_INTEREST:
r = results.get(domain, {})
if 'error' in r:
print(f"{domain:<20} {'ERR':>8} {'ERR':>8} {'ERR':>8} {'ERR':>8} {'ERR':>8} {'ERR':>8}")
fail_count += 1
continue
nr = 'PASS' if r['null_rates']['pass'] else 'FAIL'
em = 'PASS' if r['emails']['pass'] else 'FAIL'
ph = 'PASS' if r['phones']['pass'] else 'FAIL'
dt = 'PASS' if r['dates']['pass'] else 'FAIL'
nm = 'PASS' if r['numerics']['pass'] else 'FAIL'
allp = r['null_rates']['pass'] and r['emails']['pass'] and r['phones']['pass'] and r['dates']['pass'] and r['numerics']['pass']
ov = 'PASS' if allp else 'FAIL'
if allp:
ok_count += 1
else:
fail_count += 1
print(f"{domain:<20} {nr:>8} {em:>8} {ph:>8} {dt:>8} {nm:>8} {ov:>8}")
print("-"*100)
print(f"Total: {ok_count} PASS / {ok_count+fail_count} total ({fail_count} FAIL)")
print()
print("="*100)
print("DETAILED FAILURES")
print("="*100)
for domain in DOMAINS_OF_INTEREST:
r = results.get(domain, {})
if 'error' in r:
print(f"\n## {domain}: ERROR — {r['error']}")
continue
has_fail = not (r['null_rates']['pass'] and r['emails']['pass'] and r['phones']['pass'] and r['dates']['pass'] and r['numerics']['pass'])
if not has_fail:
continue
print(f"\n## {domain}")
for section in ['null_rates', 'emails', 'phones', 'dates', 'numerics']:
sd = r[section]
if sd['pass']:
continue
for fail in sd['failures']:
col = fail.get('column', '?')
ent = fail.get('entity', '?')
if section == 'null_rates':
print(f" [NR] {ent}.{col}: exp={fail['expected']} act={fail['actual']} delta={fail['delta']}")
elif section == 'emails':
print(f" [EM] {col}: {fail['violations']}/{fail['total']} ({fail['pct']}%)")
for s in fail.get('samples', []):
print(f" val='{s['val']}' issues={s['issues']}")
elif section == 'phones':
print(f" [PH] {col}: {fail['violations']}/{fail['total']} ({fail['pct']}%)")
for s in fail.get('samples', []):
print(f" val='{s['val']}' issues={s['issues']}")
elif section == 'dates':
print(f" [DT] {col}: {fail['violations']}/{fail['total']} ({fail['pct']}%)")
for s in fail.get('samples', []):
print(f" val='{s}'")
elif section == 'numerics':
print(f" [NM] {ent}.{col}: {fail['neg_count']}/{fail['total']} negative, samples={fail['samples']}")
print("\nDONE")